结构化输出 + 推理：JSON 约束下的思维链

核心矛盾：自由推理 vs 格式约束

方案一：Two-Stage 架构（最推荐）

最可靠的解耦方案：第一阶段让推理模型自由分析，第二阶段让小模型将分析结果格式化为 JSON。

import anthropic
import json
from pydantic import BaseModel, Field
from typing import List, Optional

client = anthropic.Anthropic()

# ── 定义目标数据结构 ─────────────────────────────────────
class BugInfo(BaseModel):
    line: int                    # bug 所在行号
    severity: str               # high / medium / low
    description: str            # 问题描述

class CodeReviewResult(BaseModel):
    bugs: List[BugInfo]
    suggestions: List[str]
    overall_score: int = Field(ge=0, le=10)  # 0-10 分
    summary: str

def reasoning_code_review(code: str) -> CodeReviewResult:
    """Two-Stage 代码审查：深度推理 → 结构化提取"""

    # ── Stage 1: 推理模型深度分析 ───────────────────────
    reasoning_response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=10000,
        thinking={"type": "enabled", "budget_tokens": 6000},
        # 不要在这里要求 JSON，让模型自由分析
        system="你是资深 Python 代码审查专家，有 15 年工程经验。请全面、深入地分析代码。",
        messages=[{"role": "user",
                   "content": f"请深度审查以下代码，找出所有 bug、潜在问题和改进点：\n```python\n{code}\n```"}]
    )
    # 提取推理后的文字分析（质量由推理模型保证）
    analysis_text = next(b.text for b in reasoning_response.content if b.type == "text")

    # ── Stage 2: 小模型格式化（不需要推理）──────────────
    extract_response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 便宜的小模型做格式转换
        max_tokens=2000,
        system="""将代码审查分析提取为 JSON。
严格输出合法 JSON，不要任何其他内容，不要注释，不要 markdown 代码块：
{"bugs": [{"line": 行号, "severity": "high/medium/low", "description": "描述"}],
 "suggestions": ["建议1"],
 "overall_score": 0到10之间的整数,
 "summary": "一句话总结"}""",
        messages=[{"role": "user",
                   "content": f"请从以下审查分析中提取结构化信息：\n\n{analysis_text}"}]
    )

    # 解析并验证 JSON（Pydantic 自动验证字段类型和约束）
    raw_json = extract_response.content[0].text.strip()
    return CodeReviewResult.model_validate_json(raw_json)

# 使用示例
buggy_code = """
def process_data(items):
    result = []
    for i in range(len(items)):
        if items[i] > 0:
            result.append(items[i] * 2)
    return result[0]  # 可能 IndexError
"""

review = reasoning_code_review(buggy_code)
print(f"评分：{review.overall_score}/10")
print(f"发现 {len(review.bugs)} 个 bug")
for bug in review.bugs:
    print(f"  [{bug.severity}] 行{bug.line}: {bug.description}")

方案二：XML 标签隔离（单模型方案）

如果希望用单次 API 调用完成推理和格式化，可以用 XML 标签将推理区域和输出区域显式分离：

DUAL_ZONE_SYSTEM = """你是代码审查专家。
工作方式：
1. 在 <analysis> 标签内进行自由、深入的思考和分析（格式任意）
2. 在 <output> 标签内输出严格合法的 JSON（不含注释，不含多余文字）

严格格式：
<analysis>
[你的自由分析内容，可以包含推导过程、权衡、不确定性等]
</analysis>
<output>
{"bugs": [...], "suggestions": [...], "overall_score": 数字, "summary": "字符串"}
</output>"""

import re

def dual_zone_review(code: str) -> dict:
    """单次调用：XML 标签分离推理区和输出区"""
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=6000,
        # 这里不启用 Extended Thinking，而是用 XML 标签模拟分离
        system=DUAL_ZONE_SYSTEM,
        messages=[{"role": "user",
                   "content": f"请审查：\n```python\n{code}\n```"}]
    )
    text = response.content[0].text

    # 提取 <analysis> 中的推理过程（可选，用于调试）
    analysis_match = re.search(r'<analysis>\s*(.*?)\s*</analysis>', text, re.DOTALL)
    analysis = analysis_match.group(1) if analysis_match else ""

    # 提取 <output> 中的 JSON
    output_match = re.search(r'<output>\s*(.*?)\s*</output>', text, re.DOTALL)
    if not output_match:
        raise ValueError("模型未输出 <output> 标签")

    return json.loads(output_match.group(1))

