成本控管 - Kerwin AI Information

📌 本文重點

單一 Agent PoC 聰明，上線即暴露治理問題

需要 Agent Harness 統一管控工具、成本與政策

先建可觀測、可回放的 Harness，再談多 Agent 擴展

將治理與業務邏輯解耦，避免生產環境變實驗場

企業在玩 PoC 時，單一 Agent 看起來很聰明；一上線就暴露本質問題：工具亂叫、成本失控、記憶混亂、錯誤難重現、政策無法落地。這不是模型不夠強，而是缺少一層專門管控 Agent 行為的 Agent Harness（Agent 的 runtime & control plane）。

💡 關鍵： 問題不在模型本身，而在缺少專門負責治理、成本與行為管控的中介層。

這一層做的事很務實：集中管控工具註冊與版本 / 權限、step-level tracing + replay、記憶與知識更新、錯誤恢復與重試、成本與 Token Budget、政策注入（policy engine）、標準化的 human-in-the-loop 介面。有了它，你才真的能把多 Agent 系統放進生產環境，而不是「高級 demo」。

重點說明：Agent Harness 的 7 大能力

1. 工具 / 函式註冊與版本管理

把所有可被 Agent 呼叫的 API/工具收斂到一個 工具註冊中心，統一管理：

白名單 + Scope：不同 Agent 只能看到被授權的工具，避免「直接讓模型自由呼叫所有內部 API」。
版本管理：工具有 version 與 deprecation 概念，支援灰度切換與回滾。
結構化 schema：對應到 LLM 的 tool schema / function calling，並強制要求輸入輸出型別。

2. 觀察、Tracing 與 Replay

沒有 step-level tracing，任何「為什麼產生這個請款請求？」都變成佛系排查。Harness 要提供：

每一步 prompt / tool call / response 的結構化 log。
replay 能力：拿同一套 input/context/tool version 在測試環境重放。
與 APM/日誌系統整合（如 OpenTelemetry, Datadog）。

💡 關鍵： 有了 step-level tracing 與 replay，才能在出錯時精準重現並修復 Agent 行為，而不是盲目排查。

3. 記憶與知識源更新

多數系統踩的坑是「把 RAG/記憶全塞進 context」，結果是：

token 爆炸 → 成本高、延遲高
資訊新鮮度難管控

Harness 要把記憶拆成：

短期工作記憶（task-local state）：存在 run/session 中。
長期記憶（user / account profile, conversation history）：向量庫或 KV 存儲，按策略查再放入 context。
權威知識源（DB / data warehouse / API）：由工具抽象，不直接塞原始資料進 prompt。

4. 錯誤恢復與重試策略

Agent 在實際環境裡會遇到：工具 5xx、timeout、schema mismatch、LLM 回傳壞 JSON 等。Harness 要集中：

可配置重試策略（per tool 的 max_attempts、backoff）。
降級策略：改用備援工具、或中止並要求人工介入。
把錯誤與重試記錄在 tracing 中，方便事後分析。

5. 成本與 Token Budget 控制

程式化 Agent（CI、排程、webhook 觸發）最容易「安靜地燒錢」。Harness 應實作：

per-run token budget：單一任務總 token / cost 上限。
per-tool cost 上限：避免某個向量搜尋或報表 API 被無限 loop 呼叫。
依租戶 / team 維度聚合成本，餵到帳務系統或告警系統。

💡 關鍵： 透過 per-run 與 per-tool 的 token 與成本上限，可以在不影響功能的前提下，防止自動化流程悄悄造成巨額開銷。

6. 策略 / 合規政策注入（Policy Engine）

只在 prompt 放幾條「不要外流 PII」是不夠的。企業需要一個 policy engine：

針對工具呼叫、輸入輸出內容，做 策略判斷與拒絕。
支援條件式規則，如「財務資料工具只能在工作時間、由 Finance-Agent 使用」。
配合零信任閘道（如 SolonGate 類產品）做額外的身分驗證與審計。

7. Human-in-the-loop 標準化接口

有些動作永遠不該全自動（退款、合約變更…）。Harness 要提供：

標準化的 approval 任務結構：包含 who, what, diff, risks。
挂在 ticket 系統 / 內部前端（如模仿 LangGraph 的 human node）。
Agent 在收到人類決策後能繼續流程，而不是整個 run 重來。

