第 6 章

OpenAI 兼容端点与零改动迁移:从 GPT-4 到 Claude 的完整迁移指南

第六章:响应格式控制:JSON 模式、XML 标签与结构化输出

6.1 为什么响应格式控制至关重要

在生产系统中,Claude 的输出通常不是直接展示给用户的,而是作为数据管道中的一个环节——被解析、存储、传递到下一个系统。如果输出格式不可靠,整个管道就会在运行时失败。

格式控制的核心挑战:语言模型本质上是概率系统,它生成的是"在这个上下文中概率最高的下一个 token",而不是"严格符合 JSON schema 的字符串"。即使给了明确的 JSON 格式要求,模型偶尔也可能:

本章系统介绍三种主要方法:Prompt 级别的格式指令、XML 标签组织、以及通过工具调用(Tool Use)强制结构化输出。

6.2 Prompt 级别的 JSON 格式控制

基础方法:直接指令

最简单的方法是在 prompt 中明确要求 JSON 输出:

import json
import anthropic

client = anthropic.Anthropic()

def extract_product_info(product_description: str) -> dict:
    """从产品描述中提取结构化信息"""
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="""你是一个产品信息提取器。
从用户提供的产品描述中提取结构化信息。

你必须只返回有效的 JSON,不要包含任何其他文字、代码块标记或解释。
JSON 结构:
{
  "name": "产品名称",
  "category": "产品类别",
  "price": 数字(如果提到)或 null,
  "features": ["特性1", "特性2", ...],
  "target_audience": "目标用户群体"
}""",
        messages=[
            {"role": "user", "content": product_description}
        ]
    )
    
    raw_output = response.content[0].text.strip()
    
    try:
        return json.loads(raw_output)
    except json.JSONDecodeError as e:
        # 尝试修复常见问题
        return repair_json(raw_output, e)

def repair_json(text: str, error: json.JSONDecodeError) -> dict:
    """尝试修复轻微的 JSON 格式问题"""
    import re
    
    # 去除可能的代码块标记
    text = re.sub(r'^```json\s*', '', text, flags=re.MULTILINE)
    text = re.sub(r'^```\s*', '', text, flags=re.MULTILINE)
    text = re.sub(r'\s*```$', '', text, flags=re.MULTILINE)
    text = text.strip()
    
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        raise ValueError(f"无法解析模型输出为 JSON: {text[:200]}...") from error

更可靠的方法:预填充 Assistant 响应

Claude 支持"assistant prefill"技巧:在 messages 数组中预先填充 assistant 的响应开头,强制模型在此基础上继续生成:

def extract_with_prefill(text: str) -> dict:
    """
    使用 Assistant Prefill 强制 JSON 输出
    原理:预填充 "{",模型必须从这里继续,几乎不可能生成非 JSON 内容
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="从文本中提取信息,以 JSON 格式返回。只返回 JSON,不要其他内容。",
        messages=[
            {"role": "user", "content": f"提取以下文本的关键信息:\n\n{text}"},
            {"role": "assistant", "content": "{"}  # 预填充开头
        ]
    )
    
    # 模型从 { 之后继续,需要手动加上开头的 {
    json_str = "{" + response.content[0].text
    
    return json.loads(json_str)

提供 JSON Schema 作为指导

对于复杂的 JSON 结构,提供 JSON Schema 能显著减少格式错误:

INVOICE_SCHEMA = {
    "type": "object",
    "required": ["invoice_number", "date", "total", "line_items"],
    "properties": {
        "invoice_number": {"type": "string"},
        "date": {"type": "string", "format": "date"},
        "vendor": {"type": "string"},
        "total": {"type": "number"},
        "currency": {"type": "string", "enum": ["USD", "EUR", "CNY", "JPY"]},
        "line_items": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["description", "quantity", "unit_price"],
                "properties": {
                    "description": {"type": "string"},
                    "quantity": {"type": "number"},
                    "unit_price": {"type": "number"},
                    "total": {"type": "number"}
                }
            }
        }
    }
}

def extract_invoice(invoice_text: str) -> dict:
    schema_str = json.dumps(INVOICE_SCHEMA, ensure_ascii=False, indent=2)
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""从以下发票文本中提取数据,严格按照提供的 JSON Schema 格式输出。
只返回 JSON,不要解释。

JSON Schema:
{schema_str}

发票文本:
{invoice_text}"""
        }]
    )
    
    return json.loads(response.content[0].text.strip())

