AI Agent 架构解析：构建 Plan-and-Execute 智能体

AI Agent 架构解析：构建 Plan-and-Execute 智能体

引言

随着大语言模型（LLM）能力的不断提升，AI Agent（智能体）在解决复杂任务方面展现出巨大潜力。然而，传统的交互模式在面对多步骤、长链条任务时往往显得力不从心。本文深入探讨一种新兴的架构模式——Plan-and-Execute（规划与执行），并通过 Python 和 LangGraph 框架提供完整的实践指南。

传统 ReAct 架构的困境

ReAct（Reasoning + Acting）架构曾是 AI Agent 的主流范式，它要求模型在每一步行动前都进行推理。尽管灵活，但在实际生产环境中暴露出显著缺陷：

1. 执行效率低下：ReAct 采用单步循环模式，每完成一个动作必须等待 LLM 推理下一步。这种串行处理导致延迟累积，如同在每个路口重新启动车辆，严重拖慢整体流程。
2. 上下文记忆缺失：由于缺乏长期规划能力，Agent 容易在执行中途迷失方向。一旦任务跨度较大，早期信息容易被后续对话淹没，导致关键上下文丢失。
3. 资源消耗高：每一步都需要调用一次 LLM，对于包含数十个步骤的任务，Token 消耗呈线性增长，成本难以控制。

Plan-and-Execute 架构优势

Plan-and-Execute 架构通过分离'思考'与'执行'，从根本上优化了上述问题：

核心优势

• 高效执行：规划器一次性生成完整计划，执行器按序运行，减少了中间推理的等待时间。
• 成本控制：规划阶段可使用较小模型或特定提示词，执行阶段可复用工具，降低对昂贵 LLM 的依赖。
• 质量提升：全局视角的规划能预判潜在冲突，提高任务完成率。

核心组件

该架构主要包含两个关键模块：

1. 规划器（Planner）：负责将用户目标拆解为有序的子任务列表。它需要确保每个步骤所需的信息完备，避免遗漏。
2. 执行器（Executor）：接收具体子任务，调用相应工具（如搜索、代码解释器）获取结果，并反馈给系统。

基于 LangGraph 的实现方案

LangGraph 提供了构建有状态、多节点工作流的强大能力，非常适合实现 Plan-and-Execute 模式。

1. 状态定义

首先定义工作流的状态结构，用于存储输入、计划、历史步骤及最终响应。

from typing import TypedDict, List, Annotated
import operator

class PlanExecute(TypedDict):
    input: str
    plan: List[str]
    past_steps: List[tuple]
    response: str

2. 规划节点

规划节点利用 LLM 的结构化输出功能生成计划。需确保 Prompt 明确约束步骤数量和信息完整性。

from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

class Plan(BaseModel):
    steps: List[str] = Field(description="Step by step plan")

planner_prompt = ChatPromptTemplate.from_messages([
    ("system", """For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. \
Do not add any superfluous steps. The result of the final step should be the final answer. \
Make sure that each step has all the information needed - do not skip steps.""")
])

async def plan_step(state: PlanExecute):
    # 假设 model 已初始化且支持结构化输出
    plan = await planner_prompt | model.with_structured_output(Plan).ainvoke({"messages": [("user", state["input"])]})
    return {"plan": plan.steps}

3. 执行节点

执行节点负责处理当前步骤。若步骤涉及外部工具，此处应集成相应的 ToolNode。

async def execute_step(state: PlanExecute):
    plan = state["plan"]
    task = plan[0]
    # 格式化计划上下文，帮助执行器理解全局目标
    plan_str = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan))
    task_formatted = f"""For the following plan:
{plan_str}

You are tasked with executing step 1: {task}."""
    
    agent_response = await agent_executor.ainvoke(
        {"messages": [("user", task_formatted)]}
    )
    
    return {
        "past_steps": [(task, agent_response["messages"][-1].content)],
    }

4. 重规划机制

当执行失败或环境变化时，系统需具备动态调整计划的能力。这通常通过条件边实现。

from typing import Union

class Response(BaseModel):
    response: str

class Act(BaseModel):
    action: Union[Response, Plan] = Field(
        description="Action to perform. If you want to respond to user, use Response. \
If you need to further use tools to get the answer, use Plan."
    )

replanner_prompt = ChatPromptTemplate.from_template(
    """For the given objective, come up with a simple step by step plan. \
Your objective was this: {input}
Your original plan was this: {plan}
You have currently done the follow steps: {past_steps}
Update your plan accordingly. If no more steps are needed and you can return to the user, then respond with that. \
Otherwise, fill out the plan. Only add steps to the plan that still NEED to be done."""
)

async def replan_step(state: PlanExecute):
    output = await replanner_prompt | model.with_structured_output(Act).ainvoke(state)
    if isinstance(output.action, Response):
        return {"response": output.action.response}
    else:
        return {"plan": output.action.steps}

5. 工作流编排

最后，将所有节点连接成图，定义流转逻辑。

from langgraph.graph import StateGraph, START, END

def should_end(state: PlanExecute):
    # 判断是否已完成所有步骤或得到最终回答
    if state.get("response"):
        return "END"
    elif len(state.get("plan", [])) == 0:
        return "END"
    else:
        return "agent"

workflow = StateGraph(PlanExecute)
workflow.add_node("planner", plan_step)
workflow.add_node("agent", execute_step)
workflow.add_node("replan", replan_step)

workflow.add_edge(START, "planner")
workflow.add_edge("planner", "agent")
workflow.add_edge("agent", "replan")
workflow.add_conditional_edges(
    "replan",
    should_end,
    ["agent", END]
)

app = workflow.compile()

最佳实践与优化建议

在实际部署中，建议关注以下几点以提升稳定性：

1. 超时控制：为每个执行步骤设置最大耗时，防止死循环。
2. 错误恢复：在执行器中捕获异常，若某一步骤失败，触发重规划而非直接终止。
3. 模型分层：规划阶段可使用轻量级模型，仅在执行复杂推理时使用强模型，平衡性能与成本。
4. 人类介入：在关键决策点（如删除文件、支付操作）引入人工确认环节。

结语

Plan-and-Execute 架构通过解耦规划与执行，有效解决了传统 Agent 在处理复杂任务时的效率与记忆瓶颈。结合 LangGraph 等现代开发框架，开发者可以构建出更加稳健、可控的智能体系统。随着相关生态的成熟，这一模式有望成为企业级 AI 应用的标准架构之一。