實作範例：最小可用 Agent Harness（Python）

以下是簡化版的「Agent 執行器」，示範：

工具註冊 + 白名單
logging + tracing id
超時 / 重試
per-run token budget
per-tool 次數上限

import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional

# === 1. 工具註冊中心 ===

@dataclass
class ToolConfig:
    name: str
    func: Callable[[Dict[str, Any]], Any]
    version: str = "v1"
    max_calls_per_run: int = 5
    timeout_sec: float = 10.0
    cost_per_call: float = 0.001  # 自訂邏輯用
    enabled: bool = True


class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, ToolConfig] = {}

    def register(self, tool: ToolConfig):
        key = f"{tool.name}:{tool.version}"
        self._tools[key] = tool

    def get_allowed_tools(self, whitelist: List[str]) -> Dict[str, ToolConfig]:
        # whitelist 用的是 name，不帶 version
        out = {}
        for key, tool in self._tools.items():
            if tool.name in whitelist and tool.enabled:
                out[key] = tool
        return out


# === 2. Harness 執行設定 ===

@dataclass
class RunBudget:
    max_tokens: int
    max_cost: float
    used_tokens: int = 0
    used_cost: float = 0.0

    def charge_tokens(self, tokens: int):
        self.used_tokens += tokens
        if self.used_tokens > self.max_tokens:
            raise RuntimeError("Run token budget exceeded")

    def charge_cost(self, cost: float):
        self.used_cost += cost
        if self.used_cost > self.max_cost:
            raise RuntimeError("Run cost budget exceeded")


@dataclass
class AgentRunContext:
    run_id: str
    user_id: str
    allowed_tools: Dict[str, ToolConfig]
    budget: RunBudget
    tool_call_counts: Dict[str, int] = field(default_factory=dict)


# === 3. LLM 客戶端包一層（示意） ===

class LLMClient:
    def __init__(self, model: str):
        self.model = model

    def chat(self, messages: List[Dict[str, str]], tools_schema: List[Dict]) -> Dict[str, Any]:
        """
        回傳格式假設：
        {
          "content": "...",
          "tool_calls": [
            {"tool": "search:v1", "arguments": {"q": "foo"}}
          ],
          "usage": {"input_tokens": 200, "output_tokens": 150}
        }
        """
        # 這裡應呼叫實際 LLM API；為示意略過
        return {
            "content": "stub",
            "tool_calls": [],
            "usage": {"input_tokens": 50, "output_tokens": 30},
        }


# === 4. Harness 核心執行器 ===

class AgentHarness:
    def __init__(self, llm: LLMClient, tool_registry: ToolRegistry, logger):
        self.llm = llm
        self.tool_registry = tool_registry
        self.logger = logger

    def run(self, *, user_id: str, messages: List[Dict[str, str]],
            tool_whitelist: List[str], max_steps: int = 8,
            max_tokens: int = 8000, max_cost: float = 0.5) -> Dict[str, Any]:

        run_id = str(uuid.uuid4())
        allowed_tools = self.tool_registry.get_allowed_tools(tool_whitelist)
        ctx = AgentRunContext(
            run_id=run_id,
            user_id=user_id,
            allowed_tools=allowed_tools,
            budget=RunBudget(max_tokens=max_tokens, max_cost=max_cost),
        )

        self.logger.info(f"run_start", extra={"run_id": run_id, "user_id": user_id})

        for step in range(max_steps):
            step_id = step + 1
            self.logger.info("llm_step_start", extra={"run_id": run_id, "step": step_id})

            tools_schema = [
                {"name": t.name, "version": t.version} for t in allowed_tools.values()
            ]
            resp = self.llm.chat(messages, tools_schema)

            usage = resp.get("usage", {})
            ctx.budget.charge_tokens(usage.get("input_tokens", 0) + usage.get("output_tokens", 0))

            tool_calls = resp.get("tool_calls", [])
            if not tool_calls:
                # 任務完成
                self.logger.info("run_complete", extra={"run_id": run_id, "step": step_id})
                return {"run_id": run_id, "final": resp["content"]}

            # 執行工具呼叫
            for call in tool_calls:
                tool_key = self._resolve_tool_key(call["tool"], allowed_tools)
                tool_cfg = allowed_tools[tool_key]

                count = ctx.tool_call_counts.get(tool_key, 0) + 1
                if count > tool_cfg.max_calls_per_run:
                    raise RuntimeError(f"tool {tool_key} call limit exceeded")
                ctx.tool_call_counts[tool_key] = count

                result = self._call_tool_with_retry(tool_cfg, call["arguments"], ctx)
                messages.append({"role": "tool", "name": tool_cfg.name, "content": str(result)})

        raise RuntimeError("max_steps_exceeded")

    def _resolve_tool_key(self, tool_name: str, allowed_tools: Dict[str, ToolConfig]) -> str:
        # 簡化：同名只允許一個版本
        matches = [k for k, t in allowed_tools.items() if t.name == tool_name]
        if not matches:
            raise RuntimeError(f"tool {tool_name} not allowed")
        return matches[0]

    def _call_tool_with_retry(self, tool_cfg: ToolConfig, args: Dict[str, Any], ctx: AgentRunContext):
        attempts = 0
        while True:
            attempts += 1
            start = time.time()
            try:
                # 超時控制示意：可用 asyncio.wait_for 或 thread + join
                result = tool_cfg.func(args)
                elapsed = time.time() - start
                self.logger.info(
                    "tool_call",
                    extra={
                        "run_id": ctx.run_id,
                        "tool": tool_cfg.name,
                        "version": tool_cfg.version,
                        "attempts": attempts,
                        "elapsed_sec": elapsed,
                    },
                )
                ctx.budget.charge_cost(tool_cfg.cost_per_call)
                return result
            except Exception as e:
                if attempts >= 3:
                    self.logger.error("tool_failed", extra={"tool": tool_cfg.name, "err": str(e)})
                    raise
                time.sleep(0.5 * attempts)

