標籤: Grafana Tempo

  • Eval-Driven Agents:讓 LLM 代理變成可維運系統

    Eval-Driven Agents:讓 LLM 代理變成可維運系統

    📌 本文重點

    • Evals 是 Agent 的 CI/CD Gate,讓更新可量化
    • 四個支柱讓 Agent 從黑盒變成可維運系統
    • 一週內可導入最小可行的 eval pipeline

    LLM Agent 最大的痛點很直接:一改 prompt 或策略,整體效果「好像」有變好,但沒人說得出到底好多少、哪裡變差、能不能安全上線。 Eval-Driven Agents 的核心,就是把 evals 當成 Agent 的 CI/CD,讓代理不再是黑盒魔法,而是可以版本管理、回溯、觀察與持續優化的軟體系統。


    重點說明

    1. Evals 是 Agent 的 CI/CD Gate

    把 Agent 當成服務在維護,就不能只靠「體感」判斷更新好壞。核心做法是:

    • 為每種任務設計 評估集(eval set) 與 指標(metrics)
    • 每次改 prompt / 工具 / 策略 / 模型,都先在 replay data 上跑一輪 eval
    • 只有達到門檻(類似單元測試全部通過)才允許 rollout

    💡 關鍵: 先建立穩定比較機制,再來追求模型效果,才能在特定場景量化「+8% 成功率」這種改善。

    關鍵不是追求完美模型,而是建立 穩定的比較機制,讓你敢說:這次改動在「退款流程」場景成功率 +8%,且「錯誤升級」場景沒退步。


    2. 四個支柱:讓 Agent 變得可維運

    圍繞「evals 是 CI/CD」這個核心,一個 production-grade Agent 至少要有四個支柱:

    1. 評估集與指標設計:為非確定性輸出定義 pass/fail 與容錯區間
    2. 任務切成多個「用例」:例如 FAQ 回答、工單分類、報表生成
    3. 每個用例定義:成功條件、允許誤差、關鍵失敗模式
    4. 指標不只看單一 aggregate score,而要區分場景與錯誤類型

    5. 資料與結構化回饋:用 Pydantic / schema 把工具調用、記憶與決策結構化

    6. 每次 Agent 執行都產出可重放的 trace:輸入、工具調用、模型回應、最終決策
    7. 讓 eval runner 可以重播舊版本 vs 新版本,逐步比較

    8. 觀察性與追蹤:整合 OpenTelemetry + Grafana Tempo 做分散式追蹤

    9. 將整條 agent workflow 變成 trace:每一次 tool call、一段 prompt 改動都看得到
    10. 問問題變得具體:「為什麼這次多調用了三個工具?」、「哪一段 prompt 改壞了成功率?」

    11. Eval-loop 與版本管理:把 eval 接進部署管線

    12. 類似單元測試 gate:每次改 prompt / 策略 / 模型先跑 eval
    13. 區分 實驗 eval(探索新策略)與 回歸 eval(保護既有能力)

    3. 這對你的專案有什麼實際好處?

    如果你現在有一個已上線但很脆弱的 Agent:

    • 降低改壞風險:每次調 prompt 不再是賭運氣,而是有數據護欄
    • 讓 debug 有方向:看到哪個場景、哪個工具路徑在退化,而不是「整體感覺怪怪的」
    • 更好控成本:配合 trace,知道哪裡 token 浪費最多(工具過度調用、不必要長上下文)
    • 團隊協作更順:PM/ML/Backend 看同一套 eval 報表,不再靠各自的 demo 體驗說服對方

    實作範例

    以下用簡化版 Python + Pydantic + OpenTelemetry,示範如何把一個脆弱 Agent,在一週內升級到有 eval pipeline 的狀態。


    1. 用 Pydantic 結構化 Agent Trace

    先定義 agent 的執行結果與步驟:

    from pydantic import BaseModel, Field
    from typing import List, Literal, Optional
    
    class ToolCall(BaseModel):
        name: str
        input: dict
        output: dict
        latency_ms: float
    
    class AgentStep(BaseModel):
        step_type: Literal["plan", "tool", "reflect", "final"]
        prompt: str
        response: str
        tool_call: Optional[ToolCall] = None
    
    class AgentTrace(BaseModel):
        trace_id: str
        user_input: str
        steps: List[AgentStep]
        final_output: str
        metadata: dict = Field(default_factory=dict)
    

    好處:

    • 每次執行都可重播:你可以拿 user_input + steps,在新版本模型上重跑,生成新的 trace
    • 易於比較舊版 vs 新版:對齊同一個 trace_id,逐步看哪個 step 不同

    在現有 Agent 中,只需要在 orchestrator 把每次決策與工具調用寫進這個 AgentTrace,就有一份可用作 eval 的資料。


    2. 定義 Eval 任務與指標:非確定性也能 Pass/Fail

    假設你有一個客服 Agent,需要回答退款相關問題,我們定義一個簡單的 eval:

    from pydantic import BaseModel
    
    class RefundEvalCase(BaseModel):
        case_id: str
        user_input: str
        expected_keywords: list[str]  # 例如 ["退款條件", "處理時間"]
        must_not_include: list[str]   # 例如 ["保證立即退款"]
    
    class EvalResult(BaseModel):
        case_id: str
        passed: bool
        score: float
        missing_keywords: list[str]
        bad_phrases: list[str]
    

    簡單的 eval runner:

    def run_refund_eval(agent_fn, cases: list[RefundEvalCase]) -> list[EvalResult]:
        results = []
        for case in cases:
            output = agent_fn(case.user_input)
            lower_output = output.lower()
    
            missing = [k for k in case.expected_keywords if k.lower() not in lower_output]
            bad = [p for p in case.must_not_include if p.lower() in lower_output]
    
            # 線性打分:關鍵字命中率 - 違規懲罰
            keyword_score = 1 - len(missing) / max(len(case.expected_keywords), 1)
            penalty = 0.3 * len(bad)
            final_score = max(0.0, keyword_score - penalty)
    
            results.append(EvalResult(
                case_id=case.case_id,
                passed=(final_score >= 0.8),  # **容錯區間**:>= 0.8 視為可接受
                score=final_score,
                missing_keywords=missing,
                bad_phrases=bad,
            ))
        return results
    

    💡 關鍵: 將非確定性輸出轉成「>= 0.8 即通過」的標準,讓 CI 能用 pass/fail 自動守門。

    這裡重點不是打分公式有多精緻,而是:

    • 先把「成功條件」具體化:有哪些必講的資訊、哪些不能亂承諾
    • 為每個 eval case 保留 error breakdown:缺失關鍵字 vs 違規措辭
    • 讓 CI 上只看 passed 率,而錯誤細節回報給開發者/PM 作 prompt 調整依據

    3. 用 OpenTelemetry + Grafana Tempo 做 Agent Trace

    把每次 Agent 流程變成可視化 trace,方便查為何多調用了三個工具、是哪一段 prompt 變壞。

    簡化版埋點(以 OpenTelemetry Python 為例):

    from opentelemetry import trace
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    
    # 初始化
    provider = TracerProvider()
    exporter = OTLPSpanExporter(endpoint="http://tempo:4318/v1/traces")
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    tracer = trace.get_tracer(__name__)
    
    # 在 Agent orchestrator 中
    
    def run_agent(user_input: str) -> AgentTrace:
        with tracer.start_as_current_span("agent_run") as span:
            span.set_attribute("agent.user_input", user_input)
    
            trace_model = AgentTrace(trace_id="...", user_input=user_input, steps=[])
    
            # Step 1: plan
            with tracer.start_as_current_span("plan_step") as s:
                plan_prompt = make_plan_prompt(user_input)
                plan_resp = call_llm(plan_prompt)
                s.set_attribute("llm.tokens", plan_resp.usage.total_tokens)
    
            # Step 2: tool call example
            with tracer.start_as_current_span("tool_call:search_order") as s:
                tool_input = {"order_id": "123"}
                tool_output = search_order(tool_input)
                s.set_attribute("tool.latency_ms", 42.0)
    
            # ...其餘步驟
    
            return trace_model
    

    好處:

    • 在 Grafana Tempo 裡可以完整看到 agent_run 的 timeline
    • 搭配 token 計費(參考 The Economics of Agents 那篇),可以計出 每個步驟的成本與貢獻
    • 當成功率下降時,你能具體問:是 plan_step 的 tokens 被砍太多,還是 tool_call latency 飆高導致超時?

    4. 把 eval 接進部署管線(CI/CD Gate)

    最後把 eval 變成 CI 裡的一個 stage:

    # .github/workflows/agent-eval.yml
    name: agent-eval
    
    on:
      pull_request:
        paths:
          - "agents/**"
          - "prompts/**"
    
    jobs:
      run-evals:
        runs-on: ubuntu-latest
        steps:
          - uses: actions/checkout@v4
    
          - name: Set up Python
            uses: actions/setup-python@v5
            with:
              python-version: "3.11"
    
          - name: Install deps
            run: pip install -r requirements.txt
    
          - name: Run regression evals
            run: python evals/run_regression.py
    
          - name: Check thresholds
            run: python evals/check_thresholds.py
    

    check_thresholds.py 裡只做幾件事:

    import sys
    from evals.results import load_results
    
    REGRESSION_MIN_PASS_RATE = 0.9  # **回歸 eval 門檻**
    
    if __name__ == "__main__":
        results = load_results("artifacts/regression.json")
        pass_rate = results.pass_rate()
    
        if pass_rate < REGRESSION_MIN_PASS_RATE:
            print(f"Regression eval failed: pass_rate={pass_rate}")
            sys.exit(1)  # CI 失敗,禁止合併
        else:
            print(f"Regression eval passed: pass_rate={pass_rate}")
    

    在實務上你會分兩組 eval:

    • 回歸 eval:保護現有功能,不允許退化
    • 實驗 eval:探索新功能、新策略,不當成 blocking gate,但會留報表做決策

    💡 關鍵: 將 REGRESSION_MIN_PASS_RATE 設為 0.9 這類門檻,讓 CI 自動阻擋明顯退化的改版。


    建議與注意事項

    1. 一週內最小成本導入 eval pipeline 的路線圖

    如果你現在有一個「已上線但很脆弱」的 Agent,可以這樣在一週內切入:

    Day 1-2:蒐集與結構化 trace

    • 在現有 Agent 增加最薄的 AgentTrace 結構(如上 Pydantic 範例)
    • 將最近 1-2 週的真實請求轉成 trace,存到資料庫或物件儲存(S3 / SeaweedFS)

    Day 3-4:定義 1-2 個核心場景的 eval

    • 選擇對業務影響最大的 1-2 個流程(例如退款、升級、帳單說明)
    • 用簡單規則+少量人工標註,做出第一版 pass/fail 判準與 score
    • 不求完美,只要能在版本之間穩定比較就夠

    Day 5-7:接進 CI,建立第一個 gate

    • 在 PR pipeline 裡加入 run_regression.py + check_thresholds.py
    • 門檻先設寬一點,只要不要明顯退化就放行
    • 讓團隊開始習慣:改 prompt / 策略前先跑 eval,之後再慢慢收緊標準

    2. 常見坑與避雷建議

    1. 過度依賴人工標註

    2. 你不需要為每個回應都人工打分,容易拖垮迭代速度

    3. 建議:少量標註 + 規則/模型輔助,人工只介入灰色地帶

    4. 用單一 aggregate 分數掩蓋失敗模式

    5. 整體 score 變高不代表沒有嚴重退化,例如:一般 FAQ 變強,但退款場景大幅下降

    6. 建議:按 場景 / 任務類別分開看指標,並保留 error breakdown

    7. 沒有分離「實驗 eval」與「回歸 eval」

    8. 把所有 eval 都當 blocking gate,會讓團隊不敢做大幅創新

    9. 建議:

      • 回歸 eval:保護既有能力,作為 CI gate
      • 實驗 eval:以 dashboard & report 呈現,不直接 block 部署
    10. 只評估最終輸出,忽略中間決策與工具路徑

    11. 這會讓你無法回答「為什麼這次多調用了三個工具?」

    12. 建議:在 trace 中至少記:每個 step 的 prompt、回應、工具調用與耗時,並用 OpenTelemetry 對應到 visualize 的 trace

    13. 沒把成本納入 eval 指標

    14. Agent 問題常常不是性能,而是「單位經濟」:同樣成功率但 token 成本翻倍

    15. 建議:在 eval 報告中加上 每個場景的平均 token / latency / tool calls 次數,作為優化目標之一

    核心結論:

    把 evals 當成 Agent 的 CI/CD,不是為了追求「更高的神奇效果」,而是讓你敢在生產環境持續迭代,而不會每次改動都賭運氣。從今天開始,先把你的 Agent 當成一個 可維運的軟體系統:有 trace、有 eval、有 gate,其他的魔法再說。這樣你才能在之後的模型升級、工具新增、記憶架構重構時,保持速度又不失控。

    🚀 你現在可以做的事

    • 在現有 Agent 專案中加入 AgentTrace 結構,開始紀錄並保存每次執行的 trace
    • 為一個關鍵業務場景(如退款流程)撰寫首批 5–10 個 RefundEvalCase 並實作 run_refund_eval
    • 在 CI(如 GitHub Actions)新增 agent-eval workflow,實作 check_thresholds.py 並先將 REGRESSION_MIN_PASS_RATE 設為 0.9