Agent 结构化输出 —— 让 LLM 返回可校验的 JSON，而非自由文本

一、问题：自由文本不可靠

前几篇聊的 Agent，一直有一个隐藏假设：LLM 输出的内容，是给人看的自然语言。但实际场景里，Agent 的输出很多时候需要被另一个程序消费——这就尴尬了。

场景 A: Agent 查询天气 → 下游系统需要 {city, temp, condition} 结构化数据→ 用正则从"北京今天晴天，25°C"中提取？做不到准确率 100%场景 B: Agent 执行数据分析 → 前端需要 JSON 渲染图表→ LLM 偶尔会在 JSON 前后加 "这是结果：" 之类的废话，解析失败场景 C: Agent 输出的文本中混入了 Markdown 或 HTML→ 下游解析器崩溃

核心矛盾其实就一个：LLM 最擅长的是自然语言，但程序需要的是结构化、可校验、可序列化的数据。如果直接用自由文本当输出，就等于把“解析”的责任甩给了调用方——每次都得搞 ad-hoc 的正则匹配和 try-catch，谁受得了？

二、方案：用一个特殊的 Tool 来输出

思路其实很直接——既然问题出在输出格式上，那把“输出”也变成一个工具不就行了？

flowchart TBsubgraph OLD[之前：自由文本输出]O1[LLM 生成文本] --> O2[return msg.content]O2 --> O3[调用方自己解析]O3 --> O4[正则 / try-catch / 猜测]endsubgraph NEW[之后：结构化输出]N1[LLM 调用 final_output 工具]N1 --> N2{参数符合 Schema?}N2 -->|是| N3[返回结构化 JSON]N2 -->|否| N4[json schema 校验失败]N3 --> N5[调用方直接 JSON.parse]end

不是让 LLM “以 JSON 格式回答”——那种靠 prompt 约束的方式太不可靠了。而是给它一个必须填参数的 tool，LLM 调用这个 tool 的行为本身就是结构化的：因为 tool 的 parameters 就是 JSON Schema，天然强制校验。

三、最简单的实现：final_output 工具

OUTPUT_TOOL_NAMES = {"final_output"}def make_final_output_tool() -> dict:return {"type": "function","function": {"name": "final_output","description": ("以结构化格式输出最终答案。调用此工具表示回答完成。""result 字段放结构化数据，summary 字段放给用户看的总结。"),"parameters": {"type": "object","properties": {"result": {"type": "object","description": "最终答案的 JSON 结构化数据",},"summary": {"type": "string","description": "给用户看的一句话总结",},},"required": ["result"],},},}def final_output_handler(result: dict, summary: str = "") -> str:sanitize_output(result)if summary:sanitize_string(summary)return json.dumps({"result": result, "summary": summary}, ensure_ascii=False)

final_output 本质上是一个信号工具——它不执行任何 I/O 操作，唯一的作用就是告诉 Agent 循环“回答完了，这是答案”。

主循环中的终止逻辑

for tc in tool_calls:name = tc["function"]["name"]args = json.loads(tc["function"]["arguments"])result = call_with_timeout(active_tool_map[name], kwargs=args, timeout=30)# ← 新增：final_output 是最终答案，直接返回if name in OUTPUT_TOOL_NAMES:return result# ← 短路，不再循环

之前判断“LLM 是否回答完了”靠的是 if tool_calls is None——这个信号其实不太可靠。用 final_output 工具作为显式的“我完成了”信号，比隐式的“没有 tool_call”要准确得多。

四、三层安全防线

单靠 final_output 的工具 Schema 约束还不够——LLM 仍然可能在 result 字段里填入恶意文本。Structure.py 提供了三层防护：

flowchart TDIN[LLM 调用 final_output] --> L1subgraph L1[第一层: Schema 安全校验]S1[校验 description 长度 < 200]S2[校验 properties 数量 < 50]S3[校验嵌套深度 < 5]S4[拒绝递归 schema]endL1 --> L2subgraph L2[第二层: 输出格式校验]V1[jsonschema.validate 严格模式]V2[禁止额外字段 additionalProperties: false]endL2 --> L3subgraph L3[第三层: 注入检测]I1[检测 Markdown 注入 - 代码块]I2[检测 HTML 注入 - script 标签]I3[检测 prompt 注入 - ignore 指令]I4[检测 SQL 注入 - DROP TABLE]endL3 --> OUT[安全的结构化 JSON 输出]

4.1 第一层：Schema 安全校验

防止恶意构造的 Schema 攻击校验器本身——描述太长可能是注入 payload，嵌套太深可能炸递归：