6.3 XML 标签:组织复杂的多部分输出

当响应需要包含多个不同性质的部分时,XML 标签比 JSON 更灵活。特别适合:

基本 XML 标签模式

def analyze_code_with_xml(code: str) -> dict:
    """
    要求模型用 XML 标签分隔不同类型的输出
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system="""你是代码审查专家。按以下 XML 格式回复:

<analysis>
<severity>critical|high|medium|low|none</severity>
<issues>
  <issue>
    <type>bug|performance|security|style</type>
    <line>行号(如果适用)</line>
    <description>问题描述</description>
  </issue>
</issues>
<fixed_code>修复后的代码</fixed_code>
<explanation>修改说明</explanation>
</analysis>""",
        messages=[{"role": "user", "content": f"审查以下代码:\n\n```python\n{code}\n```"}]
    )
    
    return parse_xml_response(response.content[0].text)

import re
from xml.etree import ElementTree as ET

def parse_xml_response(text: str) -> dict:
    """解析包含 XML 标签的响应"""
    # 提取 <analysis>...</analysis> 块
    match = re.search(r'<analysis>(.*?)</analysis>', text, re.DOTALL)
    if not match:
        raise ValueError(f"响应中未找到 <analysis> 标签: {text[:200]}")
    
    xml_content = f"<analysis>{match.group(1)}</analysis>"
    
    try:
        root = ET.fromstring(xml_content)
        
        result = {
            "severity": root.findtext("severity"),
            "issues": [],
            "fixed_code": root.findtext("fixed_code"),
            "explanation": root.findtext("explanation")
        }
        
        for issue in root.findall(".//issue"):
            result["issues"].append({
                "type": issue.findtext("type"),
                "line": issue.findtext("line"),
                "description": issue.findtext("description")
            })
        
        return result
    except ET.ParseError as e:
        raise ValueError(f"XML 解析失败: {e}") from e

用正则表达式提取特定标签

对于只需要提取特定标签的场景,正则比完整 XML 解析更简单:

def extract_tagged_sections(text: str, tags: list[str]) -> dict[str, str]:
    """
    从响应文本中提取指定 XML 标签的内容
    
    示例:
    extract_tagged_sections(text, ["answer", "confidence", "sources"])
    """
    result = {}
    for tag in tags:
        pattern = rf'<{tag}>(.*?)</{tag}>'
        match = re.search(pattern, text, re.DOTALL)
        if match:
            result[tag] = match.group(1).strip()
        else:
            result[tag] = None
    return result


# 使用示例
response_text = """
<answer>
量子纠缠是两个粒子之间的非局域关联现象。
</answer>
<confidence>0.95</confidence>
<sources>
- 物理学教材第三章
- Einstein-Podolsky-Rosen 1935 年论文
</sources>
"""

extracted = extract_tagged_sections(
    response_text, 
    ["answer", "confidence", "sources"]
)
print(extracted["answer"])      # 量子纠缠是两个粒子...
print(extracted["confidence"])  # 0.95

6.4 工具调用(Tool Use):最可靠的结构化输出方式

工具调用(Tool Use / Function Calling)是强制结构化输出最可靠的机制。它的工作原理:定义一个工具(函数),模型在需要"调用"该工具时,必须按照你定义的 JSON schema 生成参数。这绕过了自然语言生成的不确定性。

将 Tool Use 用于纯数据提取

即使你不真正"调用"任何工具,也可以用 tool use 强制模型输出符合特定 schema 的 JSON:

import anthropic
import json

client = anthropic.Anthropic()

def extract_entities_structured(text: str) -> dict:
    """
    用 Tool Use 强制结构化输出
    这个工具不会真正执行,只是用来强制格式
    """
    tools = [
        {
            "name": "save_entities",
            "description": "保存从文本中提取的实体信息",
            "input_schema": {
                "type": "object",
                "required": ["persons", "organizations", "locations", "dates"],
                "properties": {
                    "persons": {
                        "type": "array",
                        "description": "提到的人名列表",
                        "items": {
                            "type": "object",
                            "required": ["name", "role"],
                            "properties": {
                                "name": {"type": "string"},
                                "role": {"type": "string", "description": "此人在文本中的角色或头衔"}
                            }
                        }
                    },
                    "organizations": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "提到的组织名称"
                    },
                    "locations": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "提到的地点"
                    },
                    "dates": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "提到的日期或时间点"
                    }
                }
            }
        }
    ]
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        tool_choice={"type": "tool", "name": "save_entities"},  # 强制使用这个工具
        messages=[{
            "role": "user",
            "content": f"从以下文本中提取所有实体:\n\n{text}"
        }]
    )
    
    # 从工具调用中提取 JSON
    for block in response.content:
        if block.type == "tool_use" and block.name == "save_entities":
            return block.input
    
    raise ValueError("模型未生成工具调用")


# 使用示例
text = """
2024年3月15日,阿里巴巴集团在杭州总部举行了年度合作伙伴大会。
CEO 张勇宣布与谷歌(Google)签署了新的战略合作协议。
活动在杭州国际博览中心举行,共有来自全球各地的500余家企业参与。
"""

result = extract_entities_structured(text)
print(json.dumps(result, ensure_ascii=False, indent=2))
# {
#   "persons": [{"name": "张勇", "role": "CEO"}],
#   "organizations": ["阿里巴巴集团", "谷歌"],
#   "locations": ["杭州", "杭州国际博览中心"],
#   "dates": ["2024年3月15日"]
# }

多工具组合:让模型选择合适的输出类型

def classify_and_extract(user_query: str) -> dict:
    """
    让模型根据输入内容选择合适的提取工具
    """
    tools = [
        {
            "name": "classify_sentiment",
            "description": "当用户输入是情感分析请求时使用",
            "input_schema": {
                "type": "object",
                "required": ["sentiment", "score", "explanation"],
                "properties": {
                    "sentiment": {
                        "type": "string",
                        "enum": ["positive", "negative", "neutral", "mixed"]
                    },
                    "score": {
                        "type": "number",
                        "minimum": -1.0,
                        "maximum": 1.0,
                        "description": "情感分数:-1.0(最负面)到 1.0(最正面)"
                    },
                    "explanation": {"type": "string"}
                }
            }
        },
        {
            "name": "extract_key_facts",
            "description": "当用户输入是事实性文本需要信息提取时使用",
            "input_schema": {
                "type": "object",
                "required": ["facts", "summary"],
                "properties": {
                    "facts": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "fact": {"type": "string"},
                                "confidence": {"type": "number"}
                            }
                        }
                    },
                    "summary": {"type": "string"}
                }
            }
        }
    ]
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        tool_choice={"type": "auto"},  # 让模型自动选择工具
        messages=[{"role": "user", "content": user_query}]
    )
    
    for block in response.content:
        if block.type == "tool_use":
            return {
                "tool_used": block.name,
                "data": block.input
            }
    
    # 模型选择不使用工具(直接文本响应)
    text_blocks = [b for b in response.content if b.type == "text"]
    return {
        "tool_used": None,
        "text": text_blocks[0].text if text_blocks else ""
    }

6.5 解析器的健壮性设计

无论使用哪种格式控制方法,生产系统都需要健壮的解析器来处理格式不完美的输出。

宽容型 JSON 解析器

import re
import json
from typing import Any

def robust_json_parse(text: str) -> Any:
    """
    尽力解析可能包含格式问题的 JSON 字符串
    处理常见问题:代码块标记、前后说明文字、单引号等
    """
    # 1. 尝试直接解析
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    
    # 2. 去除代码块标记
    cleaned = re.sub(r'^```(?:json)?\s*\n?', '', text.strip(), flags=re.MULTILINE)
    cleaned = re.sub(r'\n?```\s*$', '', cleaned, flags=re.MULTILINE)
    cleaned = cleaned.strip()
    
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        pass
    
    # 3. 提取 JSON 对象或数组
    # 找到第一个 { 或 [ 开始
    json_match = re.search(r'(\{[\s\S]*\}|\[[\s\S]*\])', cleaned)
    if json_match:
        try:
            return json.loads(json_match.group(1))
        except json.JSONDecodeError:
            pass
    
    # 4. 使用容错 JSON 库(如果安装了的话)
    try:
        import json5  # pip install json5
        return json5.loads(cleaned)
    except (ImportError, Exception):
        pass
    
    raise ValueError(f"无法解析 JSON: {text[:300]}")

带重试的格式修复

当解析失败时,可以让 Claude 自己修复格式问题:

def parse_with_repair(raw_output: str, expected_schema: dict) -> dict:
    """
    解析失败时,让 Claude 修复 JSON 格式
    """
    try:
        return robust_json_parse(raw_output)
    except ValueError:
        # 让 Claude 修复格式
        repair_response = client.messages.create(
            model="claude-haiku-4-5-20251001",  # 用便宜的模型做修复
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""以下文本应该是 JSON 格式,但无法解析。
请修复格式问题,只返回有效的 JSON,不要任何其他文字。

需要符合的 Schema:
{json.dumps(expected_schema, ensure_ascii=False)}

需要修复的文本:
{raw_output}"""
            }]
        )
        
        repaired = repair_response.content[0].text.strip()
        return json.loads(repaired)  # 如果还是失败,让异常传播