# XML 标签方案 vs Two-Stage 的选择：
# - 需要推理模型质量 → Two-Stage（Stage 1 可用 Extended Thinking）
# - 成本优先，单次调用 → XML 标签
# - 需要向用户展示思考过程 → Two-Stage（可以把 analysis 展示给用户）

方案三：Instructor 库（自动化提取）

Instructor 是一个流行的 Python 库，将 Pydantic 模型与 LLM 结构化输出自动整合，极大简化了结构化提取的代码。

# 安装：pip install instructor anthropic pydantic
import instructor
from pydantic import BaseModel, Field
from typing import List
import anthropic

# Instructor 封装 Claude 客户端（自动处理格式验证和重试）
client_instructor = instructor.from_anthropic(
    anthropic.Anthropic(),
    mode=instructor.Mode.ANTHROPIC_TOOLS  # 使用工具调用实现结构化输出
)

class TechRecommendation(BaseModel):
    recommended_stack: List[str] = Field(
        description="推荐的技术栈列表，按重要性排序"
    )
    rationale: str = Field(
        description="选择理由（简洁，100字以内）"
    )
    risks: List[str] = Field(
        description="潜在风险列表",
        min_length=1  # 至少要说一个风险
    )
    estimated_cost: str = Field(
        description="月度估算成本（人民币）"
    )

# Instructor 自动处理：JSON 生成 → 解析 → Pydantic 验证 → 重试（失败时）
result = client_instructor.chat.completions.create(
    model="claude-sonnet-4-6",
    max_tokens=2000,
    response_model=TechRecommendation,  # 指定 Pydantic 模型
    messages=[{"role": "user",
               "content": "为日活 50 万的电商应用推荐后端技术栈，团队 3 人，月预算 3000 元"}]
)

# 直接使用类型安全的 Pydantic 对象
print(result.recommended_stack)   # ['FastAPI', 'PostgreSQL', 'Redis']
print(result.estimated_cost)      # '约 2800 元/月'
print(result.risks)              # ['单点故障风险', '...'}

# Instructor + Extended Thinking 组合（需要用 Two-Stage）：
# Step 1: 用 Extended Thinking 进行深度分析
# Step 2: 将分析结果传给 Instructor 客户端进行格式化提取
def reasoning_then_instructor(question: str) -> TechRecommendation:
    # Stage 1：深度推理分析
    raw_client = anthropic.Anthropic()
    analysis_resp = raw_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=10000,
        thinking={"type": "enabled", "budget_tokens": 6000},
        messages=[{"role": "user", "content": question}]
    )
    analysis = next(b.text for b in analysis_resp.content if b.type == "text")

    # Stage 2：Instructor 格式化
    return client_instructor.chat.completions.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1000,
        response_model=TechRecommendation,
        messages=[{"role": "user",
                   "content": f"请将以下分析提取为结构化数据：\n{analysis}"}]
    )

鲁棒性处理：JSON 解析失败的应对策略

import json
import re
from typing import Optional

def robust_json_parse(text: str, fallback_schema: dict) -> dict:
    """多层降级的 JSON 解析策略"""

    # 策略1：直接解析（如果输出恰好是干净的 JSON）
    try:
        return json.loads(text.strip())
    except json.JSONDecodeError:
        pass

    # 策略2：提取 markdown 代码块中的 JSON
    code_block = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text)
    if code_block:
        try:
            return json.loads(code_block.group(1))
        except:
            pass

    # 策略3：提取第一个 { 到最后一个 } 之间的内容
    json_match = re.search(r'\{[\s\S]*\}', text)
    if json_match:
        try:
            return json.loads(json_match.group())
        except:
            pass

    # 策略4：让小模型修复 JSON 格式（最后防线）
    fix_response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1000,
        messages=[{"role": "user",
                   "content": f"以下内容应该是 JSON，但格式有问题。请修复为合法 JSON，只输出 JSON：\n\n{text[:2000]}"}]
    )
    try:
        return json.loads(fix_response.content[0].text.strip())
    except:
        # 最终降级：返回 fallback
        return fallback_schema