企业官网建设流程全解析

1. 项目概述：当大模型“边想边答”变成可拆解、可调度的流水线

ReWoo 这个名字乍一听像某个开源工具库的代号，但它的核心思想其实直击当前大语言模型推理范式的软肋——我们总在默认一个黑箱：用户抛出问题，模型内部一顿狂算，最后吐出答案。中间那层“思考过程”，要么被隐藏（如标准的 auto-regressive 生成），要么被强行外显但无法干预（如 chain-of-thought 提示）。ReWoo 不是教你怎么写更好的 prompt，而是从根本上重构了这个流程：它把“推理”（Reasoning）和“行动”（Working）彻底解耦，让模型先专注构建逻辑骨架，再按需调用外部工具或检索模块去填充血肉。这背后不是玄学，而是一套清晰、可编程、可审计的两阶段架构。关键词Reasoning、Working、decoupling、LLM reasoning architecture、tool-augmented inference在整个设计中反复出现，不是装饰词，而是每个模块命名、接口定义和数据流向的底层依据。如果你正被以下问题困扰——提示工程效果不稳定、复杂任务准确率上不去、需要频繁调用 API 却难以控制调用时机、或者想让模型的“思考路径”真正可追溯可调试——那么 ReWoo 提供的不是又一种 prompt 技巧，而是一套可落地的系统级解决方案。它适合两类人：一类是正在构建 RAG、Agent 或智能工作流的工程师，需要稳定、可控、低幻觉的推理链；另一类是研究者或技术决策者，想理解下一代 LLM 推理范式如何从“端到端拟合”转向“分阶段协同”。我第一次在论文里看到 ReWoo 的 workflow 图时，第一反应是：“这不就是我们团队去年做客服知识引擎时，手动拆出来的三段式逻辑吗？只是他们把它标准化、泛化了。” 后来实操下来发现，它不只是理论漂亮，关键在于每个环节的输入输出格式极其干净，连最基础的 Python 脚本都能无缝接入，这对快速验证和迭代太友好了。

2. 核心设计思路与架构解耦逻辑

2.1 为什么必须解耦？从“端到端黑箱”到“分阶段白盒”的必然性

要理解 ReWoo 的价值，得先看清传统做法的硬伤。主流的 CoT（Chain-of-Thought）方法，本质是让模型在生成答案前，先生成一段自然语言的推理草稿。比如问“小明有5个苹果，吃了2个，又买了3个，现在有几个？”，模型可能输出：“小明原来有5个苹果。他吃了2个，剩下5-2=3个。他又买了3个，所以现在有3+3=6个。答案是6。” 这看起来很聪明，但问题藏在细节里：这段推理文本是模型“编”出来的，不是“算”出来的。它没有强制校验机制，5-2 算成4也不会被拦住；它和后续的计算动作完全绑定，无法把“识别出需要计算”和“执行减法运算”这两个动作分开。更致命的是，一旦涉及真实世界操作——比如查天气、调用数据库、发邮件——CoT 的推理文本就只能停留在“想象层面”，模型会自信地编造一个“查到了，北京今天25度”，而实际根本没调用任何 API。ReWoo 的解耦，正是为了解决这个“意图与执行脱节”的根本矛盾。它把整个推理过程明确切成两个阶段：Reasoning 阶段只负责逻辑规划，不碰任何外部世界；Working 阶段只负责忠实执行，不参与任何逻辑判断。这种分离，不是为了炫技，而是为了引入确定性。就像一个严谨的软件工程流程：需求分析师（Reasoner）写好详细的需求文档（Plan），开发工程师（Worker）严格按文档编码（Execute），测试工程师（Evaluator）再独立验证结果。每个角色职责单一，错误定位就快，模块替换也容易。我试过把同一个 ReWoo 流程里的 Worker 换成不同实现——一个用 requests 调 REST API，一个用 sqlite3 查本地数据库，甚至一个用 print 模拟——只要 Plan 格式一致，整个流程就能跑通。这种灵活性，在传统端到端模型里是不可想象的。

2.2 ReWoo 架构的三层核心组件与数据契约

ReWoo 的架构看似简单，只有三个核心组件，但每个组件之间的“契约”（Contract）设计得极为精妙，这是它能稳定运行的关键。这三层不是松散拼凑，而是通过严格定义的数据格式紧密咬合：

