结构化输出封装

对话里模型回的是自然语言，但程序要的是数据：拿到「帮我订 3 张明天去上海的高铁票」这句话，程序需要的是 {"city": "上海", "tickets": 3, "date": "明天"} 这样的 JSON。本章做一个 extract_json()：提示词约束模型输出 JSON → json.loads 解析 → 按 schema 校验 → 不合格就把错误回喂给模型重试。全程不依赖各家私有参数，任何 OpenAI 兼容服务都能跑。

1. 思路：提示词 + 校验兜底

部分服务商支持 response_format={"type": "json_object"} 或 json_schema 来强制 JSON 输出，写法以各家官方文档为准。本教程刻意不依赖它，改用「提示词要求 + 程序校验 + 失败带修正重发」的兜底方案，保证可移植；服务商支持时只需给 chat() 透传该参数（见第 5 节），无需改解析逻辑。

2. 完整代码 extract.py

"""extract.py —— 结构化输出封装：提示词 + JSON 解析 + schema 校验 + 失败重试。"""
import json
import re
from typing import Any, Dict, List, Optional

from llm_client import LLMClient

_PROMPT = (
    "请严格输出一个 JSON 对象作为最终回复，不要输出任何解释、前后缀或 markdown 代码块。\n"
    "输出必须符合下面的 JSON Schema：\n{json_schema}\n"
)


def _loads_json(text: str) -> Any:
    """从回复文本中提取 JSON（容忍代码块围栏与多余前后缀）。"""
    match = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)  # 去掉 ```json 围栏
    if match:
        text = match.group(1)
    text = text.strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        start, end = text.find("{"), text.rfind("}")   # 兜底：截取首尾大括号之间
        if start != -1 and end > start:
            return json.loads(text[start:end + 1])
        raise


def _check(value: Any, schema: Dict[str, Any], path: str = "根") -> List[str]:
    """极简 JSON Schema 校验：支持 object/array/string/integer/number/boolean。"""
    errors: List[str] = []
    typ = schema.get("type")

    if typ == "object":
        if not isinstance(value, dict):
            return [f"{path}: 应为对象(object)，实际是 {type(value).__name__}"]
        for key, sub in schema.get("properties", {}).items():
            if key in value:
                errors += _check(value[key], sub, f"{path}.{key}")
        for key in schema.get("required", []):
            if key not in value:
                errors.append(f"{path}: 缺少必填字段 {key}")
        return errors

    if typ == "array":
        if not isinstance(value, list):
            return [f"{path}: 应为数组(array)"]
        for i, item in enumerate(value):
            errors += _check(item, schema.get("items", {"type": "string"}), f"{path}[{i}]")
        return errors

    if typ == "integer":
        ok = isinstance(value, int) and not isinstance(value, bool)  # bool 是 int 子类
    elif typ == "number":
        ok = isinstance(value, (int, float)) and not isinstance(value, bool)
    elif typ == "string":
        ok = isinstance(value, str)
    elif typ == "boolean":
        ok = isinstance(value, bool)
    else:
        ok = True
    if not ok:
        errors.append(f"{path}: 应为 {typ}，实际是 {type(value).__name__}")
    return errors


def extract_json(llm: LLMClient, user_text: str,
                 schema: Optional[Dict[str, Any]] = None,
                 max_retries: int = 2) -> Any:
    """让模型输出 JSON 并按 schema 校验；失败时把错误回喂模型重试。

    返回解析并校验通过的对象（dict/list）；重试耗尽仍失败则抛 ValueError。
    """
    if schema:
        schema_text = json.dumps(schema, ensure_ascii=False, indent=2)
    else:
        schema_text = "（自由 JSON，但最外层必须是对象）"
    messages: List[dict] = [
        {"role": "system", "content": _PROMPT.format(json_schema=schema_text)},
        {"role": "user", "content": user_text},
    ]

    for attempt in range(max_retries + 1):
        reply = llm.chat_text(messages)
        print(f"[第 {attempt + 1} 次回复] {reply[:150]}")
        try:
            data = _loads_json(reply)
            errors = _check(data, schema) if schema else []
            if not errors:
                return data
            feedback = "；".join(errors[:5])
        except json.JSONDecodeError as e:
            feedback = f"回复不是合法 JSON：{e}。只输出一个 JSON 对象"

        # 把失败原因作为新的一轮 user 消息回喂，让模型“看到错误再改”
        messages.append({"role": "assistant", "content": reply})
        messages.append({"role": "user",
                         "content": f"校验未通过：{feedback}。请修正后重新只输出 JSON。"})
    raise ValueError("重试多次仍无法得到符合 schema 的 JSON，请检查 schema 或提示词")


if __name__ == "__main__":
    # 演示：把一句自然语言抽取成结构化字段
    schema = {
        "type": "object",
        "properties": {
            "name": {"type": "string", "description": "姓名"},
            "age": {"type": "integer", "description": "年龄"},
            "city": {"type": "string", "description": "所在城市"},
            "hobbies": {"type": "array", "items": {"type": "string"},
                        "description": "爱好列表"},
        },
        "required": ["name", "age", "city", "hobbies"],
    }
    text = "我叫李昂，今年 26 岁，住在杭州，平时喜欢编程、摄影和骑行。"
    data = extract_json(llm=LLMClient(), user_text=text, schema=schema)
    print("解析结果：", data)
    print(f"姓名：{data['name']}，年龄：{data['age']}，城市：{data['city']}")

3. 运行与输出

python extract.py
# 输出示例（随模型与时间变化）：
# [第 1 次回复] {"name": "李昂", "age": 26, "city": "杭州", "hobbies": ["编程", "摄影", "骑行"]}
# 解析结果： {'name': '李昂', 'age': 26, 'city': '杭州', 'hobbies': ['编程', '摄影', '骑行']}
# 姓名：李昂，年龄：26，城市：杭州

如果模型某次回复带了多余文字、字段类型写错（如年龄写成字符串 "26"）或漏字段，校验器会给出具体错误并把「修正要求」发给模型再试一次，最多 max_retries + 1 次。

4. 校验器设计要点

• schema 采用 JSON Schema 的最小子集：type 有 object/array/string/integer/number/boolean，object 看 properties + required，array 看 items。够用且易懂，想支持 enum/嵌套约束可自行扩展 _check。
• _loads_json 先剥掉 ```json 围栏，再截首尾大括号兜底——模型偶尔会「好心」加上解释或代码块，这两招能救回大半。
• 校验错误用中文拼成一句 feedback 回喂，模型读得懂才能改得对；限制 errors[:5] 防止错误太多刷爆上下文。
• 失败重试走的是普通 user 消息，不是改写 system——保留原始诉求，只追加修正意见。

5. 可选：服务商原生 JSON 模式

服务商支持时，可在调用处给 chat() 透传（extract_json 内部未用，读者可自行接入）：

resp = llm.chat(messages, response_format={"type": "json_object"})  # 部分服务商

注意：json_object 模式通常要求提示词里出现 “json” 字样，json_schema 模式则要给出完整 schema，具体字段与限制以各家官方文档为准。本章的提示词+校验方案在任何服务上行为一致，建议作为默认。 小结：extract_json() = 提示词要求 JSON + json.loads 解析 + schema 校验 + 错误回喂重试，是智能体把自然语言翻译成程序可执行数据的标准姿势；第 5 章的工具参数解析正是它的直接应用。