智能体架构 -- 规划架构，让你的智能体效率翻倍！

想象一下，你刚接手一个复杂需求：分析欧洲三大首都城市的人口，并和美国的做个对比。你派给新来的实习生，他立刻开始查巴黎人口，查完再查柏林，查到一半突然问我“罗马人口要不要？”……是不是有点抓狂？

其实，这就是很多AI智能体（Agent）的日常。它们虽然聪明，但往往“目光短浅”，像ReAct模式那样，拿到任务就一头扎进去，每一步都要重新思考、重新决策。这在任务路径清晰时，不仅效率低，还容易跑偏。

那么，有没有办法让AI也学会“三思而后行”？今天，我们就来深入探讨规划（Planning）架构。我们将手把手教你构建一个先制定完整计划、再按部就班执行的智能体，并让它和传统的ReAct模式正面交锋，看看谁才是处理多步骤任务的王者。

推荐阅读：

智能体架构 -- 反思模式，让AI自己当代码审查员

智能体架构 -- 工具使用，秒变真·智能助手

智能体架构 -- ReAct，让AI学会“三思而后行”

为什么AI需要“先想后做”？

在开始编码前，我们先来理解什么是规划架构。它的核心理念很简单：在行动之前，先把任务拆解成一个清晰的步骤清单。

什么是规划架构？

你可以把它想象成一次长途自驾游。ReAct模式就像是开着导航，每到一个路口才告诉你下一步怎么走；而规划架构则是在出发前，你就已经规划好了全程路线：从A城市到B城市，再到C城市，最后到达目的地。

在技术世界里，规划架构是这样工作的：

1. 接收目标：智能体拿到一个复杂任务，比如“计算巴黎、柏林和罗马的人口总和，并与美国人口对比”。
2. 规划阶段：一个专门的“规划器”模块分析任务，生成一个有序的子任务列表。比如：
• web_search('巴黎人口')
• web_search('柏林人口')
• web_search('罗马人口')
• web_search('美国人口')
• 计算并比较
3. 执行阶段：一个“执行器”模块拿到这个清单，按顺序调用网络搜索工具，收集数据。
4. 综合阶段：最后，一个“综合器”模块将收集到的所有数据整合起来，形成最终答案。

什么时候该用它？

规划架构特别适合以下场景：

• 多步骤工作流：比如生成一份周报，需要先抓取数据、再处理数据、最后生成报告。操作顺序是明确的。
• 项目管理助手：把一个“发布新功能”的大目标，拆解为“设计评审”“开发”“测试”“上线”等可执行的子任务。
• 教育辅导：为一个学生制定学习“微积分”的计划，从“极限”到“导数”再到“积分”，循序渐进。

优势和局限

优势：

• 过程透明：你能清楚地看到AI的每一步计划，方便调试和优化。
• 执行高效：对于路径清晰的任务，它避免了ReAct模式中“思考-行动-再思考”的重复循环，节省了时间和算力。

局限：

• 应变能力差：一旦计划制定好，如果执行过程中环境发生变化（比如某个数据源失效），它可能无法灵活调整。这时，ReAct那种“走一步看一步”的模式反而更具优势。

动手实战：搭建一个规划型智能体

理论说完了，我们直接上代码。我们将分几步走：先搭建一个基线（ReAct智能体），再构建我们的规划型智能体，最后让它们同台竞技。

阶段 0：准备工作

所有实战都需要一个干净的起点。我们先来安装必要的库，并配置好我们的API密钥。这里我们使用Nebius作为LLM服务商，Tavily作为网络搜索工具，并用LangSmith来追踪和调试。

# 安装依赖库
# !pip install -q -U langchain-nebius langchain langgraph rich python-dotenv langchain-tavily

import os
import re
from typing import List, Annotated, TypedDict, Optional
from dotenv import load_dotenv

# LangChain 核心组件
from langchain_nebius import ChatNebius
from langchain_core.messages import BaseMessage, ToolMessage, SystemMessage
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_tavily import TavilySearch

# LangGraph 组件 (用于构建工作流)
from langgraph.graph import StateGraph, END
from langgraph.graph.message import AnyMessage, add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# 美观打印
from rich.console import Console
from rich.markdown import Markdown

# --- 加载API密钥和追踪设置 ---
load_dotenv()
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Agentic Architecture - Planning (Nebius)"# 在LangSmith中创建的项目名

