《AI Agent 开发入门》6.3 结构化输出约束与解析兜底

# minagent/parse.py
from __future__ import annotations
import json
import re
from typing import Any

_FENCE_RE = re.compile(r"```(?:json|JSON)?\s*(.*?)```", re.DOTALL)


class ParseError(ValueError):
    """模型输出里找不到合法 JSON。"""


def strip_fence(text: str) -> str:
    """剥掉 Markdown 代码围栏；没有围栏时原样返回。"""
    m = _FENCE_RE.search(text)
    return m.group(1) if m else text

def _balanced(text: str, start: int) -> str | None:
    """从 start 处的 { 或 [ 起，扫到配平的右括号；字符串内的括号不计。"""
    opener = text[start]
    depth = 0
    in_str = False
    esc = False
    for i in range(start, len(text)):
        c = text[i]
        if in_str:
            if esc:
                esc = False
            elif c == "\\":
                esc = True
            elif c == '"':
                in_str = False
        elif c == '"':
            in_str = True
        elif c in "{[":
            depth += 1
        elif c in "}]":
            depth -= 1
            if depth == 0:
                return text[start:i + 1]
    return None

def parse_json(text: str) -> Any:
    """稳健解析：去围栏 → 直接 loads → 提取首个配平对象 → 抛 ParseError。"""
    cleaned = strip_fence(text).strip()
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        pass
    for i, c in enumerate(cleaned):
        if c in "{[":
            cand = _balanced(cleaned, i)
            if cand is None:
                continue
            try:
                return json.loads(cand)
            except json.JSONDecodeError:
                continue
    raise ParseError("未能在输出中找到合法 JSON")

# run63.py
import sys
sys.path.insert(0, "/tmp/ab6work")
from minagent.parse import parse_json, ParseError

CASES = [
    ("干净 JSON", '{"city": "北京", "temp": 22}', {"city": "北京", "temp": 22}),
    ("带 json 围栏", '```json\n{"ok": true}\n```', {"ok": True}),
    ("无语言围栏", '```\n[1, 2, 3]\n```', [1, 2, 3]),
    ("前置寒暄", '好的，这是结果：{"score": 8}，希望有帮助。', {"score": 8}),
    ("后置解释", '{"a": {"b": 1}} 这是嵌套对象。', {"a": {"b": 1}}),
    ("字符串里带括号", '{"note": "用 {花括号} 和 [方括号]"}', {"note": "用 {花括号} 和 [方括号]"}),
    ("前后都有噪声", '前言 {"k": "v"} 后记', {"k": "v"}),
]

passed = 0
for name, raw, expect in CASES:
    try:
        got = parse_json(raw)
        ok = got == expect
    except ParseError as e:
        got, ok = f"ParseError: {e}", False
    passed += ok
    print(f"[{'PASS' if ok else 'FAIL'}] {name:14} -> {got}")

# 真·非法输入必须抛错，而不是返回半个对象
for name, raw in [("残缺 JSON", '{"a": 1'), ("单引号键", "{'a': 1}"), ("纯文本", "我不确定。")]:
    try:
        got = parse_json(raw)
        print(f"[FAIL] {name:14} -> 意外返回 {got}")
    except ParseError:
        passed += 1
        print(f"[PASS] {name:14} -> 正确抛 ParseError")

print(f"\n{passed}/{len(CASES) + 3} 通过")

[PASS] 干净 JSON        -> {'city': '北京', 'temp': 22}
[PASS] 带 json 围栏      -> {'ok': True}
[PASS] 无语言围栏          -> [1, 2, 3]
[PASS] 前置寒暄           -> {'score': 8}
[PASS] 后置解释           -> {'a': {'b': 1}}
[PASS] 字符串里带括号        -> {'note': '用 {花括号} 和 [方括号]'}
[PASS] 前后都有噪声         -> {'k': 'v'}
[PASS] 残缺 JSON        -> 正确抛 ParseError
[PASS] 单引号键           -> 正确抛 ParseError
[PASS] 纯文本            -> 正确抛 ParseError

10/10 通过

# run63b.py
import sys
sys.path.insert(0, "/tmp/ab6work")
from minagent.types import Message, ModelResponse
from minagent.models import ScriptedModel
from minagent.parse import parse_json, ParseError

def repair(model, messages, tools=None):
    """解析失败时，把错误回灌给模型要求重出，最多一轮。"""
    resp = model.complete(messages, tools or [])
    try:
        return parse_json(resp.message.content)
    except ParseError as e:
        messages = messages + [
            resp.message,
            Message(role="user", content=f"你上次的输出无法解析为 JSON：{e}。请只输出合法 JSON。"),
        ]
        retry = model.complete(messages, tools or [])
        return parse_json(retry.message.content)

model = ScriptedModel([
    ModelResponse(message=Message(role="assistant", content="当然可以！结果是：{score: 8}")),  # 非法
    ModelResponse(message=Message(role="assistant", content='{"score": 8}')),                # 修正
])
msgs = [Message(role="system", content="只输出 JSON。"),
        Message(role="user", content="给这份周报打个分。")]
print("修复结果：", repair(model, msgs))
print("模型被调用了", len(model.calls), "次")
print("第二次请求的最后一条：", model.calls[-1][0][-1].content)

修复结果： {'score': 8}
模型被调用了 2 次
第二次请求的最后一条： 你上次的输出无法解析为 JSON：未能在输出中找到合法 JSON。请只输出合法 JSON。