1. Reasoner（推理器）：它的唯一输入是用户的原始 query，唯一输出是一个结构化的Plan。这个 Plan 不是自由文本，而是一个 JSON 数组，每个元素是一个带 type 和 args 的对象。例如，对于“查上海明天的天气并告诉我是否需要带伞”，Reasoner 可能输出：

   [ {"type": "weather_api", "args": {"city": "上海", "date": "tomorrow"}}, {"type": "decision_logic", "args": {"condition": "rain_probability > 0.5"}} ]

   注意：这里weather_api和decision_logic是 Worker 的注册名，args是传递给它的参数。Reasoner 完全不关心这些 type 对应的具体实现，它只负责“想清楚要调什么、传什么”。

2. Worker（执行器）：它是一组预先注册的、功能明确的函数集合。每个 Worker 函数接收一个args字典，执行具体操作，并返回一个结构化的result。关键点在于：Worker 必须是纯函数或有明确副作用的函数，且其输入输出格式必须与 Reasoner 输出的 Plan 中的 type 和 args 严格匹配。比如weather_apiWorker 的代码可能是：

   def weather_api(args): city = args["city"] date = args["date"] # 实际调用天气API... return {"temperature": 25, "rain_probability": 0.7, "condition": "cloudy"}

   它不处理任何逻辑分支，只管“干活”。

3. Orchestrator（编排器）：这是整个流程的“大脑”和“粘合剂”。它接收用户 query，调用 Reasoner 得到 Plan，然后按顺序（或并行）调用对应的 Worker，将每个 Worker 的result按索引存入一个context字典。最后，它把原始 query、完整的 Plan 和所有 Worker 的result一起喂给一个Final Answer Generator（通常就是一个微调过的或精心设计的 LLM），让它基于这些“事实性上下文”生成最终答案。Orchestrator 不做推理，也不执行动作，它只做三件事：调度、聚合、转发。

这个三层架构的威力，在于它把“不确定性”（Reasoner 的生成）和“确定性”（Worker 的执行）隔离开。Reasoner 可以用一个较小的、擅长规划的模型（如 Phi-3），Worker 可以是任何可靠的外部服务，Final Answer Generator 可以用一个更大的、擅长总结的模型（如 Qwen2.5）。它们可以独立升级、替换、监控，互不影响。我在一个电商客服项目里，就把 Reasoner 换成了自己微调的小模型（节省成本），Worker 保持调用公司内部的订单和库存 API，Final Answer Generator 则用客户指定的大模型。上线后，问题解决率提升了18%，而平均响应时间反而下降了22%，因为 Reasoner 规划得更准，Worker 调用更少、更精准。

2.3 解耦带来的四大核心优势与真实场景映射

ReWoo 的解耦不是纸上谈兵，它在真实业务场景中直接转化为可量化的工程优势。我结合过去三年做过的五个不同项目，总结出最突出的四点：

1. 可调试性（Debuggability）跃升一个数量级：在传统 CoT 中，如果答案错了，你得回溯整段生成文本，像大海捞针一样找哪一句逻辑出了问题。而在 ReWoo 中，错误定位是线性的。假设最终答案是“库存充足”，但实际缺货。你只需检查 Plan 里是否有check_inventory这个 step；如果有，就看对应 Worker 的result返回的stock_level是多少；如果result是正确的，那问题一定出在 Final Answer Generator 的总结逻辑上。我们有个金融问答项目，上线初期幻觉率偏高，用 ReWoo 架构后，一周内就定位到是 Reasoner 在解析“年化收益率”时，把APY和EAR混淆了，导致 Plan 里调用了错误的计算 Worker。修复 Reasoner 的 few-shot 示例后，问题立刻解决。这种“分段归因”的能力，在端到端模型里是梦寐以求的。

2. 工具调用效率与成本显著优化：Reasoner 的核心能力是“规划”，而不是“执行”。这意味着它可以学习到哪些信息是冗余的、哪些 API 是可以跳过的。例如，在一个多步骤的旅行规划中，Reasoner 可能先调用一个轻量级的get_city_infoWorker 获取城市基本信息，如果发现该城市没有机场，它就不会再规划book_flight这个昂贵的 step。我们在一个酒店预订系统里观察到，ReWoo 架构下，平均每次查询调用的外部 API 次数比传统 Agent 方案减少了37%，直接降低了35% 的云服务调用成本。Reasoner 学会了“先探路，再行动”，而不是“边走边问，问完再走”。

