肖哥弹架构 跟大家“弹弹” 大模型Agent 设计与实战应用，需要代码关注

欢迎 关注，点赞，留言。

关注公号Solomon肖哥弹架构获取更多精彩内容

历史热点文章

• 解锁大语言模型参数：零基础掌握大型语言模型参数奥秘与实践指南
• 6张图掌握提示词工程师工作范围与工作技巧(提示词原理篇)

• 一个项目代码讲清楚DO/PO/BO/AO/E/DTO/DAO/ POJO/VO
• 写代码总被Dis：5个项目案例带你掌握SOLID技巧,代码有架构风格
• 里氏替换原则在金融交易系统中的实践，再不懂你咬我

⚠️ 原创不易 搬运必究

PlanAndSolve是面向复杂业务流程的智能规划执行框架，通过"规划-执行-重规划"的闭环机制解决传统AI系统在复杂任务中的不可解释、难恢复、缺乏自适应等痛点。本文从技术背景、核心流程到落地实践，完整解析其显式规划、动态调整、可观测追踪三大核心能力，结合电商下单等真实场景展示从用户请求到最终响应的完整数据流。无论你是构建智能客服、电商系统还是企业自动化流程，都能掌握这套可解释、可恢复、可扩展的智能体工作流解决方案。

一、PlanAndSolve技术背景

1. 没有 PlanAndSolve 的常见问题

• 即问即答/一步到位：复杂任务被当成单步调用，容易遗漏依赖（如地址解析→库存→价格→下单→支付的顺序约束）。
• 不可解释：缺少可视化的“计划/执行轨迹”，难以审计、复盘和排查故障。
• 难以自适应：外部环境易变（库存、权限、网络抖动），缺少“执行后复盘→调整计划”的闭环。
• 难以恢复：中途失败后，只能整条链路重跑，成本高、可用性差。
• 生态割裂：规划、执行、并行编排、权限、SSE 流、追踪常常分散实现，维护困难。

2. PlanAndSolve 提供的能力

• 显式规划：把用户目标拆解为结构化 Plan（steps），让“要做什么、先后顺序”清晰可见。
• 可解释执行：每一步输入/输出都可记录到追踪系统（ES），便于回放与定位。
• 动态重规划：执行后基于上下文复盘，按需生成 Action（Response/新 Plan），适配不确定环境。
• 可控成本：max_replan_rounds 控制收敛上限，避免无限纠结；必要时用备用 LLM 收尾。
• 生态整合：天然与 OxyRequest.call 的权限校验、超时、并行、SSE 流、断点续跑等机制配合。

3. 典型适用场景

• 多步骤强依赖的业务流程：电商下单/售后、报表生成、流水线式数据处理、检索→抽取→写库→汇总。
• 外部依赖复杂/易变：库存、价格、第三方接口限流或间歇性失败、权限门槛变化。
• 有审计和可观测性要求：需要“能还原经过、能解释决策、能快速定位问题”。
• 需要人机协作：SSE 流式可见进度；支持用户取消、从中间节点恢复重跑。

4. 不太需要使用的场景

• 单步问答、单工具查询这类“简单、确定、无依赖”的任务，用直呼 LLM/工具更省时。

5. 与常见“Prompt Chain/Toolformer 风格”的差异

• PlanAndSolve 的 Plan/Action/Response 明确而结构化，循环与上限可控；
• 和 OxyGent 生态（权限/并行/追踪/SSE/断点）天然对齐，工程落地方便、可维护性更高。

6. ROI（效果与价值）

• 提升成功率：通过拆解、复盘与重规划，降低“首发失败”概率。
• 降低恢复成本：失败后按节点恢复重跑，而非整条链路重做。
• 提升可维护性：有计划与轨迹，问题可定位、可复用、可优化。

二、PlanAndSolve流程

1. 基础规划与执行（无重新规划）

2. 启用重新规划的动态调整

3. 使用预设计划步骤

三、 PlanAndSolve落地

1. 公开 API 与关键配置

