推理模型 × Agent：Plan-and-Execute 架构

为什么推理模型是天然的规划器

Plan-and-Execute 完整实现

import anthropic
import json
from typing import List, Dict, Callable, Any

client = anthropic.Anthropic()

# ── 工具定义 ─────────────────────────────────────────────
def search_web(query: str) -> str:
    """模拟 Web 搜索工具"""
    return f"[搜索结果: {query}] 找到 5 篇相关文章..."

def calculate(expression: str) -> str:
    """安全计算工具"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

TOOL_MAP: Dict[str, Callable] = {
    "search_web": search_web,
    "calculate": calculate,
}

# ── 规划器：使用推理模型一次性规划所有步骤 ──────────────
def reasoning_planner(goal: str, available_tools: List[str]) -> List[Dict]:
    """
    使用推理模型生成完整的执行计划
    返回步骤列表，每步包含 tool、input、purpose
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=10000,
        thinking={"type": "enabled", "budget_tokens": 5000},
        # 系统提示：只说背景，不指导推理
        system=f"""你是任务规划专家。
可用工具：{available_tools}
输出严格的 JSON 列表（不含注释）：
[{{"step": 1, "tool": "工具名", "input": "输入参数", "purpose": "这步的目的"}}]""",
        messages=[{"role": "user", "content": f"目标：{goal}"}]
    )

    # 提取 text block（推理已在 thinking 中完成）
    text = next(b.text for b in response.content if b.type == "text")
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        # JSON 解析失败时，让小模型修复格式
        fix_response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=1000,
            messages=[{"role": "user",
                       "content": f"将以下内容修复为合法 JSON 列表：\n{text}"}]
        )
        return json.loads(fix_response.content[0].text)

# ── 执行器：用快速模型逐步执行计划 ─────────────────────
def execute_plan(plan: List[Dict]) -> Dict:
    """逐步执行计划，支持步骤间的结果引用"""
    results = {}
    context = {}  # 存储每步结果，供后续步骤引用

    for step in plan:
        step_num = step["step"]
        tool_name = step["tool"]
        tool_input = step["input"]

        # 支持在 input 中引用前面步骤的结果 {step1_result}
        try:
            tool_input = tool_input.format(**context)
        except KeyError:
            pass  # 没有需要替换的引用

        # 调用对应工具
        if tool_name in TOOL_MAP:
            result = TOOL_MAP[tool_name](tool_input)
        else:
            result = f"工具 {tool_name} 不存在"

        context[f"step{step_num}_result"] = result
        results[step_num] = {
            "tool": tool_name,
            "input": tool_input,
            "output": result,
            "purpose": step["purpose"]
        }

    return results

# ── 汇总器：将执行结果综合为最终答案 ───────────────────
def synthesize_results(goal: str, execution_results: Dict) -> str:
    results_text = "\n".join([
        f"步骤{k}（{v['purpose']}）: {v['output']}"
        for k, v in execution_results.items()
    ])
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2000,
        messages=[{"role": "user", "content":
            f"原始目标：{goal}\n\n执行结果：\n{results_text}\n\n请综合以上结果，给出完整答案："}]
    )
    return response.content[0].text

# ── 完整流程 ────────────────────────────────────────────
goal = "计算 Tesla 的市盈率并分析是否被高估"
tools = ["search_web", "calculate"]

plan = reasoning_planner(goal, tools)
print(f"规划了 {len(plan)} 个步骤")

results = execute_plan(plan)
final_answer = synthesize_results(goal, results)
print(final_answer)

ReWOO：推理与观察完全解耦

ReWOO（Reasoning WithOut Observation）是 Plan-and-Execute 的极致版本：规划器一次性生成所有工具调用计划（包括结果的引用占位符），然后工具并行执行，最后综合结果。核心优势是工具调用可以并行，大幅降低延迟。

import asyncio

async def rewoo_agent(task: str, tools: dict) -> str:
    """
    ReWOO 三阶段：Planner → Worker（并行）→ Solver
    适合工具调用可以并行的场景（如：同时搜索多个话题）
    """
    # ── Phase 1: Planner（一次性生成所有工具调用计划）────
    plan_prompt = f"""为以下任务制定工具调用计划（只规划，不执行）：
任务：{task}
可用工具：{list(tools.keys())}

计划格式：
#E1 = tool_name["输入参数"]
#E2 = tool_name["输入，可引用 #E1 的结果"]

注意：可以并行执行的步骤请分组标注"""

    planner_response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=6000,
        thinking={"type": "enabled", "budget_tokens": 3000},
        messages=[{"role": "user", "content": plan_prompt}]
    )
    plan_text = next(b.text for b in planner_response.content if b.type == "text")
    tool_plan = parse_rewoo_plan(plan_text)  # 解析 #E1=... 格式

    # ── Phase 2: Worker（并行执行所有工具调用）──────────
    evidence = {}
    # 找出哪些步骤可以并行（不依赖其他步骤的结果）
    independent_steps = [s for s in tool_plan if "#E" not in s["input"]]
    dependent_steps = [s for s in tool_plan if "#E" in s["input"]]

    # 并行执行独立步骤
    independent_tasks = [
        call_tool_async(s["tool"], s["input"])
        for s in independent_steps
    ]
    results = await asyncio.gather(*independent_tasks)
    for step, result in zip(independent_steps, results):
        evidence[step["id"]] = result

    # 顺序执行依赖步骤（替换引用占位符）
    for step in dependent_steps:
        actual_input = step["input"]
        for ref_id, ref_val in evidence.items():
            actual_input = actual_input.replace(ref_id, ref_val)
        result = await call_tool_async(step["tool"], actual_input)
        evidence[step["id"]] = result

    # ── Phase 3: Solver（综合证据给出最终答案）──────────
    evidence_text = "\n".join([f"{k}: {v}" for k, v in evidence.items()])
    final = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2000,
        messages=[{"role": "user", "content":
            f"任务：{task}\n\n工具执行结果：\n{evidence_text}\n\n综合以上信息给出完整答案："}]
    )
    return final.content[0].text

推理模型在 Agent 中的最佳实践总结