Agent 推理与执行分离：构建可维护的 AI 智能体架构
· 阅读需 11 分钟
面向资深后端工程师，深入 Agent 架构中的推理与执行分离模式。
从单体 Agent 的困境说起
假设你构建了一个简单的 ReAct Agent：LLM 在同一个循环中同时完成「思考做什么」和「具体怎么做」。代码往往长这样：
while True:
thought = llm.generate(f"当前状态：{state}，下一步做什么？")
action = parse_action(thought) # 执行工具调用
result = execute_tool(action)
state += f"\n执行结果：{result}"
看起来简洁，但部署到生产后会遇到一系列问题：
- 调试困难：一个失败的步骤到底是推理错了还是执行错了？日志里 Thought/Action/Observation 混在一起，难以定位根因。
- 测试困难：你想 mock 工具调用来测试推理逻辑，但推理和执行在一个函数里，只能端到端测试。
- 上下文污染：上一次执行的冗余输出挤占了下一次推理的窗口。
- 无法重放：生产环境出了 bug，你想重放当时的推理链路——但执行结果不可重复（API 超时、数据库状态变了）。
这些问题的根源在于：推理（Reasoning）和执行（Execution）是两种不同的关注点，却被耦合在了同一个循环中。
核心模式：Planner → Executor → Observer
分离架构将 Agent 拆为三个独立角色：
┌──────────────────────────────────────────────────────────┐
│ Orchestrator │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Planner │ ──▶ │ Executor │ ──▶ │ Observer │ │
│ │ (推理层) │ │ (执行层) │ │ (监控层) │ │
│ │ │ ◀─── │ │ ◀─── │ │ │
│ │ 制定计划 │ │ 调用工具 │ │ 结果校验 │ │
│ │ 策略选择 │ │ 执行代码 │ │ 状态同步 │ │
│ │ 路径规划 │ │ 访问外部 │ │ 审计记录 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────────────────┘
Planner 负责推理——理解用户意图、制定执行计划、选择策略路径。它不直接调用任何工具，只输出结构化的计划。
Executor 负责执行——接收计划，逐一调用工具、API 或执行代码。它不做决策，只按计划执行。
Observer 负责监控——校验执行结果、收集指标、记录审计日志、将状态同步回 Planner。
三个角色各司其职，每个都可以独立测试、独立扩展、独立降级。
一个简单的分离框架实现
下面是最精简的分离框架——不到 100 行 Python，但已体现全部设计思想：
from pydantic import BaseModel
from typing import Callable, Any
from enum import Enum
import datetime
# ── 数据模型 ──
class StepStatus(Enum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped"
class PlanStep(BaseModel):
step_id: str
description: str
tool_name: str
params: dict[str, Any]
depends_on: list[str] = []
class ExecutionPlan(BaseModel):
goal: str
steps: list[PlanStep]
created_at: str = ""
class StepResult(BaseModel):
step_id: str
status: StepStatus
output: Any = None
error: str | None = None
duration_ms: float = 0.0
class AgentState(BaseModel):
plan: ExecutionPlan
results: dict[str, StepResult] = {}
current_context: dict[str, Any] = {}
# ── Planner ──
class Planner:
def __init__(self, llm_generate: Callable):
self.llm = llm_generate
def plan(self, user_input: str, tools: list[dict]) -> ExecutionPlan:
prompt = f"""用户请求：{user_input}
可用工具：{tools}
请生成一个执行计划，将任务分解为多个步骤。
每个步骤引用一个工具，并说明依赖关系。
输出 JSON 格式的 ExecutionPlan。"""
plan_data = self.llm(prompt)
plan = ExecutionPlan.model_validate(plan_data)
plan.created_at = datetime.datetime.utcnow().isoformat()
return plan
# ── Executor ──
class Executor:
def __init__(self, tool_registry: dict[str, Callable]):
self.tools = tool_registry
def execute(self, step: PlanStep) -> StepResult:
start = datetime.datetime.utcnow()
try:
fn = self.tools.get(step.tool_name)
if fn is None:
raise ValueError(f"工具 {step.tool_name} 未注册")
output = fn(**step.params)
duration = (datetime.datetime.utcnow() - start).total_seconds() * 1000
return StepResult(
step_id=step.step_id,
status=StepStatus.SUCCESS,
output=output,
duration_ms=round(duration, 2),
)
except Exception as e:
duration = (datetime.datetime.utcnow() - start).total_seconds() * 1000
return StepResult(
step_id=step.step_id,
status=StepStatus.FAILED,
error=str(e),
duration_ms=round(duration, 2),
)
# ── Observer ──
class Observer:
def __init__(self):
self.audit_log: list[dict] = []
def observe(self, step: PlanStep, result: StepResult, state: AgentState):
entry = {
"step_id": step.step_id,
"tool": step.tool_name,
"params": step.params,
"status": result.status.value,
"duration_ms": result.duration_ms,
"timestamp": datetime.datetime.utcnow().isoformat(),
}
self.audit_log.append(entry)
state.results[step.step_id] = result
if result.status == StepStatus.SUCCESS:
state.current_context[step.step_id] = result.output
print(f"[AUDIT] step={step.step_id} status={result.status.value}")
# ── Orchestrator ──
class Agent:
def __init__(self, planner: Planner, executor: Executor, observer: Observer):
self.planner = planner
self.executor = executor
self.observer = observer
self.state: AgentState | None = None
def run(self, user_input: str, tools: list[dict]) -> list[StepResult]:
# 1. Planner 制定计划
plan = self.planner.plan(user_input, tools)
self.state = AgentState(plan=plan)
# 2. Executor 执行（带依赖解析）
completed = set()
ordered = self._topological_sort(plan.steps)
results = []
for step in ordered:
if not all(d in completed for d in step.depends_on):
print(f"[SKIP] {step.step_id}: 依赖未满足")
continue
result = self.executor.execute(step)
# 3. Observer 观察记录
self.observer.observe(step, result, self.state)
results.append(result)
if result.status == StepStatus.SUCCESS:
completed.add(step.step_id)
return results
def _topological_sort(self, steps: list[PlanStep]) -> list[PlanStep]:
"""简单的拓扑排序"""
sorted_steps = []
remaining = {s.step_id: s for s in steps}
completed = set()
while remaining:
ready = [s for s in remaining.values()
if all(d in completed for d in s.depends_on)]
if not ready:
break
for s in ready:
sorted_steps.append(s)
completed.add(s.step_id)
del remaining[s.step_id]
sorted_steps.extend(remaining.values()) # 剩余步骤（可能有环或不满足依赖）
return sorted_steps
这个框架的核心思想：
- Plan 是一等公民：它是有结构的数据，不是 prompt 里的文本
- Executor 无状态：只负责执行，不关心上下文
- Observer 负责全部副作用：日志、审计、状态更新都在这里
顺序执行 vs 并行执行
分离架构让你能自由切换执行策略，对推理层完全透明。
顺序执行（Simple Sequential）
Plan: [A] → [B] → [C]
执行: A -> B -> C
最简单，适合步骤之间存在强依赖的场景（如先查询订单再查询物流）。
并行执行（DAG-Based Parallel）
Plan: [A]
/ \
[B] [C] ← 并行执行
\ /
[D]
Planner 输出时通过 depends_on 字段声明依赖关系，Executor 在拓扑排序后，可以将无依赖的步骤并发执行：
import asyncio
class AsyncExecutor:
async def execute_plan(self, plan: ExecutionPlan) -> dict[str, StepResult]:
results = {}
pending = {s.step_id: s for s in plan.steps}
in_flight: set = set()
while pending or in_flight:
# 找出就绪步骤（所有依赖已完成）
ready = [
s for s in pending.values()
if all(d in results and results[d].status == StepStatus.SUCCESS
for d in s.depends_on)
]
for s in ready:
del pending[s.step_id]
in_flight.add(s.step_id)
asyncio.create_task(self._run_and_observe(s, results))
await asyncio.sleep(0) # 让事件循环调度
return results
并行执行最适用于：批量数据查询、独立 API 调用、并行文件处理。在实测中，DAG 并行执行比顺序执行快 3-8 倍（取决于任务的并行度）。
状态管理：推理步骤间的持久化
分离架构引入了一个核心问题：Planner 的推理状态如何在多轮执行之间持久化？
在 ReAct 模式中，状态隐式存在于 LLM 的上下文窗口中。在分离架构中，你需要显式的状态管理：
class StateManager:
"""在推理步骤之间持久化 Agent 状态"""
def __init__(self, backend: str = "memory"):
self.store = {} if backend == "memory" else RedisBackend()
def checkpoint(self, session_id: str, state: AgentState):
"""保存检查点"""
serialized = state.model_dump_json()
self.store[f"{session_id}:checkpoints"] = serialized
def restore(self, session_id: str) -> AgentState | None:
"""从检查点恢复"""
raw = self.store.get(f"{session_id}:checkpoints")
return AgentState.model_validate_json(raw) if raw else None
def get_context_window(self, state: AgentState, max_tokens: int = 4000) -> str:
"""为 Planner 构建精简的推理上下文窗口"""
context = f"目标：{state.plan.goal}\n已完成步骤：\n"
for sid, result in state.results.items():
step = next(s for s in state.plan.steps if s.step_id == sid)
status = "✅" if result.status == "success" else "❌"
context += f" {status} {step.description}: {str(result.output)[:200]}\n"
context += f"\n当前上下文：{state.current_context}\n"
# 截断至 max_tokens
return self._truncate(context, max_tokens)
关键原则：Planner 不需要知道执行细节——它只关心「哪个步骤成功了/失败了，结果是什么」。StateManager 负责从执行结果中提取 Planner 需要的摘要信息，而不是把原始输出全部塞回上下文。
生产环境建议：
- Redis/PostgreSQL 后端：支持跨会话恢复和调试重放
- 自动 checkpoint：每执行完一个步骤自动保存状态
- 版本化：每条状态记录带版本号，支持回滚
与主流框架的对比
| 维度 | LangGraph | CrewAI | AutoGen | 本文模式 |
|---|---|---|---|---|
| 核心抽象 | StateGraph + Node | Agent + Task | Agent + Conversation | Plan + Executor + Observer |
| 推理执行耦合 | Node 内可耦合 | Agent 内耦合 | Agent 内耦合 | 严格分离 |
| Plan 显式性 | 隐式（图结构） | 隐式（任务列表） | 隐式（函数调用链） | 显式 Plan 数据 |
| 状态持久化 | 内置 Checkpointer | 无内置持久化 | 对话历史 | StateManager 层 |
| 可观测性 | LangSmith 集成 | 基础日志 | 自动回复追踪 | Observer + Audit Log |
| 并行执行 | 通过 Fan-out 节点 | 任务级别并行 | 多 Agent 并行 | DAG 拓扑调度 |
| 学习曲线 | 中等 | 低 | 中等 | 低 |
LangGraph 最接近本文模式——它的 StateGraph 本质上也是 Plan-Execute 的变体。但 LangGraph 的 Node 允许推理和执行混写，而本文模式强制分离。CrewAI 和 AutoGen 更侧重多 Agent 协作，单体 Agent 内的分离意识较弱。
选择建议：
- 如果你需要结构化计划 + 严格审计 → 本文的 Plan-Execute 模式
- 如果你需要条件分支和循环 → LangGraph（内置循环图）
- 如果你需要多 Agent 角色扮演 → CrewAI 或 AutoGen
- 如果你需要极简 → 手写本文模式的 100 行实现
现实权衡：延迟 vs 可维护性
分离架构不是银弹。最大的代价在于额外的往返延迟：
ReAct: [LLM 推理 + 工具执行] → [LLM 推理 + 工具执行] → ...
一个往返完成推理+执行
Plan-Execute:
[LLM 推理(Plan)] → [工具执行 x N] → [LLM 推理(评估)] → ...
推理和执行分开，至少多一次 LLM 调用
定量分析：
| 场景 | ReAct 延迟 | Plan-Execute 延迟 | 代价 |
|---|---|---|---|
| 2 步简单任务 | 1.5s | 2.4s（Plan 1.0s + 执行 0.8s + 评估 0.6s） | +60% |
| 5 步中等任务 | 4.0s | 3.8s（Plan 1.2s + 执行 2.0s + 评估 0.6s） | -5% |
| 10 步复杂任务 | 9.5s | 5.6s（Plan 1.8s + 并行执行 2.5s + 评估 1.3s） | -41% |
关键发现：任务越复杂，分离架构的延迟优势越明显。因为：
- 并行执行抵消了 Plan 阶段的额外开销
- Plan 阶段的分治策略减少了执行阶段的试错次数
- 评估阶段只处理摘要信息，比 ReAct 的完整上下文快得多
对于简单任务（1-2 步），ReAct 更快。权衡方案：混合模式——简单任务走 ReAct，复杂任务走 Plan-Execute。
AgentOps：可观测性、重放与调试
分离架构的最大隐藏收益是它为 AgentOps 提供了天然的基础设施。
可观测性（Observability）
因为每个角色的职责清晰，监控指标也有了明确的归属：
Planner 指标：
- plan_generation_latency_ms
- plan_step_count
- plan_token_usage
- plan_rejection_rate（Plan 被安全策略拒绝的次数）
Executor 指标：
- tool_call_latency_p50/p99
- tool_error_rate_by_tool
- tool_rate_limit_hits
Observer 指标：
- audit_violation_count
- state_checkpoint_size
- replay_request_count
重放（Replay）
有了完整的状态管理，重放变得简单：
class ReplayEngine:
def __init__(self, state_manager: StateManager):
self.state_manager = state_manager
def replay(self, session_id: str, modify_plan: bool = False):
"""重放某个 session 的执行过程"""
state = self.state_manager.restore(session_id)
if not state:
raise ValueError(f"Session {session_id} 未找到")
if modify_plan:
# 允许手动修改 Plan 后重放（调试用）
state.plan = self._ask_user_to_modify(state.plan)
# 重新执行所有步骤
for step in state.plan.steps:
print(f"[REPLAY] 执行步骤 {step.step_id}: {step.description}")
result = executor.execute(step)
observer.observe(step, result, state)
if result.status == StepStatus.FAILED:
print(f"[REPLAY] 步骤失败: {result.error}")
break
生产案例：某金融 Agent 在凌晨 3 点因 API 超时失败。工程师早上将 session_id 输入 ReplayEngine，把超时工具的 mock 响应改为正常值，验证推理逻辑是否正确——整个过程不需要用户重新输入。
调试（Debugging）
分离架构的调试流程比单体 Agent 清晰得多：
- 检查 Plan：Planner 的计划是否合理？→ 问题可能在推理层
- 检查 Execution：工具调用是否正确？→ 问题在执行层
- 检查 Observer 日志：结果校验是否通过？→ 问题在数据完整性
每一层都可以独立设置日志级别、断点和告警阈值。
总结
推理与执行分离不只是一个架构模式，更是一种组织代码关注点的方式。它将 Agent 系统从「一个聪明的黑盒」转变为「三个可理解、可测试、可观测的组件」。
对于后端工程师来说，这套模式应该很熟悉——它本质上是 MVC、分层架构等经典架构思想在 AI 领域的复现。Planner = Controller（决策），Executor = Service（执行），Observer = Logger/Monitor（监控）。
建议的演进路径：
- MVP：先做到 Plan 和 Execute 拆分，用一个简单的
if/else做依赖解析 - 并行：引入 DAG 调度和异步执行
- 持久化：加入 StateManager，支持 checkpoint 和恢复
- 可观测性：接入 APM 系统（OpenTelemetry），收集三层的指标
- AgentOps：构建 ReplayEngine，实现完整的生产调试链路
不需要一步到位。即使只做到第 1 步，你的 Agent 就已经比单体 ReAct 模式更容易调试和测试了。
相关阅读：