MAX_SCHEMA_DEPTH = 5MAX_DESCRIPTION_LEN = 200MAX_PROPERTIES = 50def validate_schema(schema: dict, depth: int = 0):if depth > MAX_SCHEMA_DEPTH:raise ValueError("Schema 嵌套过深")for key, prop in schema.get("properties", {}).items():if len(prop.get("description", "")) > MAX_DESCRIPTION_LEN:raise ValueError(f"description 过长: {key}")if prop.get("type") == "object":validate_schema(prop, depth + 1)

4.2 第二层：jsonschema 严格校验

LLM 的输出即使工具参数格式正确，也未必符合期望的 Schema。比如要求 temperature 是 number，LLM 却输出字符串 "25"：

def validate_output(data: dict, schema: dict) -> dict:jsonschema.validate(instance=data, schema=schema,format_checker=jsonschema.FormatChecker())return data

4.3 第三层：注入检测

递归遍历整个输出结构，对每个字符串值执行注入检测——不管在顶层、嵌套对象还是数组元素里：

INJECTION_PATTERNS = [(r"```", "Markdown 代码块"),(r"<script", "HTML script 标签"),(r"ignore.*previous.*instruction", "Prompt 注入"),(r"DROPs+TABLE", "SQL 注入"),]def sanitize_output(data):if isinstance(data, str):for pattern, label in INJECTION_PATTERNS:if re.search(pattern, data, re.IGNORECASE):raise ValueError(f"检测到可疑内容 ({label}): {data[:80]}...")elif isinstance(data, dict):for v in data.values(): sanitize_output(v)elif isinstance(data, list):for item in data: sanitize_output(item)return data
五、工厂函数：为不同场景定制输出 Schema
Structure.py 还提供了一个工厂函数，让不同场景注册各自的输出工具：
def make_final_output_tool(name, description, output_schema):validate_schema(output_schema)  # 第一层tool_def = {"type": "function","function": {"name": name, "description": description,"parameters": {"type": "object","properties": {"data": output_schema,"summary": {"type": "string", "description": "一句话总结"},},"required": ["data"],},},}def handler(data: dict, summary: str = "") -> str:validate_output(data, output_schema)  # 第二层sanitize_output(data)   # 第三层return json.dumps({"result": data, "summary": summary}, ensure_ascii=False)return tool_def, handler
使用示例——注册一个天气查询输出 Skill：
weather_tool, weather_handler = make_final_output_tool(name="output_weather",description="输出天气查询的结构化结果",output_schema={"type": "object","properties": {"city": {"type": "string"},"condition": {"type": "string"},"temperature": {"type": "number"},},"required": ["city", "condition", "temperature"],},)skills.register(Skill(name="weather-output",tools=[weather_tool],tool_map={"output_weather": weather_handler},system_prompt="查询天气后必须调用 output_weather，不要直接返回文本。",))
通过 Skill 注册，Agent 动态加载 weather-output 后，LLM 严格生成符合 Schema 的数据。
六、完整数据流
sequenceDiagramparticipant U as 用户participant RA as run_agent_with_traceparticipant LLM as LLMparticipant SM as SkillManagerparticipant ST as Structure 三层校验participant OUT as 下游系统U->>RA: "北京天气怎么样？"RA->>SM: get_active_tools()SM-->>RA: [search_web, final_output]RA->>LLM: messages + toolsLLM-->>RA: tool_call: search_web("北京")RA->>RA: 执行 search_webRA->>LLM: messages + 搜索结果LLM-->>RA: tool_call: final_output(result, summary)Note over RA,ST: 输出工具触发终止RA->>ST: sanitize_output(result)ST->>ST: 递归检测所有字符串值ST-->>RA: 安全的结果RA->>OUT: JSON 结构化数据OUT-->>U: 图表渲染 / 数据存储
七、设计精要
7.1 以 Tool 为输出载体，而非 Prompt 约束
Prompt "请以 JSON 格式回答" 是不可靠的。但当输出是一个 tool 的参数时，LLM 天然受 JSON Schema 约束——API 层面就要求参数是合法 JSON。
7.2 显式终止信号优于隐式判断
if name in OUTPUT_TOOL_NAMES 比 if tool_calls is None 更明确。前者是"Agent 主动说完了"，后者是"Agent 没话说了"。
7.3 安全是分层递进的
三层校验各司其职：Schema 校验防攻击者、jsonschema 校验防格式错误、注入检测防恶意内容。任何一层出问题都能精确定位。
7.4 零侵入集成
主循环只加了一行判断 + 一个提前返回，前序六层完全不受影响。
八、完整演进路径
Phase 1: 基础 Agent (~50 行)Phase 2: 可观测 + 压缩 + 持久化 (+~200 行)Phase 3: Skill 渐进加载(+~150 行)Phase 4: 多轮对话循环(+~120 行)Phase 5: 流式输出(+~80 行)Phase 6: 重试 + 超时 (+~128 行)Phase 7: 结构化输出(+~140 行)← 本文