6.6 格式一致性测试框架

生产系统中,格式控制的有效性需要持续监控:

import json
import jsonschema
from typing import Callable
from dataclasses import dataclass

@dataclass
class FormatTestResult:
    success: bool
    parse_errors: int
    schema_errors: int
    total: int
    success_rate: float
    error_examples: list[dict]

def test_format_consistency(
    prompt_fn: Callable[[str], str],  # 接受输入,返回模型输出
    test_inputs: list[str],
    expected_schema: dict,
    max_error_examples: int = 3
) -> FormatTestResult:
    """
    测试格式控制的一致性
    
    prompt_fn: 接受输入文本,返回模型响应
    test_inputs: 测试输入列表
    expected_schema: 期望输出的 JSON schema
    """
    parse_errors = 0
    schema_errors = 0
    error_examples = []
    
    for inp in test_inputs:
        raw_output = prompt_fn(inp)
        
        try:
            parsed = robust_json_parse(raw_output)
            jsonschema.validate(parsed, expected_schema)
        except (ValueError, json.JSONDecodeError) as e:
            parse_errors += 1
            if len(error_examples) < max_error_examples:
                error_examples.append({
                    "input": inp[:100],
                    "output": raw_output[:200],
                    "error": str(e),
                    "error_type": "parse"
                })
        except jsonschema.ValidationError as e:
            schema_errors += 1
            if len(error_examples) < max_error_examples:
                error_examples.append({
                    "input": inp[:100],
                    "output": raw_output[:200],
                    "error": e.message,
                    "error_type": "schema"
                })
    
    total = len(test_inputs)
    total_errors = parse_errors + schema_errors
    
    return FormatTestResult(
        success=total_errors == 0,
        parse_errors=parse_errors,
        schema_errors=schema_errors,
        total=total,
        success_rate=(total - total_errors) / total if total > 0 else 0,
        error_examples=error_examples
    )

