Day 19 我們加入了第一版 retry。
當 Agent 的 JSON 輸出發生 format_error 時,runner 可以產生 retry task,讓 Agent 有一次修正機會。
目前流程大致是:
Agent answer
-> evaluator
-> if format_error: retry once
-> evaluator again
-> save eval result
這讓平台不只會發現錯誤,也開始具備 recovery path。
這一篇要往 Guardrails 推進。
Day 13 我們已經做過 JSON 驗證,但那時候它是 evaluator 的一部分。
換句話說,它的角色是:
測試結束後,判斷 Agent output 是否通過。
這次要把 schema validation 改成另一種角色:
Agent output 進入後續流程前,先檢查它是否符合 schema。
不符合就阻擋,並記錄 violation。
這就是 guardrail。
這一篇要做到的是:
建立一個最小 JSON Schema Guardrail,讓 JSON 任務的輸出必須符合欄位與型別要求,否則不能被視為可進入後續流程的結果。
會完成幾件事:
output_schema 概念。guardrails/json_schema.py。GuardrailResult。evals/runner.py,在 evaluator 前先套用 guardrail。guardrail_passed、guardrail_type、guardrail_reason。先不做:
範圍先收斂到一小塊:
針對 JSON output 任務,檢查輸出是否是合法 JSON object,且 required keys 和型別符合要求。
Day 13 的 json_exact evaluator 會檢查:
actual 是否等於 expected
例如:
{
"expected": {
"answer": 15
},
"actual": {
"answer": 15
}
}
這是 evaluation。
它回答的是:
這題有沒有答對?
Guardrail 的問題不太一樣。
它先不判斷答案值是否正確,而是判斷:
這個 output 是否安全、合法、可被後續程式處理?
例如我們要求 schema:
{
"type": "object",
"required": {
"answer": "number"
}
}
那下面這個 output 會通過 guardrail:
{
"answer": 16
}
即使 16 不是正確答案,它仍然是合法的結構。
接著 evaluator 再判斷:
answer 是否等於 15?
兩者分工可以這樣看:
| 元件 | 問題 | 範例 |
|---|---|---|
| Guardrail | output 能不能進入後續流程? | 是否為合法 JSON、是否有 answer、型別是否為 number |
| Evaluator | output 是否符合測試預期? | answer 是否等於 15 |
這個分工很值得先釐清。
因為在真實產品中,很多時候你不能等到最後才發現格式不合法。
如果後續流程需要讀:
parsed["answer"]
那 guardrail 就應該先保證 parsed 是 dict,而且 answer 存在。
這次會新增一個 guardrails/ package,並修改 runner 和部分 JSON test cases。
agent-testing-platform/
evals/
cases.json
runner.py
evaluators.py
guardrails/
__init__.py
json_schema.py
agents/
simple_agent.py
data/
eval_runs/
eval_run_*.json
會新增:
| 檔案 | 用途 |
|---|---|
guardrails/__init__.py |
讓 guardrails 成為 Python package |
guardrails/json_schema.py |
實作 JSON schema guardrail |
會修改:
| 檔案 | 修改內容 |
|---|---|
evals/cases.json |
替 JSON output cases 加入 output_schema |
evals/runner.py |
在 evaluator 前先執行 guardrail |
不修改:
agents/simple_agent.py
agents/prompts.py
evals/evaluators.py
因為這次不是改 Agent 的生成行為,也不是改最終評分規則。
這次是在 evaluation pipeline 中加入一個前置檢查點。
Day 10 的 JSON cases 原本大致長這樣。
修改 evals/cases.json 中的 case_013:
{
"id": "case_013",
"input": "請用 JSON 格式回傳 10 + 5 的答案,欄位名稱使用 answer",
"expected": {
"answer": 15
},
"grading_method": "json_exact",
"task_type": "json_output"
}
這次要加上 output_schema。
修改 evals/cases.json:
{
"id": "case_013",
"input": "請用 JSON 格式回傳 10 + 5 的答案,欄位名稱使用 answer",
"expected": {
"answer": 15
},
"output_schema": {
"type": "object",
"required": {
"answer": "number"
}
},
"grading_method": "json_exact",
"task_type": "json_output"
}
case_014 也一樣是 answer 欄位。
修改 evals/cases.json 中的 case_014:
{
"id": "case_014",
"input": "請用 JSON 格式回傳 8 * 7 的答案,欄位名稱使用 answer",
"expected": {
"answer": 56
},
"output_schema": {
"type": "object",
"required": {
"answer": "number"
}
},
"grading_method": "json_exact",
"task_type": "json_output"
}
case_015 則是 status 欄位。
修改 evals/cases.json 中的 case_015:
{
"id": "case_015",
"input": "請用 JSON 格式回傳狀態 success,欄位名稱使用 status",
"expected": {
"status": "success"
},
"output_schema": {
"type": "object",
"required": {
"status": "string"
}
},
"grading_method": "json_exact",
"task_type": "json_output"
}
這裡的 output_schema 不是完整 JSON Schema 規格。
它只是本系列自己的簡化格式:
{
"type": "object",
"required": {
"field_name": "field_type"
}
}
支援的型別先放四種:
| schema type | Python 對應 |
|---|---|
string |
str |
number |
int 或 float |
boolean |
bool |
object |
dict |
這樣已經足夠處理目前 JSON output cases。
新增 guardrails/__init__.py:
這個檔案可以先留空。
它的用途和前面的 agents/__init__.py、evals/__init__.py 一樣,是讓 Python 把 guardrails 視為 package。
之後 evals/runner.py 才能匯入:
from guardrails.json_schema import validate_json_schema
新增 guardrails/json_schema.py。
先建立 result model。
新增 guardrails/json_schema.py:
import json
from dataclasses import dataclass
from typing import Any
@dataclass
class GuardrailResult:
passed: bool
guardrail_type: str
failure_reason: str | None = None
parsed_output: dict[str, Any] | None = None
GuardrailResult 和 EvaluationResult 有點像,但用途不同。
| 欄位 | 說明 |
|---|---|
passed |
是否通過 guardrail |
guardrail_type |
這次固定是 json_schema |
failure_reason |
不通過時的原因 |
parsed_output |
通過時解析後的 JSON object |
parsed_output 不是必要欄位,但很實用。
因為通過 guardrail 後,後續流程就可以使用已解析過的 dict,不一定要再 json.loads() 一次。
接著做一個小 helper,用來檢查欄位型別。
繼續修改 guardrails/json_schema.py:
def matches_type(value: Any, expected_type: str) -> bool:
if expected_type == "string":
return isinstance(value, str)
if expected_type == "number":
return isinstance(value, (int, float)) and not isinstance(value, bool)
if expected_type == "boolean":
return isinstance(value, bool)
if expected_type == "object":
return isinstance(value, dict)
return False
這裡有個小細節:
isinstance(True, int)
在 Python 裡會是 True。
但 JSON schema 的 number 不應該把 boolean 當成 number。
所以 number 的判斷要加上:
not isinstance(value, bool)
這是小地方,但很值得在教學裡提醒。
接著實作主要 function。
繼續修改 guardrails/json_schema.py:
def validate_json_schema(
actual: str | None,
schema: dict[str, Any] | None,
) -> GuardrailResult:
if schema is None:
return GuardrailResult(
passed=True,
guardrail_type="json_schema",
)
if actual is None:
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason="Actual output is None",
)
try:
parsed = json.loads(actual)
except json.JSONDecodeError as exc:
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason=f"Output is not valid JSON: {exc.msg}",
)
if schema.get("type") != "object":
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason="Only object schema is supported",
)
if not isinstance(parsed, dict):
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason="JSON output must be an object",
)
required = schema.get("required", {})
for key, expected_type in required.items():
if key not in parsed:
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason=f"Missing required key: {key}",
)
if not matches_type(parsed[key], expected_type):
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason=(
f"Key '{key}' must be {expected_type}, "
f"but got {type(parsed[key]).__name__}"
),
)
return GuardrailResult(
passed=True,
guardrail_type="json_schema",
parsed_output=parsed,
)
這個 function 的流程如下:
沒有 schema
-> 直接通過
有 schema
-> actual 不能是 None
-> actual 必須是合法 JSON
-> schema type 目前只支援 object
-> parsed output 必須是 dict
-> required keys 必須存在
-> required keys 的型別必須正確
如果都通過,就回傳:
GuardrailResult(
passed=True,
guardrail_type="json_schema",
parsed_output=parsed,
)
如果不通過,就回傳明確的 failure_reason。
整理後,guardrails/json_schema.py 會像這樣:
import json
from dataclasses import dataclass
from typing import Any
@dataclass
class GuardrailResult:
passed: bool
guardrail_type: str
failure_reason: str | None = None
parsed_output: dict[str, Any] | None = None
def matches_type(value: Any, expected_type: str) -> bool:
if expected_type == "string":
return isinstance(value, str)
if expected_type == "number":
return isinstance(value, (int, float)) and not isinstance(value, bool)
if expected_type == "boolean":
return isinstance(value, bool)
if expected_type == "object":
return isinstance(value, dict)
return False
def validate_json_schema(
actual: str | None,
schema: dict[str, Any] | None,
) -> GuardrailResult:
if schema is None:
return GuardrailResult(
passed=True,
guardrail_type="json_schema",
)
if actual is None:
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason="Actual output is None",
)
try:
parsed = json.loads(actual)
except json.JSONDecodeError as exc:
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason=f"Output is not valid JSON: {exc.msg}",
)
if schema.get("type") != "object":
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason="Only object schema is supported",
)
if not isinstance(parsed, dict):
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason="JSON output must be an object",
)
required = schema.get("required", {})
for key, expected_type in required.items():
if key not in parsed:
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason=f"Missing required key: {key}",
)
if not matches_type(parsed[key], expected_type):
return GuardrailResult(
passed=False,
guardrail_type="json_schema",
failure_reason=(
f"Key '{key}' must be {expected_type}, "
f"but got {type(parsed[key]).__name__}"
),
)
return GuardrailResult(
passed=True,
guardrail_type="json_schema",
parsed_output=parsed,
)
這裡沒有使用第三方 JSON Schema library。
原因是本系列目前只需要最小可用版本。
等 schema 變複雜後,可以再換成:
jsonschema
但 MVP 階段先把 guardrail 概念做清楚比較重要。
接著修改 evals/runner.py。
先加入 import。
修改 evals/runner.py:
from guardrails.json_schema import validate_json_schema
接著在每一筆 case 執行完 Agent 後、呼叫 evaluator 前,加入 guardrail。
Day 19 後,runner 裡可能有這段:
result = agent.run(test_case["input"])
save_trace(result.trace)
evaluation = evaluate(test_case, result.answer)
這次改成:
result = agent.run(test_case["input"])
save_trace(result.trace)
guardrail = validate_json_schema(
actual=result.answer,
schema=test_case.get("output_schema"),
)
if guardrail.passed:
evaluation = evaluate(test_case, result.answer)
else:
evaluation = EvaluationResult(
passed=False,
failure_reason=guardrail.failure_reason,
failure_type="format_error",
)
這裡需要從 evals.evaluators 匯入 EvaluationResult。
修改 evals/runner.py:
from evals.evaluators import EvaluationResult, evaluate
這段邏輯的意思是:
如果 guardrail 通過
-> 繼續交給 evaluator 判斷答案是否正確
如果 guardrail 不通過
-> 不再進 evaluator
-> 直接視為 format_error
為什麼不通過時不進 evaluator?
因為 guardrail 的角色就是阻擋不合格 output。
如果輸出不是合法 JSON,後續流程就不應該假裝它可以繼續被處理。
Day 19 已經有 retry 流程。
加入 guardrail 後,retry 的判斷來源可以更清楚。
流程可以變成:
Agent 第一次回答
-> json schema guardrail
-> 如果 blocked,產生 format_error
-> 如果符合 retry 條件,retry once
-> retry 後再跑一次 guardrail
-> 通過 guardrail 後,再跑 evaluator
程式上可以整理成一個 helper,避免第一次與 retry 後重複寫太多邏輯。
修改 evals/runner.py,新增 evaluate_with_guardrail():
def evaluate_with_guardrail(
test_case: dict[str, Any],
actual: str | None,
) -> tuple[EvaluationResult, Any]:
guardrail = validate_json_schema(
actual=actual,
schema=test_case.get("output_schema"),
)
if not guardrail.passed:
return (
EvaluationResult(
passed=False,
failure_reason=guardrail.failure_reason,
failure_type="format_error",
),
guardrail,
)
return evaluate(test_case, actual), guardrail
這樣原本:
evaluation = evaluate(test_case, result.answer)
可以改成:
evaluation, guardrail = evaluate_with_guardrail(
test_case,
result.answer,
)
retry 後也用同一個 helper:
evaluation, guardrail = evaluate_with_guardrail(
test_case,
result.answer,
)
這樣可以確保第一次和 retry 後使用同一套 guardrail + evaluator 流程。
接著修改 result JSON。
在 append result 時加入:
"guardrail_passed": guardrail.passed,
"guardrail_type": guardrail.guardrail_type,
"guardrail_reason": guardrail.failure_reason,
修改 evals/runner.py 的 completed result:
results.append(
{
"case_id": test_case["id"],
"input": test_case["input"],
"expected": test_case["expected"],
"output_schema": test_case.get("output_schema"),
"grading_method": test_case["grading_method"],
"task_type": test_case["task_type"],
"status": "completed",
"actual": result.answer,
"passed": evaluation.passed,
"failure_type": evaluation.failure_type,
"failure_reason": evaluation.failure_reason,
"guardrail_passed": guardrail.passed,
"guardrail_type": guardrail.guardrail_type,
"guardrail_reason": guardrail.failure_reason,
"retry_enabled": retry_enabled,
"retry_count": retry_count,
"initial_actual": initial_actual,
"initial_failure_type": initial_failure_type,
"initial_failure_reason": initial_failure_reason,
"trace_session_id": result.trace.session_id,
"trace_session_ids": trace_session_ids,
"error": None,
}
)
這樣一筆 JSON output 失敗時,就能看出它是:
guardrail 擋下來
還是:
guardrail 通過,但 evaluator 判定答案值錯誤
例如格式錯誤:
{
"case_id": "case_013",
"guardrail_passed": false,
"guardrail_type": "json_schema",
"guardrail_reason": "Output is not valid JSON: Expecting value",
"passed": false,
"failure_type": "format_error"
}
例如格式正確但答案錯:
{
"case_id": "case_013",
"guardrail_passed": true,
"guardrail_type": "json_schema",
"guardrail_reason": null,
"actual": "{\"answer\": 16}",
"passed": false,
"failure_type": "format_error",
"failure_reason": "Expected key 'answer' to be '15', but got '16'"
}
這裡的 failure_type 仍然可能是 format_error,因為 Day 16 的分類規則目前把 json_exact 失敗都歸到格式類。
這不是最完美的分類。
但這次先不修改 failure type classifier。
Day 21 回顧時可以討論:json_exact 裡其實還可以細分成 schema error 和 value error。
如果 Agent 執行發生 exception,例如 API 429 或 API key 沒設定,guardrail 根本還沒機會執行。
為了讓 result schema 一致,except 區塊也補上欄位。
修改 evals/runner.py 的 except:
except Exception as exc:
results.append(
{
"case_id": test_case["id"],
"input": test_case["input"],
"expected": test_case["expected"],
"output_schema": test_case.get("output_schema"),
"grading_method": test_case["grading_method"],
"task_type": test_case["task_type"],
"status": "error",
"actual": None,
"passed": False,
"failure_type": "execution_error",
"failure_reason": "Agent execution error",
"guardrail_passed": None,
"guardrail_type": None,
"guardrail_reason": None,
"retry_enabled": retry_enabled,
"retry_count": 0,
"initial_actual": None,
"initial_failure_type": None,
"initial_failure_reason": None,
"trace_session_id": None,
"trace_session_ids": [],
"error": str(exc),
}
)
guardrail_passed 在這裡是 None,不是 False。
原因是:
False 表示 guardrail 執行了,但沒有通過。
None 表示 guardrail 根本沒有執行。
這對分析很有幫助。
如果你最近遇到 Gemini API 429,這些案例會是:
{
"status": "error",
"failure_type": "execution_error",
"guardrail_passed": null,
"error": "429 ..."
}
這表示問題在 API request 階段,還沒進到 output guardrail。
這點對目前專案很實用。
如果你的 Gemini API key 已經遇到 429,Day 20 不需要一直打 Gemini 才能驗證。
今天要測的是:
validate_json_schema() 是否正確阻擋不合格 JSON
runner 是否能記錄 guardrail metadata
這可以用本地輸出驗證。
例如先在 Python shell 測:
from guardrails.json_schema import validate_json_schema
schema = {
"type": "object",
"required": {
"answer": "number",
},
}
print(validate_json_schema('{"answer": 15}', schema))
print(validate_json_schema('```json\n{"answer": 15}\n```', schema))
print(validate_json_schema('{"answer": "15"}', schema))
print(validate_json_schema('{"result": 15}', schema))
預期結果是:
第一筆通過
第二筆失敗:不是合法 JSON
第三筆失敗:answer 型別不是 number
第四筆失敗:缺少 answer
這些測試完全不需要 Gemini API。
等 API quota 恢復後,再跑完整 eval。
如果想讓驗證更穩定,可以新增一個小檔案。
新增 tests/test_json_schema_guardrail.py:
from guardrails.json_schema import validate_json_schema
SCHEMA = {
"type": "object",
"required": {
"answer": "number",
},
}
def test_valid_json_passes() -> None:
result = validate_json_schema('{"answer": 15}', SCHEMA)
assert result.passed is True
assert result.parsed_output == {"answer": 15}
def test_markdown_code_fence_fails() -> None:
result = validate_json_schema('```json\n{"answer": 15}\n```', SCHEMA)
assert result.passed is False
assert result.failure_reason is not None
assert "not valid JSON" in result.failure_reason
def test_wrong_type_fails() -> None:
result = validate_json_schema('{"answer": "15"}', SCHEMA)
assert result.passed is False
assert result.failure_reason == "Key 'answer' must be number, but got str"
def test_missing_key_fails() -> None:
result = validate_json_schema('{"result": 15}', SCHEMA)
assert result.passed is False
assert result.failure_reason == "Missing required key: answer"
執行:
python3 -m pytest tests/test_json_schema_guardrail.py
如果你還沒有安裝 pytest,也可以先跳過這個檔案,用前面的 Python shell 手動驗證。
這裡的重點是:
Guardrail 的主要邏輯應該可以本地測,不應該每次都依賴 real LLM。
這也呼應最近遇到的 Gemini 429 問題。
LLM API 適合做真實 baseline,但平台邏輯最好能用本地測試穩定驗證。
完成後,eval result 會多幾個 guardrail 欄位。
例如通過 guardrail 且通過 evaluator:
{
"case_id": "case_013",
"actual": "{\"answer\": 15}",
"guardrail_passed": true,
"guardrail_type": "json_schema",
"guardrail_reason": null,
"passed": true,
"failure_type": null,
"failure_reason": null
}
被 guardrail 擋下:
{
"case_id": "case_013",
"actual": "```json\n{\"answer\": 15}\n```",
"guardrail_passed": false,
"guardrail_type": "json_schema",
"guardrail_reason": "Output is not valid JSON: Expecting value",
"passed": false,
"failure_type": "format_error",
"failure_reason": "Output is not valid JSON: Expecting value"
}
通過 guardrail 但 evaluator 失敗:
{
"case_id": "case_013",
"actual": "{\"answer\": 16}",
"guardrail_passed": true,
"guardrail_type": "json_schema",
"guardrail_reason": null,
"passed": false,
"failure_type": "format_error",
"failure_reason": "Expected key 'answer' to be '15', but got '16'"
}
這三種狀態的差異要看清楚。
它讓我們可以分清楚:
輸出格式不合法
輸出格式合法但答案錯
輸出格式合法且答案對
Day 19 做 retry。
Day 20 做 guardrail。
兩者可以互相配合。
Guardrail 負責偵測:
這個 output 是否符合 schema?
Retry 負責修正:
如果 output 不符合 schema,要不要給 Agent 一次重試?
所以比較完整的流程會是:
Agent output
-> schema guardrail
-> if blocked and retry enabled
-> retry once
-> schema guardrail again
-> evaluator
-> save result
這樣會比只靠 prompt 更可靠。
Prompt 是第一層:
請模型盡量輸出正確格式。
Guardrail 是第二層:
程式檢查輸出是否真的符合格式。
Retry 是第三層:
如果格式不符合,嘗試自動修正一次。
這三層加起來,才比較像可靠性工程,而不是只靠一句 prompt。
這次的 guardrail 仍然是 MVP。
這次的 schema 格式是我們自己定義的簡化版。
它還不支援:
但對目前的 JSON output cases 已經夠用。
目前 guardrail 放在 evals/runner.py。
這對教學最簡單,因為我們正在 eval pipeline 裡觀察結果。
但真實產品中,guardrail 可能會放在:
本系列先從 runner 開始,之後可以再抽成可重用的 pipeline。
json_exact 的 failure type 還可以更細目前 Day 16 的 classify_failure() 可能會把所有 json_exact 失敗都歸到:
format_error
但今天加入 guardrail 後,其實可以更細分:
| 情境 | 更精準的分類 |
|---|---|
| 不是合法 JSON | format_error |
| 缺 required key | schema_error |
| 欄位型別錯 | schema_error |
| schema 通過但值錯 | wrong_answer |
這次先不改分類,避免同一天改太多。
Day 21 回顧時會把這件事列成下一步改善方向。
這次把 JSON validation 從 evaluator 的一部分,往 guardrail 推進。
完成內容包含:
evals/cases.json 的 JSON cases 加入 output_schema。guardrails/__init__.py。guardrails/json_schema.py。GuardrailResult。validate_json_schema()。evals/runner.py,在 evaluator 前先套用 guardrail。guardrail_passed、guardrail_type、guardrail_reason。做完後,JSON output 的流程從:
Agent output
-> evaluator
變成:
Agent output
-> schema guardrail
-> evaluator
如果再接上 Day 19 retry,流程會更完整:
Agent output
-> schema guardrail
-> if blocked: retry once
-> schema guardrail again
-> evaluator
這表示平台開始從「測試結果」往「可靠性控制」前進。
Day 21 會做第三週回顧。
我們會把 Day 15 到 Day 20 串起來:
重點不是宣稱 Agent 已經可靠,而是整理目前平台已經能回答哪些問題。
例如:
真實 LLM baseline 和 fake baseline 差在哪?
最多的 failure type 是什麼?
prompt 改動有沒有改善?
retry 有沒有降低 format_error?
schema guardrail 擋下了哪些輸出?
哪些問題應該留到第四週處理?
Day 21 回顧完後,Day 22 會進入 Tool Guardrail。
到那時候,guardrail 不只會檢查輸出格式,也會開始檢查 Agent 使用工具的方式是否符合限制。