📌 本文重點
- 單一 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 budgetper-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-runtoken budget，從一個 Agent 開始逐步遷移到 Harness 架構
