# 別只叫 Agent「不要說謊」：為 LangGraph 加上確定性 API 與 LLM-as-Judge 雙重防線

我們的 AIOps agent 有個很具體的問題：它會捏造 trace ID。

當 on-call 工程師問「payment service 有沒有 error trace？」，agent 有時會信心滿滿地回答「是的，見 trace `a1b2c3d4...`」——但這串 ID 根本不存在 Tempo 裡。工程師點進去，404。壞的不只是使用者體驗，而是這讓整個 RCA 結論失去可信度。

另一個問題是 k8s 操作。我們的 agent 可以提議執行 `rollout_undo` 或 `scale`。這些操作有真實影響：把 replica 縮到 0 可以讓服務整個掛掉。我們有 blast-radius gate、circuit breaker，但這些 gate 都是規則型的，覆蓋不到「action 本身語意上合不合理」的問題。

這兩個問題看似南轅北轍，但指向了同一個工程現實：光靠 Prompt 壓不死，我們必須在答案生成之後，做`「事後驗證」（Post-Generation Verification）`。

驗證分兩條路：一種是`確定性的事實查核（去資料庫對答案）`，另一種是`語意上的邊界審查`。而後者，正是 `LLM-as-a-Judge` 的核心戰場。

* * *

## 教科書解法：LangChain 的 RubricMiddleware

