Agent 输出下游解析不了

Agent 生成的输出格式不稳定,下游系统在解析 JSON、代码块或结构化字段时频繁失败。本文分析格式漂移根因并给出强制结构化输出方案。

你的流水线要求 Agent 输出 JSON,但 LLM 有时在 JSON 前加了一句「以下是结果:」,有时在 JSON 后加了「如有疑问请联系我」,有时直接输出了 Markdown 代码块包裹的 JSON——而你的解析代码只处理了裸 JSON 的情况,于是每隔几次就 json.JSONDecodeError。或者 Agent 输出的 Python 代码块有时用 ```python,有时用 ```py,有时直接不带语言标记,正则提取逻辑只覆盖了前两种,第三种静默失败,下游拿到空字符串。

常见原因

1. 没有使用模型的结构化输出功能

OpenAI 的 response_format={"type": "json_object"} 和 Anthropic 的 tool use 模式都能强制模型输出合法 JSON,但很多开发者仍在用自由文本 prompt(「请以 JSON 格式输出」)来控制格式,这在边界输入下必然不稳定。

怎么判断:检查 API 调用参数,是否有 response_formattools 参数。如果只靠 prompt 描述格式,就是这个根因。

2. System prompt 里的格式要求在长对话中被遗忘

多轮对话里,早期设置的「必须输出 JSON」指令随着轮数增加,逐渐被模型的 attention 稀释。在第 20 轮时,模型可能已经「忘记」了格式要求,开始输出自然语言。

怎么判断:统计格式错误与对话轮数的相关性。如果前 5 轮正常,10 轮后开始出错,就是这个问题。

3. 模型在输出前后添加了解释性文字

即使使用了 json_object 模式,某些模型版本在特定情况下仍会在 JSON 前后添加自然语言前缀或后缀(如「根据您的请求,这是 JSON 格式的结果:...」)。解析代码如果直接 json.loads(response) 会失败。

怎么判断:打印 5 条实际的模型输出,检查 JSON 前后是否有非 JSON 字符。

4. 嵌套引号或特殊字符导致 JSON 损坏

模型生成的 JSON 字符串字段里包含未转义的引号(如 "message": "He said "hello"")或换行符,导致 JSON 语法错误。这在要求模型把代码或自然语言文本放进 JSON 字段时很常见。

怎么判断:用 json.JSONDecodeErrorpos 属性定位错误位置,检查该位置周围是否有未转义的特殊字符。

5. 代码块语言标记不统一

不同的模型版本、不同的 prompt 措辞,会让模型生成 ```json```JSON```javascript、或无标记代码块,而下游的正则只匹配了其中一种格式。

怎么判断:收集 100 条输出,用 re.findall(r'```(\w*)', output) 统计实际出现的语言标记,与解析代码支持的标记列表对比。

6. 解析代码假设单个 JSON 对象,但模型输出了 JSON 数组或多个对象

任务要求模型分析 5 个问题并各自给出结论,模型输出了 5 个独立的 JSON 对象(换行分隔),但解析代码只调用了一次 json.loads(),只能解析第一个对象,其余被静默丢弃。

怎么判断:检查模型输出里是否有多个 JSON 对象(每行一个),与解析代码的期望格式对比。

最短修复路径

Step 1:对 OpenAI 启用结构化输出,对 Anthropic 用 tool use

# OpenAI:强制 JSON 模式
response = client.chat.completions.create(
    model="gpt-4o",
    response_format={"type": "json_object"},
    messages=[{"role": "user", "content": prompt}]
)
data = json.loads(response.choices[0].message.content)

# Anthropic:用 tool use 强制结构化输出
tools = [{
    "name": "submit_result",
    "description": "提交分析结果",
    "input_schema": {
        "type": "object",
        "properties": {
            "status": {"type": "string", "enum": ["pass", "fail", "warning"]},
            "issues": {"type": "array", "items": {"type": "string"}},
            "score": {"type": "integer", "minimum": 0, "maximum": 100}
        },
        "required": ["status", "issues", "score"]
    }
}]
response = client.messages.create(
    model="claude-sonnet-4-6",
    tools=tools,
    tool_choice={"type": "tool", "name": "submit_result"},
    messages=[{"role": "user", "content": prompt}]
)
# tool_use block 保证是合法的 JSON
result = response.content[0].input

Step 2:写健壮的 JSON 提取函数,处理常见脏数据

import re, json