3. 领域知识注入更安全、更可控：把领域知识硬塞进 LLM 的 prompt 或微调数据里，风险很高——模型可能记错、混淆，或者在无关场景下错误复用。ReWoo 把知识固化在 Worker 里。比如，一个法律咨询系统的calculate_compensationWorker，其内部逻辑是严格遵循最新《民法典》司法解释编写的 Python 函数。无论 Reasoner 如何规划，它调用的永远是这个经过律师审核的、确定性的计算逻辑。这比让 LLM “记住”法条并现场推演，可靠度高出几个数量级。我们曾用一个简单的if-elif-else的 Worker 替代了原先需要 200 行 prompt 工程才能勉强稳定的税务计算逻辑，准确率从 82% 直接拉到 99.9%。

4. 系统韧性（Resilience）与降级策略天然支持：当某个 Worker 临时不可用（比如天气 API 限流），Orchestrator 可以优雅降级。它可以在 Plan 执行失败时，记录错误，然后让 Final Answer Generator 基于已有的成功result（比如已查到的城市信息）生成一个“部分答案”，并诚实地告知用户“天气信息暂不可用”。这种“尽力而为”的策略，在端到端模型里几乎无法实现，因为整个生成过程是原子性的，一卡全卡。在一次大型展会期间，我们的现场导览系统就依赖此特性：当室内定位 API 延迟飙升时，系统自动降级为仅提供基于地图坐标的静态路线指引，用户体验几乎没有断层。

3. 核心实现细节与实操要点

3.1 Plan 格式的设计哲学：从“自由发挥”到“强约束接口”

Plan 是 ReWoo 架构的“神经中枢”，它的设计质量直接决定了整个系统的健壮性和可扩展性。很多人初学 ReWoo，会犯一个典型错误：让 Reasoner 输出过于自由的 JSON，比如：

// ❌ 错误示范：字段名随意，类型模糊 {"action": "get_weather", "location": "Shanghai", "when": "next_day"}

这看似没问题，但埋下了巨大隐患。action字段名和 Worker 注册名不一致怎么办？when字段的值"next_day"是字符串，Worker 里还得额外解析，极易出错。ReWoo 的 Plan 设计，核心是“强契约、弱语义”——即 Plan 的结构（字段名、嵌套层级、数据类型）必须绝对严格，而对字段值的语义要求则可以宽松（由 Worker 内部处理）。官方推荐的 Plan 格式是：