# 检查密钥是否已设置
for key in ["NEBIUS_API_KEY", "LANGCHAIN_API_KEY", "TAVILY_API_KEY"]:
ifnot os.environ.get(key):
        print(f"⚠️ 未找到{key}。请创建一个.env文件并填入你的密钥。")
else:
        print(f"✅ {key} 已加载")

console = Console()
print("\n🚀 环境准备就绪！")

阶段 1：基线测试 - 反应型智能体 (ReAct)

为了对比，我们先快速搭建一个标准的ReAct智能体。它的工作流程是：思考 -> 行动 -> 观察结果 -> 再思考...。它会像新手一样，一步步摸索着前进。

# 定义智能体状态，用于在图中传递消息
classAgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

# 1. 初始化Tavily搜索工具
tavily_search_tool = TavilySearch(max_results=2)

# 2. 将搜索工具包装成LangChain工具
@tool
defweb_search(query: str) -> str:
"""使用Tavily执行网络搜索，并将结果作为字符串返回。"""
    console.print(f"🔧 [cyan]正在搜索: '{query}'...[/cyan]")
    results = tavily_search_tool.invoke(query)
return results

# 3. 定义LLM并绑定工具
llm = ChatNebius(model="meta-llama/Meta-Llama-3.1-8B-Instruct", temperature=0)
llm_with_tools = llm.bind_tools([web_search])

# 4. 定义智能体节点，强制一次只调用一个工具
defreact_agent_node(state: AgentState):
    console.print("🤔 [yellow]ReAct智能体: 思考下一步...[/yellow]")
    messages_with_system_prompt = [
        SystemMessage(content="你是一个研究助手。每次必须且只能调用一个工具。收到结果后，再决定下一步。")
    ] + state["messages"]
    response = llm_with_tools.invoke(messages_with_system_prompt)
return {"messages": [response]}

# 5. 构建ReAct图
tool_node = ToolNode([web_search])
react_graph_builder = StateGraph(AgentState)
react_graph_builder.add_node("agent", react_agent_node)
react_graph_builder.add_node("tools", tool_node)
react_graph_builder.set_entry_point("agent")
react_graph_builder.add_conditional_edges("agent", tools_condition)  # 根据是否调用工具决定下一步
react_graph_builder.add_edge("tools", "agent")  # 工具调用后回到智能体

react_agent_app = react_graph_builder.compile()
print("✅ ReAct智能体编译成功。")

用ReAct智能体解决规划型任务

现在，给ReAct智能体一个需要多个步骤的任务，看看它的表现。

# 定义一个需要多步搜索和计算的任务
plan_centric_query = """
找出法国、德国和意大利首都城市的人口。
然后计算它们的总和。
最后，将这个总和与美国的人口进行比较，并说出哪个更大。
"""

console.print(f"\n🎯 [bold yellow]给ReAct智能体下达任务:[/bold yellow] '{plan_centric_query}'\n")

final_react_output = None
# 流式执行，观察每一步的思考过程
for chunk in react_agent_app.stream({"messages": [("user", plan_centric_query)]}, stream_mode="values"):
    final_react_output = chunk
    console.print(f"--- [purple]状态更新[/purple] ---")
    chunk['messages'][-1].pretty_print()
    console.print("\n")

console.print("\n--- [bold red]ReAct智能体最终输出[/bold red] ---")
console.print(Markdown(final_react_output['messages'][-1].content))

输出讨论： 从输出中，你能看到ReAct智能体是如何一步步行动的。它会先搜索巴黎人口，然后搜索柏林人口，接着是罗马人口和美国人口，最后才进行计算。虽然它最终完成了任务，但这个过程显得有些“绕路”，每一步之间都需要额外的LLM推理来决策下一步做什么。如果任务更复杂，它很容易迷失方向。

阶段 2：进阶方法 - 规划型智能体

现在，我们来构建一个“先思后行”的规划型智能体。它的核心组件有三个：规划器、执行器和综合器。

# 定义一个Pydantic模型，强制规划器输出结构化的步骤列表
classPlan(BaseModel):
"""回答用户查询所需执行的工具调用计划。"""
    steps: List[str] = Field(description="一个工具调用列表，按顺序执行后将回答用户的查询。")

# 为规划型智能体定义状态
classPlanningState(TypedDict):
    user_request: str
    plan: Optional[List[str]]
    intermediate_steps: List[ToolMessage]  # 存放工具执行结果
    final_answer: Optional[str]

