Reborn 的技术博客

核心增量：Structure.py（~140 行）+ run_agent_with_trace 中 10 行集成

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

场景 A: Agent 查询天气 → 下游需要 {city, temp, condition} 结构化数据
        → 从"北京今天晴天，25°C"中提取？准确率无法 100%
场景 B: LLM 偶尔在 JSON 前后加"这是结果："之类的废话，解析失败
场景 C: 输出混入 Markdown 或 HTML → 下游解析器崩溃

核心矛盾：LLM 擅长自然语言，但程序需要结构化、可校验、可序列化的数据。

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

不是让 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": "以结构化格式输出最终答案。调用此工具表示回答完成。",
            "parameters": {
                "type": "object",
                "properties": {
                    "result": {"type": "object", "description": "最终答案的 JSON 结构化数据"},
                    "summary": {"type": "string", "description": "给用户看的一句话总结"},
                },
                "required": ["result"],
            },
        },
    }

主循环中的终止逻辑

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:
        messages.append({"role": "assistant", "content": result})
        return result                       # ← 短路，不再循环

之前靠 if tool_calls is None 判断"回答完了"，但 LLM 可能在没查完资料时就输出不完整答案。用 final_output 作为显式终止信号更准确。

四、三层安全防线

第一层：Schema 安全校验

防止恶意构造的 Schema 攻击校验器本身：

MAX_SCHEMA_DEPTH = 5
MAX_DESCRIPTION_LEN = 200
MAX_PROPERTIES = 50

def validate_schema(schema: dict, depth: int = 0):
    if depth > MAX_SCHEMA_DEPTH:
        raise ValueError("Schema 嵌套过深")
    if schema.get("type") not in ("object",):
        raise ValueError("顶层 type 必须是 object")
    props = schema.get("properties", {})
    if len(props) > MAX_PROPERTIES:
        raise ValueError(f"properties 数量不超过 {MAX_PROPERTIES}")
    for key, prop in props.items():
        if len(prop.get("description", "")) > MAX_DESCRIPTION_LEN:
            raise ValueError(f"description 过长: {key}")
        if prop.get("type") == "object":
            validate_schema(prop, depth + 1)

第二层：jsonschema 严格校验

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

第三层：注入检测

INJECTION_PATTERNS = [
    (r"```", "Markdown 代码块"),
    (r"<script", "HTML script 标签"),
    (r"ignore.*previous.*instruction", "Prompt 注入 - ignore 指令"),
    (r"DROP\s+TABLE", "SQL 注入"),
    (r"\bpassword\b", "敏感词"),
    (r"\bsecret\b", "敏感词"),
]

def sanitize_string(value: str) -> str:
    for pattern, label in INJECTION_PATTERNS:
        if re.search(pattern, value, re.IGNORECASE):
            raise ValueError(f"检测到可疑内容 ({label}): {value[:80]}...")
    return value

sanitize_output 递归遍历整个输出结构，对每个字符串值执行注入检测。

五、工厂函数：为不同场景定制

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

六、设计精要

以 Tool 为输出载体，而非 Prompt 约束

tool 的 parameters 就是 JSON Schema，LLM 调用它天然结构化。

显式终止信号优于隐式判断

final_output 比"没有 tool_call"更准确。

安全是分层递进的

Schema 校验 → 格式校验 → 注入检测，三层递进。

零侵入集成

只改主循环中处理工具调用的一个分支，其他模块不受影响。