def extract_json(text: str) -> dict | list:
    """从模型输出里提取 JSON,处理前缀/后缀文字和代码块包裹。"""
    # 1. 尝试直接解析
    try:
        return json.loads(text.strip())
    except json.JSONDecodeError:
        pass
    
    # 2. 尝试提取代码块里的 JSON(处理多种语言标记)
    patterns = [
        r'```(?:json|JSON|javascript|js)?\s*\n([\s\S]*?)\n```',
        r'```([\s\S]*?)```',
    ]
    for pattern in patterns:
        match = re.search(pattern, text)
        if match:
            try:
                return json.loads(match.group(1).strip())
            except json.JSONDecodeError:
                continue
    
    # 3. 尝试找第一个 JSON 对象或数组
    json_match = re.search(r'(\{[\s\S]*\}|\[[\s\S]*\])', text)
    if json_match:
        try:
            return json.loads(json_match.group(1))
        except json.JSONDecodeError:
            pass
    
    raise ValueError(f"无法从输出中提取合法 JSON,原始输出长度:{len(text)}")

Step 3:用 Pydantic 校验提取结果的 schema

from pydantic import BaseModel, ValidationError

class AnalysisResult(BaseModel):
    status: str
    issues: list[str]
    score: int

def parse_agent_output(raw: str) -> AnalysisResult:
    try:
        data = extract_json(raw)
        return AnalysisResult(**data)
    except (ValueError, ValidationError) as e:
        # 解析失败时记录原始输出,方便调试
        logger.error(f"输出解析失败:{e}\n原始输出:{raw[:500]}")
        raise OutputParseError(f"Agent 输出不符合预期格式:{e}")

Step 4:在 prompt 里加负面示例

输出要求:严格的 JSON 对象,不要加任何前缀或后缀文字。

正确格式:
{"status": "pass", "issues": [], "score": 95}

错误格式(不要这样做):
好的,以下是分析结果:
{"status": "pass", ...}

Step 5:建立格式解析率监控

# 每 100 次解析记录一次成功率
parse_success = 0
parse_total = 0

def tracked_parse(raw: str) -> AnalysisResult:
    global parse_success, parse_total
    parse_total += 1
    try:
        result = parse_agent_output(raw)
        parse_success += 1
        return result
    except OutputParseError:
        if parse_total % 100 == 0:
            rate = parse_success / parse_total
            if rate < 0.95:
                alert(f"输出解析成功率低于 95%:{rate:.1%}")
        raise

预防建议

  • 对支持结构化输出的模型,优先使用 response_format 或 tool use,不要只靠 prompt 描述格式
  • 解析函数要处理所有已知的脏数据情况(代码块包裹、前后缀文字、多种语言标记)
  • 用 Pydantic 或 JSON Schema 对提取结果做 schema 验证,不要假设字段一定存在
  • 在测试集里维护 20-30 条真实的模型输出样本(包括边界情况),解析函数修改时跑回归测试
  • 对解析成功率做生产监控,低于 95% 时告警
  • prompt 里加负面示例(「不要在 JSON 前后加文字」),比正面描述更有效
  • 多轮对话里,在每次用户消息里复述格式要求(不只在 system prompt 里说一次)

常见问答 (FAQ)

Q: Anthropic 的 tool use 和 OpenAI 的 function calling 能 100% 保证输出是合法 JSON 吗? A: 几乎可以,但不是绝对的。在极少数情况下(如模型 API 故障、网络截断),响应体可能不完整。应该始终把解析放在 try/except 里,并记录解析失败的原始响应。

Q: 用 Pydantic 校验时字段缺失怎么处理? A: 对非关键字段设置默认值(field(default=None)field(default_factory=list)),对关键字段标记为 Required(无默认值)让 Pydantic 在缺失时抛 ValidationError。不要用 model.dict(exclude_none=True) 静默丢弃缺失字段。

Q: 如果模型输出了 JSON 但语义不对(字段值错误),如何检测? A: 这是 schema 验证之上的业务逻辑验证。在 Pydantic model 里用 @field_validator 添加业务规则(如 score 必须在 0-100 之间,status 必须是枚举值),超出范围时抛异常。还可以在 Agent 输出后加一个专门的「语义验证」节点,对输出的业务含义做二次检查。

Q: 下游系统能否容忍偶发的解析失败? A: 取决于业务场景。如果是异步批处理任务,可以把解析失败的任务放入「重试队列」,稍后重新请求模型。如果是实时交互场景,应该立即重试一次(模型的输出有随机性,重试通常成功),第二次还失败则降级为人工处理。

相关阅读

标签: #AI 编程 #Agents #排查