• 公开入口：PlanAndSolve._execute(oxy_request: OxyRequest) -> OxyResponse
• 关键字段：
  • planner_agent_name/executor_agent_name/replanner_agent_name(隐含)：角色绑定
  • pre_plan_steps: 预设计划（跳过规划）
  • enable_replanner: 是否启用重规划
  • max_replan_rounds: 重规划最多轮数
  • llm_model: 收尾/兜底模型
• 解析器扩展点：
  • pydantic_parser_planner/pydantic_parser_replanner
  • func_parse_planner_response/func_parse_replanner_response

2. 角色契约（Contract）

• Planner 输入：original_query 或经 PydanticOutputParser.format 的结构化提示
  • 输出：Plan(steps: List[str])
• Executor 输入：task_formatted（包含已完成/当前步骤的系统指令）
  • 输出：任意可串联为 past_steps 的文本或结构化摘要
• Replanner 输入：replan_query（包含目标、原计划、已完成）
  • 输出：Action(Response | Plan)；返回 Response 即终止。

3. 错误与超时策略（与 OxyRequest.call 协作）

• 每次子调用由 OxyRequest.call 包裹：
  • 超时：asyncio.wait_for(..., timeout=oxy.timeout) 返回 FAILED: 超时
  • 权限：is_permission_required 且调用方不在白名单 → SKIPPED
  • 异常：捕获后返回 FAILED 并写明错误；可选 retry_execute 策略

4. 可观测性映射（落 ES / 消息流）

• 建议记录：request_id/group_id/current_trace_id/from_trace_id/node_id、caller/callee、arguments摘要、output摘要、state
• SSE：通过 send_message 推送阶段事件（与 Web 前端对齐）
• 断点恢复：结合 {app}_node 回填 restart_node_order/reference_trace_id

5. 性能与行为开关

• 规划与重规划提示长度控制：对 plan_str/past_steps 做摘要，避免超长上下文
• 限流/熔断：基于 caller/callee 在外层做策略（参考 OxyRequest 文档中的限流示例）
• 并行：可在 Executor 内部 fan-out；PlanAndSolve 自身按步骤顺序推进

6. 权限与安全

• 通过 OxyRequest.call 的权限模型保护受限工具
• 避免把敏感数据放入 past_steps 的可回显内容；对日志脱敏

7. 端到端最小示例（伪代码）

plan_and_solve = PlanAndSolve(
    planner_agent_name="order_planner",
    executor_agent_name="order_executor",
    enable_replanner=True,
    max_replan_rounds=10,
)

oxy_response = await plan_and_solve.execute(
    OxyRequest(caller="user", callee="plan_and_solve", arguments={"query": "下单 iPhone"})
)

8. 测试清单（建议）

• 规划正常产出 Plan；解析器能解析多语言/边界格式
• 执行步骤串行推进，past_steps 累积正确
• 启用重规划时：响应/新计划两分支均覆盖；达到上限时兜底总结
• 超时/异常/无权限时：返回态与日志一致；断点恢复可用
• SSE 流/取消任务/多会话并发：不串线、不泄漏

9. 常见陷阱

• replanner_agent_name 未注册或无权限 → 永远无法重规划
• past_steps 过长导致提示爆炸 → 需摘要化
• pre_plan_steps 与 Planner 产出混用 → 明确优先级（当前实现：有 pre_plan_steps 则跳过规划）

10. 术语对照

• Plan：步骤清单（结构化）
• Action：Plan 或 Response
• Response：直接面向用户的答案（终止）
• past_steps：累计已完成步骤与结果的文本摘要

四、PlanAndSolve 业务数据快照：从输入到最终结果（含重规划对比）

本文档以“电商下单 iPhone 15 Pro 256G 黑色”场景，展示 PlanAndSolve 流程在不同阶段的业务数据形态，以及启用/不启用重规划时的差异，帮助直观理解其处理思路与步骤。

0) 用户输入 → 进入流程

{
  "oxy_request": {
    "caller": "user",
    "callee": "plan_and_solve",
    "arguments": {
      "query": "下单 1 台 iPhone 15 Pro 256G 黑色，寄到上海浦东，顺丰。"
    }
  }
}