[ { "type": "worker_name", // 必须与 Worker 注册名完全一致，字符串 "args": { // 必须是字典，键名必须与 Worker 函数签名的参数名一致 "param1": "value1", "param2": 42, "param3": ["a", "b"] } }, ... ]

这个设计的精妙之处在于：它把 Reasoner 的“创造性”限制在了type和args的选择上，而把所有“解析性”工作都移交给了 Worker。Reasoner 只需学会“调哪个工具、传什么参数”，不用学会“怎么解析时间字符串”。这极大降低了 Reasoner 的训练/提示难度。我在一个医疗问答项目里，让 Reasoner 学习调用lookup_drug_interaction这个 Worker，只需要给它 5 个高质量的 few-shot 示例，它就能稳定输出{"type": "lookup_drug_interaction", "args": {"drug_a": "阿司匹林", "drug_b": "华法林"}}，而不用管drug_a和drug_b后面是不是要加“片”、“胶囊”等单位——那是 Worker 的事。实操心得：Plan 的type名称务必采用 snake_case 并带业务前缀，比如med_lookup_drug_interaction、fin_calculate_apr，避免不同领域 Worker 名称冲突。args字典的 key 名，必须和 Worker 函数的参数名（def worker(param1, param2):）一字不差，这是用 Python 的inspect.signature动态校验的基础。

3.2 Reasoner 的构建：不是越大越好，而是“够用就好”

很多人以为 ReWoo 的 Reasoner 必须用最强的 LLM，这是误区。Reasoner 的核心任务是“规划”（Planning），不是“回答”（Answering）。它不需要百科全书式的知识，也不需要超强的文本生成能力，它需要的是对任务分解、工具选择、参数提取的精准把握。因此，一个经过良好提示（Prompt）设计的中等尺寸模型，往往比一个“啥都会但啥都不精”的超大模型更合适。我的经验是：Reasoner 的选型优先级是：规划能力 > 语言流畅度 > 知识广度。具体操作上，我通常采用三步法：

1. Few-shot Prompt Engineering：这是最经济高效的方法。准备 8-12 个高质量的示例，覆盖你业务中的主要任务类型。每个示例包含：input_query、expected_plan（严格按照上述强契约格式）、以及一个简短的reasoning_note（说明为什么这样规划）。例如：

   Input: "帮我查一下用户ID为U123456的最近三笔订单状态。" Reasoning Note: "需要调用订单查询Worker，并指定用户ID和返回数量。" Output: [{"type": "query_user_orders", "args": {"user_id": "U123456", "limit": 3}}]

   关键是reasoning_note，它教会模型“思考的模式”，而不是“答案的内容”。

2. 轻量微调（Fine-tuning）：当 few-shot 效果遇到瓶颈，比如在特定领域术语上总是出错，我会对一个 3B-7B 的开源模型（如 Phi-3-mini 或 Qwen2.5-1.5B）进行 LoRA 微调。数据集就是上面那些 few-shot 示例的 100 倍。微调的目标 Loss 不是语言建模 Loss，而是Plan 结构的语法正确性 Loss。我自定义了一个损失函数：如果生成的 JSON 无法被json.loads()解析，或者type字段值不在预定义的 Worker 名单里，就给予高额惩罚。这样，模型学到的首要技能是“输出合法的 Plan”，其次才是“输出正确的 Plan”。

3. 模型蒸馏（Distillation）：对于成本极度敏感的场景（如移动端），我会用一个大模型（Teacher）为大量 query 生成 Plan，然后用这些 Plan 作为监督信号，去训练一个更小的模型（Student）。这个过程不蒸馏“答案”，只蒸馏“规划”。我试过用 Qwen2.5-7B 当 Teacher，蒸馏出一个 1.5B 的 Student，其 Plan 生成准确率只比 Teacher 低 2.3%，但推理速度提升了 4.8 倍，内存占用降低了 76%。这证明了 Reasoner 的“可压缩性”。

提示：永远不要试图让 Reasoner 去“理解”Worker 的内部逻辑。它的任务清单里只有三项：1. 看懂用户要什么；2. 从已知的 Worker 列表里挑一个或多个；3. 从用户 query 里精准抽取出args。把“理解”这件事，留给 Worker 和 Final Answer Generator。

3.3 Worker 的开发规范：确定性、幂等性与错误处理

Worker 是 ReWoo 架构的“肌肉”，它的质量直接决定了整个系统的下限。一个糟糕的 Worker，会让再好的 Reasoner 也无能为力。我总结了一套严格的 Worker 开发规范，所有团队成员都必须遵守：

• 确定性（Determinism）：给定完全相同的args输入，Worker 必须返回完全相同的result输出（忽略时间戳等无关字段）。这意味着 Worker 内部不能有随机数、不能依赖未声明的全局状态、不能调用非幂等的外部 API（如 POST 创建资源）。例如，一个generate_reportWorker，如果其内部逻辑包含了random.shuffle(data)，就必须移除或固定 seed。我们曾在一个报表系统里，因为一个 Worker 里用了time.time()作为文件名的一部分，导致每次运行 Plan 都产生不同的result，最终让 Final Answer Generator 的输出变得不可预测。

• 幂等性（Idempotency）：对于有副作用的 Worker（如send_email、update_database），必须保证多次调用相同args不会产生重复副作用。这通常通过在 Worker 内部加入幂等 Key（Idempotency Key）来实现。例如，send_emailWorker 的args中必须包含一个idempotency_key字段，Worker 在执行发送前，先查询数据库确认该 Key 是否已存在，存在则直接返回上次的结果，不存在才真正发送并记录 Key。这是生产环境的铁律。

• 错误处理（Error Handling）：Worker 绝不能让异常（Exception）向上抛出到 Orchestrator。它必须捕获所有可能的异常，并返回一个结构化的错误result。标准格式是：

  {"error": true, "message": "API timeout", "code": "TIMEOUT_504", "retryable": true}

  retryable字段至关重要，它告诉 Orchestrator：这个错误是暂时的（如网络超时），可以重试；还是永久的（如参数错误），应该立即停止并报告。我在一个支付系统里，就靠retryable: false这个字段，成功拦截了大量因用户输错银行卡号导致的无效重试请求，将支付网关的无效调用量降低了 92%。

• 输入校验（Input Validation）：Worker 必须在函数入口处，对args进行严格校验。使用pydantic库定义ArgsSchema是最佳实践。例如：

  from pydantic import BaseModel, Field class WeatherArgs(BaseModel): city: str = Field(..., min_length=1, max_length=50) date: str = Field(..., pattern=r"^(today|tomorrow|\d{4}-\d{2}-\d{2})$") def weather_api(args_dict): try: args = WeatherArgs(**args_dict) # 自动校验并转换 except ValidationError as e: return {"error": true, "message": str(e), "code": "VALIDATION_ERROR", "retryable": false} # ... 执行业务逻辑

  这种防御式编程，让 Worker 成为系统中最可靠的“守门人”。

3.4 Orchestrator 的核心逻辑与 Final Answer Generator 的提示设计

Orchestrator 看似是“胶水代码”，但其逻辑设计的好坏，直接影响最终答案的质量和系统的鲁棒性。它的核心循环非常清晰，但每一步都有讲究：

1. Plan 解析与校验：Orchestrator 收到 Reasoner 的 Plan 后，第一步不是执行，而是校验。它会检查：

   • Plan 是否为合法 JSON 数组；
   • 每个 step 的type是否在预注册的 Worker 字典中；
   • 每个 step 的args是否为字典；
   • （可选）args的 key 是否都在 Worker 的预期参数列表中（可通过inspect.signature获取）。 如果校验失败，Orchestrator 应该立即返回一个清晰的错误，而不是让 Reasoner “猜错”。

2. Worker 调度与上下文管理：Orchestrator 按 Plan 顺序执行 Worker。关键点在于context的构建。context是一个字典，其 key 是 Plan 中 step 的索引（从 0 开始），value 是该 step 的result。例如：

   context = { 0: {"temperature": 25, "rain_probability": 0.7}, 1: {"should_carry_umbrella": True} }

   这个设计确保了 Final Answer Generator 能精确知道哪个result对应哪个 step，避免了信息错位。我见过有人把所有result拼成一个长字符串喂给 Final Answer Generator，结果模型把step 0的温度和step 1的决策逻辑搞混了，导致答案荒谬。

3. Final Answer Generator 的提示（Prompt）设计：这是整个链条的“临门一脚”，也是最容易被忽视的一环。它的 Prompt 必须做到三点：

   • 明确指令（Instruction）：开宗明义告诉模型，它的任务是“基于以下事实性信息，用简洁、专业的中文，直接回答用户问题。不要复述推理过程，不要添加任何推测。”
   • 结构化输入（Structured Input）：把query、plan和context用清晰的分隔符组织起来。我习惯用--- QUERY ---、--- PLAN ---、--- CONTEXT ---这样的标记。例如：

     --- QUERY --- 上海明天需要带伞吗？ --- PLAN --- [{"type": "weather_api", "args": {"city": "上海", "date": "tomorrow"}}, {"type": "decision_logic", "args": {"condition": "rain_probability > 0.5"}}] --- CONTEXT --- Step 0 Result: {"temperature": 25, "rain_probability": 0.7, "condition": "cloudy"} Step 1 Result: {"should_carry_umbrella": true}

   • 输出格式约束（Output Format Constraint）：强制要求模型输出一个 JSON 对象，包含answer和confidence两个字段。例如{"answer": "需要带伞", "confidence": 0.95}。这不仅方便前端解析，更重要的是，confidence字段可以作为下游服务的路由依据（比如低置信度的答案，自动转人工）。

注意：Final Answer Generator 的模型选择，不必追求最大。一个 7B 的模型，只要 Prompt 设计得当，其总结事实的能力，往往优于一个 70B 的模型在混乱 Prompt 下的表现。关键是“给它清晰的指令和干净的输入”，而不是“给它更大的脑子”。

4. 完整实操流程与关键配置详解

4.1 从零开始搭建一个 ReWoo 系统：以“实时股票分析助手”为例

下面我将手把手带你完成一个真实可用的 ReWoo 系统搭建。目标是：用户输入“帮我分析苹果公司（AAPL）的最新股价和市盈率”，系统能调用股票 API 获取数据，并给出专业分析。整个过程分为五个阶段，我会给出每一阶段的核心代码和关键配置说明。

阶段一：定义 Worker

首先，我们创建一个workers.py文件，定义两个核心 Worker：

# workers.py import requests import json # Worker 1: 获取股票基础信息 def get_stock_info(args): """ Args: symbol (str): 股票代码，如 'AAPL' Returns: dict: 包含 price, pe_ratio, company_name 等字段 """ symbol = args.get("symbol") if not symbol: return {"error": True, "message": "Missing symbol", "code": "MISSING_SYMBOL", "retryable": False} # 模拟调用真实API，此处用免费的 Alpha Vantage 示例 # 实际中请替换为你的认证密钥和 endpoint api_url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={symbol}&apikey=demo" try: response = requests.get(api_url, timeout=10) data = response.json() if "Global Quote" not in data: return {"error": True, "message": "Invalid API response", "code": "API_ERROR", "retryable": True} quote = data["Global Quote"] return { "price": float(quote["5. price"]), "pe_ratio": float(quote["23. PE Ratio"]) if quote["23. PE Ratio"] != "None" else None, "company_name": quote["05. price"] # 实际API中需调整 } except Exception as e: return {"error": True, "message": str(e), "code": "NETWORK_ERROR", "retryable": True} # Worker 2: 生成专业分析 def generate_analysis(args): """ Args: stock_info (dict): 来自 get_stock_info 的 result threshold_pe (float): 市盈率阈值，默认25 Returns: dict: 包含 analysis_text, recommendation 字段 """ stock_info = args.get("stock_info") threshold_pe = args.get("threshold_pe", 25.0) if not stock_info or stock_info.get("error"): return {"error": True, "message": "Stock info is invalid", "code": "INVALID_INPUT", "retryable": False} price = stock_info["price"] pe_ratio = stock_info["pe_ratio"] if pe_ratio is None: analysis = f"苹果公司当前股价为 ${price:.2f}。市盈率数据暂不可用。" recommendation = "建议关注后续财报发布。" elif pe_ratio < threshold_pe * 0.8: analysis = f"苹果公司当前股价为 ${price:.2f}，市盈率为 {pe_ratio:.1f}，低于行业均值，估值相对合理。" recommendation = "具备投资价值，可考虑增持。" else: analysis = f"苹果公司当前股价为 ${price:.2f}，市盈率为 {pe_ratio:.1f}，高于行业均值，估值偏高。" recommendation = "短期风险较高，建议观望。" return { "analysis_text": analysis, "recommendation": recommendation, "valuation_score": 100 - abs(pe_ratio - threshold_pe) / threshold_pe * 100 if pe_ratio else 50 } # 将所有 Worker 注册为一个字典，供 Orchestrator 使用 WORKERS = { "get_stock_info": get_stock_info, "generate_analysis": generate_analysis }

关键配置说明：

• get_stock_info的timeout=10是硬性要求，防止一个慢 API 拖垮整个流程。
• generate_analysis的threshold_pe参数设计为可配置，体现了 ReWoo 的灵活性——你可以为不同客户设置不同的估值标准。
• 所有 Worker 都遵循了前述的“确定性、幂等性、错误处理”规范。

阶段二：构建 Reasoner（使用 Few-shot Prompt）

我们创建reasoner.py，不依赖任何大模型框架，仅用 OpenAI 的 API（或你自己的模型 endpoint）：

# reasoner.py import openai import json # 定义 Few-shot 示例 FEW_SHOT_EXAMPLES = [ { "input": "帮我查一下特斯拉（TSLA）的最新股价。", "output": '[{"type": "get_stock_info", "args": {"symbol": "TSLA"}}]' }, { "input": "分析微软（MSFT）的股价和市盈率，用专业术语。", "output": '[{"type": "get_stock_info", "args": {"symbol": "MSFT"}}, {"type": "generate_analysis", "args": {"threshold_pe": 25.0}}]' } ] def call_reasoner(query: str) -> list: """ 调用 Reasoner，返回 Plan 列表 """ # 构建 Prompt prompt_parts = [ "你是一个专业的股票分析助手的推理规划器。你的任务是根据用户问题，生成一个精确的执行计划（Plan）。Plan 必须是 JSON 数组，每个元素是一个对象，包含 'type'（Worker 名）和 'args'（参数字典）字段。", "可用的 Worker 有：", "- get_stock_info: 用于获取股票基础信息，需要参数 'symbol'（股票代码）。", "- generate_analysis: 用于生成专业分析，需要参数 'stock_info'（来自 get_stock_info 的结果）和可选的 'threshold_pe'（市盈率阈值，默认25.0）。", "请严格遵守以下规则：", "1. Plan 必须是合法的 JSON 数组。", "2. 'type' 字段值必须是上述 Worker 名之一，且大小写完全一致。", "3. 'args' 字典的 key 必须是 Worker 所需的参数名。", "4. 不要输出任何解释性文字，只输出 Plan JSON。", "以下是几个示例：" ] for ex in FEW_SHOT_EXAMPLES: prompt_parts.append(f"用户：{ex['input']}") prompt_parts.append(f"Plan：{ex['output']}") prompt_parts.append(f"用户：{query}") prompt_parts.append("Plan：") full_prompt = "\n".join(prompt_parts) # 调用模型 response = openai.ChatCompletion.create( model="gpt-4o-mini", # 选用性价比高的模型 messages=[{"role": "user", "content": full_prompt}], temperature=0.1, # 降低随机性，提高确定性 max_tokens=512 ) plan_str = response.choices[0].message.content.strip() try: plan = json.loads(plan_str) return plan except json.JSONDecodeError as e: # Plan 解析失败，返回一个兜底 Plan print(f"Reasoner output invalid JSON: {plan_str}. Error: {e}") return [{"type": "get_stock_info", "args": {"symbol": "AAPL"}}] # 测试 if __name__ == "__main__": print(call_reasoner("帮我分析苹果公司（AAPL）的最新股价和市盈率")) # 预期输出: [{"type": "get_stock_info", "args": {"symbol": "AAPL"}}, {"type": "generate_analysis", "args": {"threshold_pe": 25.0}}]

关键配置说明：

• temperature=0.1是 Reasoner 的黄金参数，它足够低以保证输出稳定，又不至于完全僵化。
• Prompt 中的“规则”部分，是 Few-shot 的灵魂。它把隐含的约束显性化，比单纯堆砌示例更有效。
• max_tokens=512是一个安全上限，防止 Reasoner “话痨”输出无关内容。

阶段三：实现 Orchestrator

orchestrator.py是整个流程的指挥中心：

# orchestrator.py import json from workers import WORKERS class ReWooOrchestrator: def __init__(self, workers: dict): self.workers = workers def execute_plan(self, plan: list) -> dict: """ 执行 Plan，返回 context 字典 """ context = {} for i, step in enumerate(plan): step_type = step.get("type") step_args = step.get("args", {}) # 校验 Worker 是否存在 if step_type not in self.workers: context[i] = {"error": True, "message": f"Unknown worker: {step_type}", "code": "WORKER_NOT_FOUND", "retryable": False} continue # 执行 Worker try: result = self.workers[step_type](step_args) context[i] = result except Exception as e: context[i] = {"error": True, "message": f"Worker execution failed: {str(e)}", "code": "WORKER_EXCEPTION", "retryable": False} return context def generate_final_answer(self, query: str, plan: list, context: dict) -> str: """ 调用 Final Answer Generator 这里简化为一个本地函数，实际中可调用大模型API """ # 构建输入 input_str = f"--- QUERY ---\n{query}\n--- PLAN ---\n{json.dumps(plan, ensure_ascii=False)}\n--- CONTEXT ---\n" for idx, result in context.items(): input_str += f"Step {idx} Result: {json.dumps(result, ensure_ascii=False)}\n" # 简化版 Final Answer Generator：直接用规则匹配 # 实际项目中，这里会调用一个专门微调过的模型 final_answer = "" for idx, result in context.items(): if not result.get("error") and "analysis_text" in result: final_answer = result["analysis_text"] + " " + result["recommendation"] break if not final_answer: final_answer = "抱歉，未能获取有效分析结果。" return {"answer": final_answer, "confidence": 0.95} # 使用示例 if __name__ == "__main__": orchestrator = ReWooOrchestrator(WORKERS) # 模拟 Reasoner 的输出 plan = [ {"type": "get_stock_info", "args": {"symbol": "AAPL"}}, {"type": "