你可以在這層再接上：

policy engine：在 _call_tool_with_retry 前後做輸入輸出檢查。
human-in-the-loop：當特定工具被呼叫，改成發出審核任務，而不是直接執行。
tracing 平台：把 run_id、step、tool 當成 span/trace id，上報到 OpenTelemetry。

建議與注意事項

1. 工具權限與多 Agent 協作

不同 Agent（或不同租戶）應有獨立的 tool whitelist 和輸入遮罩。
多 Agent 團隊協作時，要明確：誰能看哪些記憶 / 哪些知識源，避免「助理 Agent 無意間看到 CEO 的對話紀錄」。

2. 記憶與 RAG 的分層設計

不要「一次把所有 RAG 結果塞進 context」，應做：
第一階段檢索：向量庫 / keyword 檢索
第二階段過濾 / re-rank：只把 top-k、與當前任務強相關的放入 prompt
對長期記憶，引入 TTL / 熱度分級，過舊資料只保存在冷存儲，需要時再查。

3. Tracing / Replay 一開始就要做

踩坑：先上線，出事再補 tracing → 你根本不知道 Agent 做了什麼。
最少要記：run_id、step、messages(摘要即可)、tool_calls、錯誤堆疊、tool version。
為 replay 保留「當時的工具版本與政策版本」，不然重放行為會不一致。

4. 成本 / Budget 不只是一個大數字

按 run / user / team / workflow 類型 切分配額，結合告警（例如超過預估平均三倍就發訊息）。
對「自動觸發型」Agent（CI、webhook）尤其要設 per-run token budget + max_steps，避免 prompt/工具錯誤導致無限 loop。

5. 安全與政策落地要有「硬限制」

除了 prompt 提醒，還要有：
policy engine：直接阻擋不符規則的工具呼叫 / 輸出內容。
零信任閘道（如 SolonGate 類）在真正的企業 API 前再加一層，做身分、範圍、頻率控制。

總結： 不要把「Agent 邏輯」和「治理、成本、安全控制」寫死在同一份業務程式碼裡。先抽出一層可組態、可觀測、可回放的 Agent Harness，你才能放心地擴到多 Agent、跨團隊、跨租戶，而不把整個公司變成昂貴的實驗場。

🚀 你現在可以做的事

審視現有 Agent PoC，列出目前缺少的 tracing、budget、policy 能力，畫出一層簡易 Harness 設計草圖

在現有程式碼中加入 run_id、step-level log 與最小可用的 replay 機制，先讓行為可觀測

實作一個簡單的 ToolRegistry 與 per-run token budget，從一個 Agent 開始逐步遷移到 Harness 架構

標籤: 成本控管

企業級 Agent Harness 七大能力實戰拆解