在講我們的實作前，先看 LangChain 官方近期怎麼抽象這個題目。這套 Middleware 雖然我們最終沒直接用，但它的設計提供了完美的參考指引： LangChain 的 [RubricMiddleware](https://docs.langchain.com/oss/python/deepagents/rubric) 把這個 pattern 包成 middleware，接在 `create_deep_agent()` 上：

```python
from langchain_deepagents import RubricMiddleware, create_deep_agent

middleware = RubricMiddleware(
    model="openai:gpt-4o-mini",   # grader 用的模型，可以比主 agent 便宜
    tools=[run_test_suite],        # grader 可以呼叫的工具（選用）
    max_iterations=3,              # 最多重跑幾次（上限 20）
    on_evaluation=log_verdict,     # 每次審查後的 callback（選用）
)
agent = create_deep_agent(..., middleware=[middleware])

# rubric 在呼叫時傳入，不用寫死在 agent 裡
result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "幫我寫一個 find_duplicates 函數"}]},
    rubric="""
    - 函數必須通過所有 unit test
    - 時間複雜度必須是 O(n)
    - 不能用任何第三方套件
    """,
    config={"configurable": {"thread_id": "thread-001"}},
)
```

rubric 是換行分隔的 checklist。沒有傳 rubric，middleware 就不會啟動。`thread_id` + checkpointer 讓 rubric 可以跨多次 invocation 持續，agent 可以在中斷後繼續跑。

每次 agent 產出答案，grader 就審一次，回傳五種 verdict 之一：

```plaintext
satisfied              → 全部通過，終止
needs_revision         → 注入 per-criterion feedback，agent 重跑
max_iterations_reached → 達到 max_iterations，強制終止
failed                 → rubric 本身無法評估（寫法有問題）
grader_error           → grader sub-agent 拋了 exception
```

`needs_revision` 是最重要的路徑：grader 不只說「不通過」，它會針對每個 criterion 給出具體的 gap（哪裡不對、缺什麼），然後把這份 feedback 注入回 agent 的 message history，讓 agent 下一輪有明確方向修正，而不是盲目重試。

`on_evaluation` callback 每次審查後都會觸發，拿到的是一個 `RubricEvaluation` dict：

```python
def log_verdict(evaluation: RubricEvaluation):
    print(f"iteration={evaluation['iteration']}")
    print(f"result={evaluation['result']}")       # verdict 字串
    print(f"explanation={evaluation['explanation']}")
    for c in evaluation['criteria']:
        # 每個 criterion 的個別判斷
        print(f"  [{c['passed']}] {c['criterion']}: {c['gap']}")
```

用 `stream_events(..., version="v3")` 的話，可以從 event stream 裡抓到 `rubric_evaluation_start` 和 `rubric_evaluation_end` 兩個自訂 event，適合需要即時顯示審查進度的場景。

**grader 帶工具**是這個設計最關鍵的地方。文件裡的例子是讓 grader 拿到一個 `run_test_suite` 工具，實際執行測試套件，而不是靠 LLM 自己猜程式碼對不對：

```python
@tool
def run_test_suite(code: str) -> str:
    """執行測試，回傳 pass/fail 結果"""
    # 實際跑 pytest 或 unittest
    ...

middleware = RubricMiddleware(
    model="openai:gpt-4o-mini",
    tools=[run_test_suite],  # grader 會主動呼叫這個來驗證
)
```

這把 grader 從「LLM 憑感覺打分」變成「帶工具的 sub-agent 驗證」。差別很大：前者受 LLM 幻覺影響，後者的結果是確定性的。

```markdown
flowchart TD
    U([使用者提問]) --> A[Agent 執行工具、生成答案]
    A --> G{Grader 審查}
    G -->|呼叫工具驗證| T[測試/API/DB 查詢]
    T --> G
    G -->|satisfied| R([回傳答案])
    G -->|needs_revision\nper-criterion feedback| FB[feedback 注入 history]
    FB --> A
    G -->|max_iterations_reached| R
    G -->|grader_error| R

    style G fill:#f5a623,color:#000
    style FB fill:#d0e8ff,color:#000
    style T fill:#e8f5e9,color:#000
```

這個設計的吸引力在於把「品質標準」從 prompt 裡抽出來，變成一個獨立可測試的元件。rubric 可以是字串，可以有工具，可以隨場景換，也可以在 runtime 動態決定要審什麼。

* * *

## 我們沒用 Middleware，但用了同樣的思路

我們的 agent 是 LangGraph + Gemini，不是 LangChain deep agent，沒辦法直接插 `RubricMiddleware`。但核心思路一樣：**在 agent 產出之後，用獨立的 LLM 或程式驗證，不通過就打回去重試**。

我們把這個邏輯包在 `rubric.py` 裡，實作了兩個 guard。

### Guard 1：Trace ID 驗證

思路很直接：用 regex 把答案裡所有 32-hex 的字串找出來，逐一去 Tempo HTTP API 確認存不存在。

```python
_TRACE_ID_RE = re.compile(r"\b([0-9a-f]{32})\b", re.IGNORECASE)

async def _tempo_trace_exists(trace_id: str) -> bool:
    url = f"{settings.tempo_url}/api/traces/{trace_id}"
    async with httpx.AsyncClient(timeout=3.0) as client:
        resp = await client.get(url)
    if resp.status_code == 404:
        return False
    return bool(resp.json().get("batches", []))

async def verify_trace_ids(answer: str) -> tuple[bool, str]:
    ids = {m.group(1).lower() for m in _TRACE_ID_RE.finditer(answer)}
    missing = [tid for tid in ids if not await _tempo_trace_exists(tid)]
    if not missing:
        return True, ""
    return False, (
        f"The trace IDs {missing} you cited do not exist in Tempo. "
        "Call `query_tempo_traces` again to find real traces."
    )
```

這個 guard 本身不用 LLM，直接查資料庫，3 秒 timeout，網路出錯就 pass-through（不能讓 infra 問題阻擋正常回答）。

「為什麼不直接在 prompt 裡叫 agent 不要捏造？」我們試過了。Prompt 有效果，但失敗率不是零。當工具回傳空結果，agent 有時會「補全」一個看起來合理的 trace ID。這種行為靠 prompt 壓不死，要靠事後驗證。

### Guard 2：K8s Write 預飛審查

這個 guard 才真正用到 LLM-as-judge。我們需要判斷「這個 action 在當前 context 下合不合理」，這是語意層面的問題，規則寫不完。

```python
class _K8sRubricVerdict(BaseModel):
    safe_to_proceed: bool
    reason: str

_K8S_RUBRIC_SYSTEM = """你是 Kubernetes 操作的安全審查員。

BLOCK（safe_to_proceed=false）的條件：
- deployment name 有萬用字元或異常通用（"all"、"*"）
- replicas = 0（服務整個掛掉）
- rollout_undo 但 context 顯示問題不是 bad deploy（例如 DB 過載、infra 故障）
- scale 倍數超過當前的 10 倍

其他情況 ALLOW。blast-radius 和 circuit breaker 已在前面跑過，只擋語意上明顯有問題的操作。"""

async def check_k8s_write(action: str, args: dict, context: str = "") -> tuple[bool, str]:
    try:
        llm = _k8s_rubric_llm()  # structured output → _K8sRubricVerdict
        verdict = await llm.ainvoke([
            SystemMessage(content=_K8S_RUBRIC_SYSTEM),
            HumanMessage(content=f"Action: {action}\nArgs: {args}\nContext: {context}")
        ])
        return verdict.safe_to_proceed, verdict.reason
    except Exception as e:
        return True, f"rubric check skipped ({e})"
```

Structured output（Pydantic model）確保 grader 一定回傳 `bool` 而非讓我們自己 parse 文字。例外情況一律放行——grader 出問題不能成為阻擋正當操作的理由。

* * *

## 接線到 LangGraph

兩個 guard 接的位置不一樣，這很重要。

**Trace ID guard** 在答案生成後接，兩條路都有：

```python
# headless path（alert webhook）—— run_headless() 裡
trace_ok, retry_prompt = await verify_trace_ids(answer)
if not trace_ok:
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": retry_prompt}],
         "tool_calls_used": 0,
         "budget": max(2, settings.webhook_tool_call_budget // 2)},
        config=config,
    )

# chat path —— astream_events 結束後
trace_ok, _ = await verify_trace_ids("".join(answer_parts))
if not trace_ok:
    correction = "\n\n> **[注意]** 部分 trace ID 無法驗證，請重新查詢。"
    answer_parts.append(correction)
    yield {"type": "token", "text": correction}
```

headless path 給 agent 補一個 retry turn（budget 砍半避免無限追加費用）；chat path 則直接在答案末尾附上警告（因為使用者在等，重跑太慢）。

**K8s write guard** 在 `execution.py` 的 gate 序列裡插進 gate 3b：

```mermaid
flowchart LR
    A[ActionRequest approved] --> G1[Gate 1\n前置條件重驗]
    G1 -->|fail| ABORT1([ABORTED])
    G1 -->|pass| G2[Gate 2\nBlast-Radius]
    G2 -->|over policy| ABORT2([ABORTED])
    G2 -->|pass| G3[Gate 3\nCircuit Breaker]
    G3 -->|open| ABORT3([ABORTED])
    G3 -->|pass| G3B[Gate 3b\nRubric LLM]
    G3B -->|blocked| ABORT4([ABORTED])
    G3B -->|pass| G4[Gate 4\n實際執行]
    G4 --> DONE([SUCCEEDED / FAILED])

    style G3B fill:#f5a623,color:#000
    style ABORT4 fill:#ffcccc,color:#000
```

```python
# execution.py —— gate 3b
rubric_ok, rubric_reason = await check_k8s_write(req.action, req.args, context)
if not rubric_ok and settings.actions_enabled:  # kill switch 關著時不擋
    ar_store_transition(req.request_id, Status.EXECUTING, Status.ABORTED,
                        outcome=f"rubric blocked: {rubric_reason}")
    return {"status": "aborted", "outcome": f"rubric blocked: {rubric_reason}"}
```

`settings.actions_enabled` 是全局 kill switch。當它是 `False`（預設），整個 execute 路徑都不會真正跑，所以 gate 3b 也沒必要阻擋——這讓既有測試（期待走到 `refused`）不受影響。

* * *

## 測試策略

16 個 unit test，分三層：

**底層：HTTP 行為**

用 `respx` mock Tempo API，測試四種情境：200 有 batches、200 空 batches、404、網路錯誤（均應回傳 `True` 避免誤擋）。

**中層：guard 邏輯**

Mock `_tempo_trace_exists`，不跑真實 HTTP。測試重點：

*   重複 ID 只查一次（dedup）
    
*   混合情境（一真一假）整體判為 fail
    
*   LLM exception → pass-through
    

**上層：gate 條件**

不啟動完整 execution pipeline，直接驗布林邏輯：

```python
assert not (not rubric_ok and ex.settings.actions_enabled)
# rubric 說 False，actions_enabled=False → gate 不觸發
```

這讓測試不依賴 k8s cluster 或真實 LLM 呼叫，跑起來快。

* * *

## Tradeoff 與我們沒做的事

**Token 成本**

每次答案生成後多一次 LLM 呼叫（k8s guard），或一次 HTTP 查詢（trace guard）。我們的 headless path 本來就有 confidence loop（低信心時重跑），rubric retry 是在這之外再疊一層。最壞情況：一次 alert 觸發 → 3 個 confidence loop × 1 rubric retry = 6 次 agent run。

這是可接受的，因為 k8s 操作本來就該謹慎，而 trace ID 查詢只是 HTTP GET，成本可忽略。

**Chat path 的限制**

Chat path 的 trace guard 只警告，不重跑。因為使用者在即時等待，重跑可能需要 30-60 秒，體驗比直接顯示警告更差。這個取捨不是技術問題，是 UX 決策。

**Grader 本身的幻覺**

K8s write guard 用 LLM 來 judge，但 LLM 本身也可能判錯。我們的設計是「寧可漏判，不要誤擋」：exception 一律放行，block 條件設得相對明確（不模糊）。真正的安全線是 blast-radius policy 和 circuit breaker（純規則型），rubric 是額外的語意層，不是唯一防線。

**RubricMiddleware vs 自己接**

如果你用的是 LangChain deep agent，`RubricMiddleware` 封裝得很完整，不用自己處理 retry loop 和 verdict 狀態。我們選擇自己接是因為架構是 LangGraph，而且我們需要對不同路徑（headless vs chat）有不同行為。Middleware 的抽象在我們的場景下反而是限制。

* * *

# 結語：從 Prompt Engineering 走向 System Engineering

實作完這兩層 Guard 後最大的體會是：構建可靠的 AI Agent，重心早已不在於「寫出一個讓它絕不犯錯的完美 Prompt」，而是承認它一定會犯錯，並圍繞著它建立一套低成本的糾錯與制衡系統。

確定性的事實交給 API，模糊的語意交給另一個 LLM，剩下的，交給架構設計。這大概就是 Agent 進入 Production 前，工程師最該做的事。

程式碼在 [`rubric.py`](https://github.com/tedmax100/o11y-bench/blob/feat/aiops_agent/aiops-agent/service/app/rubric.py)。