defplanner_node(state: PlanningState):
"""规划器节点：生成行动计划。"""
    console.print("📝 [bold green]规划器: 正在分解任务...[/bold green]")
    planner_llm = llm.with_structured_output(Plan)

# 使用少样本提示，让LLM知道我们想要什么样的计划
    prompt = f"""你是一个专业的规划员。你的工作是创建一个逐步的计划来回答用户的请求。
        计划中的每一步必须是对 `web_search` 工具的单次调用。

        **指令:**
        1. 分析用户的请求。
        2. 将其分解为一系列简单、合乎逻辑的搜索查询。
        3. 将输出格式化为一个字符串列表，其中每个字符串是一个单一有效的工具调用。

        **例子:**
        请求: "法国的首都是什么，它的人口是多少？"
        正确计划输出:
        [
            "web_search('capital of France')",
            "web_search('population of Paris')"
        ]

        **用户的请求:**
{state['user_request']}
    """
    plan_result = planner_llm.invoke(prompt)
    console.print(f"✅ [green]规划器: 生成的计划: {plan_result.steps}[/green]")
return {"plan": plan_result.steps}

defexecutor_node(state: PlanningState):
"""执行器节点：执行计划中的下一步。"""
    console.print("⚙️ [bold blue]执行器: 正在运行下一步...[/bold blue]")
    plan = state["plan"]
    next_step = plan[0]  # 总是取计划列表的第一个元素

# 用正则表达式从 "web_search('query')" 中提取函数名和参数
    match = re.search(r"(\w+)\((?:\"|\')(.*?)(?:\"|\')\)", next_step)
ifnot match:
        tool_name = "web_search"
        query = next_step
else:
        tool_name, query = match.groups()[0], match.groups()[1]

    console.print(f"🔧 [cyan]执行器: 调用工具 '{tool_name}'，查询参数是 '{query}'...[/cyan]")

    result = tavily_search_tool.invoke(query)

# 将结果包装成ToolMessage，存入状态
    tool_message = ToolMessage(
        content=str(result),
        name=tool_name,
        tool_call_id=f"manual-{hash(query)}"# 需要一个唯一的ID
    )

return {
"plan": plan[1:],  # 从计划中移除已执行的步骤
"intermediate_steps": state["intermediate_steps"] + [tool_message]
    }

defsynthesizer_node(state: PlanningState):
"""综合器节点：根据所有中间步骤生成最终答案。"""
    console.print("📄 [bold magenta]综合器: 正在生成最终答案...[/bold magenta]")

# 将所有工具结果拼成一个上下文
    context = "\n".join([f"工具 {msg.name} 返回: {msg.content}"for msg in state["intermediate_steps"]])

    prompt = f"""你是一个专业的综合员。根据用户的请求和收集到的数据，提供一个全面的最终答案。

    请求: {state['user_request']}
    收集的数据:
{context}
    """
    final_answer = llm.invoke(prompt).content
return {"final_answer": final_answer}

print("✅ 规划器、执行器和综合器节点定义完成。")

构建规划型智能体图

现在，我们将这三个节点串联成一个工作流。流程图非常直观：规划器 -> (如果计划不为空，则执行器 -> 循环) -> 综合器。

defplanning_router(state: PlanningState):
"""路由函数：决定下一步是执行还是综合。"""
ifnot state["plan"]:
        console.print("✅ [yellow]路由器: 计划完成。转向综合器。[/yellow]")
return"synthesize"
else:
        console.print("➡️ [yellow]路由器: 还有更多步骤。继续执行。[/yellow]")
return"execute"

# 构建图
planning_graph_builder = StateGraph(PlanningState)
planning_graph_builder.add_node("plan", planner_node)
planning_graph_builder.add_node("execute", executor_node)
planning_graph_builder.add_node("synthesize", synthesizer_node)

planning_graph_builder.set_entry_point("plan")
# 条件边：从规划节点出发，根据路由函数决定下一个节点
planning_graph_builder.add_conditional_edges("plan", planning_router, {
"execute": "execute",
"synthesize": "synthesize"
})
# 条件边：从执行节点出发，同样根据路由函数决定是继续执行还是去综合
planning_graph_builder.add_conditional_edges("execute", planning_router, {
"execute": "execute",
"synthesize": "synthesize"
})
planning_graph_builder.add_edge("synthesize", END)