6.7 三种方法的选择指南

方法对比总结:

                   Prompt 指令    XML 标签    Tool Use
─────────────────  ───────────    ────────    ────────
实现复杂度          低             低          中
格式可靠性          中(~90%)     高(~95%)  最高(~99%+)
适用场景            简单提取       多部分输出  关键路径提取
数据类型支持        JSON/文本      任意文本    强类型 JSON schema
解析复杂度          需要健壮解析器  需要 XML    SDK 直接返回 dict
成本开销            最低           低          略高(工具定义占用 tokens)

选择建议


小结

响应格式控制是将 Claude 集成到生产系统的关键技术:

  1. Prompt 指令:最简单,但需要健壮的解析器处理偶发格式错误;Assistant Prefill 能显著提高可靠性
  2. XML 标签:适合多部分混合输出,正则提取简单高效
  3. Tool Use:最可靠的结构化输出方式,通过 schema 定义强制格式,几乎消除格式错误
  4. 健壮解析器:无论选择哪种方法,都应该有解析失败时的降级处理
  5. 持续监控:用格式一致性测试框架监控生产环境的格式质量

下一章开始深入 Messages API 的完整参数参考,涵盖所有请求参数的详细语义和使用陷阱。

本章评分
4.7  / 5  (84 评分)

💬 留言讨论