---

1) 规划阶段 → 产出 Plan.steps

{
  "plan": {
    "steps": [
      "解析用户意图与收件信息",
      "检查库存与可配送性",
      "计算价格与运费",
      "创建订单",
      "发起支付并确认"
    ]
  }
}

{
  "plan_str": "1. 解析用户意图与收件信息\n2. 检查库存与可配送性\n3. 计算价格与运费\n4. 创建订单\n5. 发起支付并确认",
  "past_steps": ""
}

---

2) 执行第1步 → Executor 入参与输出

{
  "executor_request": {
    "callee": "executor_agent",
    "arguments": {
      "query": "我们已完成以下步骤: \n当前要执行的步骤是:解析用户意图与收件信息\n您应该只执行当前步骤..."
    }
  }
}

{
  "executor_response": {
    "output": "已解析：商品=iPhone15Pro-256G-黑色，数量=1；地址=上海市浦东新区；快递=SF"
  },
  "state_update": {
    "past_steps": "任务:解析用户意图与收件信息, 执行任务结果:已解析：商品=...快递=SF",
    "plan_steps_remaining": [
      "检查库存与可配送性",
      "计算价格与运费",
      "创建订单",
      "发起支付并确认"
    ]
  }
}

---

3) 执行第2步 → 库存检查

{
  "executor_request": {
    "callee": "executor_agent",
    "arguments": {
      "query": "我们已完成以下步骤: 任务:解析...SF\n当前要执行的步骤是:检查库存与可配送性\n您应该只执行当前步骤..."
    }
  },
  "executor_response": {
    "output": "库存可售，仓库=SH_MAIN，可配送至浦东，预计2日达"
  },
  "state_update": {
    "past_steps": "任务:解析...SF\n任务:检查库存与可配送性, 执行任务结果:库存可售...2日达",
    "plan_steps_remaining": [
      "计算价格与运费",
      "创建订单",
      "发起支付并确认"
    ]
  }
}

---

4) 执行第3步 → 价格与运费

{
  "executor_request": {
    "callee": "executor_agent",
    "arguments": {
      "query": "我们已完成以下步骤: 任务:解析...; 任务:检查库存...\n当前要执行的步骤是:计算价格与运费"
    }
  },
  "executor_response": {
    "output": "商品价=8999.00，运费=12.00（SF），总计=9011.00"
  },
  "state_update": {
    "past_steps": "任务:解析...;\n任务:检查库存...;\n任务:计算价格与运费, 执行任务结果:总计=9011.00",
    "plan_steps_remaining": [
      "创建订单",
      "发起支付并确认"
    ]
  }
}

---

5) 执行第4步 → 创建订单

{
  "executor_request": {
    "callee": "executor_agent",
    "arguments": {
      "query": "我们已完成以下步骤: ...总计=9011.00\n当前要执行的步骤是:创建订单"
    }
  },
  "executor_response": {
    "output": "订单已创建：ORD20251003000123"
  },
  "state_update": {
    "past_steps": "...\n任务:创建订单, 执行任务结果:ORD20251003000123",
    "plan_steps_remaining": [
      "发起支付并确认"
    ]
  }
}

---

6) 执行第5步 → 发起支付并确认 → 最终结果

{
  "executor_request": {
    "callee": "executor_agent",
    "arguments": {
      "query": "我们已完成以下步骤: ...ORD20251003000123\n当前要执行的步骤是:发起支付并确认"
    }
  },
  "executor_response": {
    "output": "支付成功：ALI20251003004567，状态=PAID"
  },
  "final_oxy_response": {
    "state": "COMPLETED",
    "output": {
      "order_id": "ORD20251003000123",
      "status": "PAID",
      "amount": 9011.00,
      "payment_tx": "ALI20251003004567",
      "eta": "2025-10-05",
      "carrier": "SF"
    }
  }
}

五、 启用“重新规划”时的关键差异快照

在每一步执行后，PlanAndSolve 会构造 replan_query，调用 replanner_agent：