planning_agent_app = planning_graph_builder.compile()
print("✅ 规划型智能体编译成功。")

阶段 3：正面交锋

现在，让我们的规划型智能体执行同一个任务，看看它的表现有何不同。

console.print(f"\n🎯 [bold green]给规划型智能体下达任务:[/bold green] '{plan_centric_query}'\n")

initial_planning_input = {"user_request": plan_centric_query, "intermediate_steps": []}
final_planning_output = planning_agent_app.invoke(initial_planning_input)

console.print("\n--- [bold green]规划型智能体最终输出[/bold green] ---")
console.print(Markdown(final_planning_output['final_answer']))

输出讨论： 注意观察输出日志，你会发现整个过程截然不同。智能体并没有立刻去搜索，而是先调用“规划器”生成了一个完整的行动计划。然后，“执行器”严格按照这个计划，按部就班地执行每一步搜索，中间没有任何额外的“思考”步骤。最后，“综合器”将所有结果汇总，一气呵成。

这个过程更透明（我们能提前看到计划）、更稳健（不会跑偏），也更高效（避免了不必要的LLM推理）。

阶段 4：量化评估

为了更客观地比较，我们引入一个“LLM作为评委”的机制，从任务完成度和过程效率两个维度打分。

classProcessEvaluation(BaseModel):
"""评估智能体解决问题的过程的模式。"""
    task_completion_score: int = Field(description="智能体是否成功完成任务，打分1-10分。")
    process_efficiency_score: int = Field(description="智能体处理过程的效率和直接性，打分1-10分。分数越高表示过程越符合逻辑、弯路越少。")
    justification: str = Field(description="对分数的简要说明。")

judge_llm = llm.with_structured_output(ProcessEvaluation)

defevaluate_agent_process(query: str, final_state: dict):
"""评估智能体的执行过程。"""
# 从不同的状态中提取执行轨迹
if'messages'in final_state:
# ReAct智能体的轨迹在messages里
        trace = "\n".join([f"{m.type}: {str(m.content)[:200]}"for m in final_state['messages']])  # 截断避免过长
else:
# 规划型智能体的轨迹在intermediate_steps里
        trace = f"计划: {final_state.get('plan', [])}\n步骤:\n" + \
"\n".join([f"  - {msg.name}: {str(msg.content)[:200]}"for msg in final_state.get('intermediate_steps', [])])

    prompt = f"""你是一个专门评判AI智能体的专家。请按照1-10分评估该智能体解决任务的过程。
    重点关注过程是否符合逻辑且高效。

    **用户任务:** {query}
    **完整智能体追踪记录:**\n```\n{trace}\n```
    """
return judge_llm.invoke(prompt)

console.print("--- 📊 评估反应型智能体的过程 ---")
react_agent_evaluation = evaluate_agent_process(plan_centric_query, final_react_output)
console.print(react_agent_evaluation.model_dump())

console.print("\n--- 📊 评估规划型智能体的过程 ---")
planning_agent_evaluation = evaluate_agent_process(plan_centric_query, final_planning_output)
console.print(planning_agent_evaluation.model_dump())

输出讨论： 评委的打分非常直观地量化了我们的观察。两个智能体的任务完成度可能都很高，但规划型智能体在过程效率上会明显胜出。评委的解释会强调，规划型智能体解决问题的方式更直接、更符合逻辑，而ReAct智能体的路径则略显迂回。

写在最后

通过这次实战，我们亲身体验了规划架构带来的改变。它让AI从“走一步看一步”的菜鸟，变成了“谋定而后动”的高手。

核心回顾

1. 原理：规划架构通过“先规划，后执行”的模式，为AI处理复杂任务提供了清晰的结构和路径。
2. 实践：我们用LangGraph构建了包含规划器、执行器、综合器的规划型智能体，并让它与ReAct智能体同台竞技。
3. 避坑：规划架构并非万能。它适用于路径清晰的“确定性”任务，而在探索性强、环境多变的任务中，ReAct模式可能更合适。

理解这两种模式的权衡，是成为一名优秀的AI系统架构师的关键。选择正确的架构，就像为问题选择正确的算法，能让你的智能体事半功倍。

在实际工作中，你会遇到哪些场景，感觉ReAct模式“力不从心”，需要一个更强的“规划者”来统筹？或者，你尝试过用其他更巧妙的方式来构建你的智能体吗？欢迎在评论区分享你的故事！