• 返回 Action.Response（直接答复用户，流程结束）

{
  "replanner_response": {
    "action": {
      "response": "库存不足，推荐改为 iPhone 15 Pro 128G；是否继续？"
    }
  },
  "final_oxy_response": {
    "state": "COMPLETED",
    "output": "库存不足，推荐改为 iPhone 15 Pro 128G；是否继续？"
  }
}

• 返回 Action.Plan（更新剩余计划，继续执行）

{
  "replanner_response": {
    "action": {
      "steps": [
        "更换为 128G 版本",
        "重新计算价格",
        "创建订单",
        "发起支付并确认"
      ]
    }
  },
  "state_update": {
    "plan_steps": [
      "更换为 128G 版本",
      "重新计算价格",
      "创建订单",
      "发起支付并确认"
    ],
    "past_steps": "保留此前已完成的描述性结果，作为上下文"
  }
}

当达到 max_replan_rounds 上限仍未收敛时，流程将调用备用 LLM 输出总结并结束。

---

1. 如何阅读日志快速定位

• 看 plan.steps：当前系统的执行意图与顺序。
• 看每个 executor_request/response：每步的实际输入/输出。
• 看 past_steps：上下文累计；重规划会参考它。
• 看 replanner_response：是否“改计划”或“直接回答”。
• 看最终 final_oxy_response：业务最终结果。

六、 PlanAndSolve 的思路与步骤

• 核心思想

  • 先“规划”后“执行”：把用户目标拆成有序步骤，再逐步执行与反馈。
  • 可选“动态调整”：每步执行后复盘，必要时重规划，直到完成或达到上限。
  • 面向“可观测”：每一步都是显式的调用、输出与累积上下文，便于调试与回放。

• 参与角色

  • 规划代理 planner：把目标转成步骤列表 Plan.steps。
  • 执行代理 executor：逐步执行当前步骤，返回结果。
  • 重规划代理 replanner（可选）：根据进展返回 Action，可能直接给用户 Response 或更新 Plan。
  • 备用 LLM：流程过长/收尾时输出最终总结。

• 数据与状态

  • original_query：用户意图原文。
  • plan_steps：待执行步骤列表（规划/重规划产出）。
  • past_steps：已完成步骤的叙述性累计（用于上下文与重规划）。
  • plan_str：当前计划的可读化文本。

• 基本流程（未启用重规划）

  1. 初始化上下文（plan_str、past_steps）。
  2. 若无预设 pre_plan_steps：调用 planner 得到 Plan.steps；否则直接用预设。
  3. 循环：
     • 取 plan_steps[0] 作为“当前任务”，构造 task_formatted（含已完成和当前指令）。
     • 调用 executor 执行，得到结果，累计到 past_steps。
     • 移除已执行步骤；若空，结束并返回最后一次结果或总结。
  4. 如全部完成，返回 OxyResponse(COMPLETED)。

• 启用重规划时的分支

  1. 每执行完一个步骤后，构造 replan_query（含用户目标、原计划、已完成内容）。
  2. 调用 replanner：
     • 若返回 Action.Response：直接面向用户回答并结束。
     • 若返回 Action.Plan：覆盖 plan_steps，继续循环。
  3. 若循环超过 max_replan_rounds：用备用 LLM 生成最终总结并返回。

• 停止条件

  • plan_steps 全部执行完毕。
  • 重规划返回 Response（直接可答）。
  • 达到最大重规划轮次（用备用 LLM 收尾）。
  • 异常由外层捕获并反馈为失败响应（OxyRequest.call 有超时与异常处理）。

• 设计动机与建议

  • 可解释性强：规划清晰、执行可回放、每步可观察输入输出。
  • 鲁棒性强：通过重规划适应外部变化（库存/权限/依赖）。
  • 可扩展：替换 planner/executor/replanner 即可迁移不同业务。
  • 建议：强结构化流程可用 pre_plan_steps；重规划场景降低 max_replan_rounds；输出尽量结构化（Pydantic 解析）。