在LangGraph中构建规划智能体：规划、执行、重新规划、报告
用于金融研究的四节点LangGraph规划代理：结构化计划、支持命名空间识别的工具注册表、重新规划限制以及可追踪的示例运行过程。
来源: Your AI Agent Isn’t Smart. Here’s How to Build One That Actually Thinks
大多数智能体不过是伪装起来的提示链
关于AI智能体有一个令人不安的事实：很多其实只是披着外衣的升级版提示链。其构成方式很常见——一个系统提示、几种工具，或许再加上一个ReAct循环，这样做出来的东西就被称作智能体。在演示环境中或一切顺利的情况下它能正常工作。但一旦某个工具返回错误，数据出现异常，或者查询需要超过三步的处理，整个系统就会崩溃。
另一种选择是规划智能体。它在使用任何工具之前先制定完整的研究计划，然后按步骤逐一执行该计划；当中间结果与预期不符时，它会重新编写剩余部分的内容。这个演示案例是一个金融研究系统。只需给出用简单英语表达的请求，比如“比较NVDA和AAPL的估值，告诉我目前哪个看起来更优”，它就会制定多步骤计划，利用数据工具执行每一步操作，出问题时自行修正，并生成类似分析师的报告。该系统基于LangGraph构建，可通过Groq调用云端模型，或通过Ollama使用本地模型，整个系统大约包含四个文件。
为何普通的ReAct不够用
ReAct（推理-行动-观察）是大多数智能体起始采用的模式。这类大语言模型会先思考下一步该做什么，调用某个工具，读取返回结果后再次思考，然后调用下一个工具。这种模式很受欢迎，操作简单，适合处理小型任务。
它的核心缺陷在于贪心，因为ReAct智能体只能规划一步之后的行动。如果要求根据三项指标分析五只股票并附带最新新闻，它往往会从NVDA的市盈率开始计算，而第一个得到的结果就会转移它的注意力。有一个原型在记住需要对比多只股票的任务之前，就为每项特斯拉相关的指标调用了七次工具，最终的报告不过是特斯拉的盈利概况加上关于其他股票的短短一句话而已。
问题不在于模型能力，而在于架构设计。如果每次推理过程都必须决定要研究什么、如何研究以及如何处理研究结果，那么这些决策都无法得到妥善处理。规划智能体能够将思考与执行分开：先进行规划，再执行计划，最后检查计划是否仍然有效。每个节点只处理一项任务，它们依次运行。
四个节点与一个循环
规划智能体是一个包含四个节点的有向图，类似于流程图，其中每个方块负责处理一项任务并传递状态。
START --> Planner --> Executor <--> Replanner --> Reporter --> END
- 规划器会接收用户的查询，进而生成一份结构化、分步骤的研究计划。每一步都会明确指出需要调用的工具、相关参数以及选择该工具的原因。
- 执行器会依次执行计划中的每一步，调用指定的工具，并将执行结果记录在临时笔记中。
- 重新规划器会在每次执行完成后审查目前已收集的所有信息。如果某个工具出现错误或数据呈现异常情况，它会重新编写剩余的步骤；否则则保持原计划不变。
- 报告生成器会将所有临时笔记的内容整合成最终的分析师报告，报告中会包含具体的数据及相应的建议。
驱动该设计的正是执行器-重规划器循环。这两个节点交替工作直至计划完成，这种交替机制带来了自我修正的能力，因为每次执行动作后，智能体都可以重新考虑。
状态：共享内存
在LangGraph中，状态就是一切，即节点之间传递的共享内存。其完整定义如下：
import operator
from typing import Annotated, TypedDict
from pydantic import BaseModel, Field
class StrategyState(TypedDict, total=False):
"""Global state that flows through the LangGraph nodes."""
query: str
plan: list[dict]
scratchpad: Annotated[list[dict], operator.add]
current_step: int
final_report: str
replan_count: int
在这种状态下，query 会原封不动地保存用户的原始请求，plan 是步骤的有序列表，scratchpad 用于收集所有工具的运行结果，current_step 指向计划中的当前步骤，而 replan_count 则用于控制 MAX_REPLANS 这一安全上限。
scratchpad 上的标注 Annotated[list[dict], operator.add] 非常重要。其 reducer 即 operator.add，会指示 LangGraph 将新条目追加到列表中，同时保留原有条目不变。当执行器返回 {"scratchpad": [new_entry]} 时，LangGraph 会将其添加到当前列表中，这样节点就无需手动处理列表，也就不会出现修改错误。
结构化计划与决策
Planner从不返回纯文本。它的输出是Pydantic模型，因此大语言模型必须以代码能够解析的格式来生成计划：
AVAILABLE_TOOLS_TEXT = """
- get_metrics(ticker, metric?): Return stock metrics. 'metric' is optional
(P/E, EPS, Revenue, Market Cap, Sector).
- search_news(ticker): Return recent news headlines for a ticker.
- compare_metrics(tickers: list, metric): Compare one metric across
multiple tickers.
"""
class PlanStep(BaseModel):
"""A single executable step inside an analysis plan."""
step_id: int = Field(description="Sequential step number")
tool: str = Field(
description=f"Tool to use. Must be one of:\n{AVAILABLE_TOOLS_TEXT}"
)
args: dict = Field(description="Arguments for the tool call")
purpose: str = Field(description="Why this step is needed")
class Plan(BaseModel):
"""The full plan generated by the planner node."""
goal: str = Field(description="The overall analysis goal")
steps: list[PlanStep] = Field(
description="Ordered list of steps to execute"
)
purpose字段的价值超出预期。它不仅说明某个步骤的功能，还解释了其存在的理由。与单纯的“call compare_metrics,”相比，若能看到“通过比较市盈率来确定相对估值”，Replanner就能更准确地判断是保留、修改还是删除该步骤。
Replanner拥有自己的架构：
class ReplanDecision(BaseModel):
"""Output of the replanner node."""
reasoning: str = Field(
description="Analysis of current progress and findings"
)
should_replan: bool = Field(
description="Whether the plan needs modification"
)
updated_steps: list[PlanStep] = Field(
default_factory=list,
description="Remaining steps if replan is needed. Empty if no changes.",
)
将should_replan设为布尔值可以迫使大语言模型做出明确选择：计划是否需要更改？如果在结构化输出中给大语言模型留出灵活空间，它每次都会做出不同选择。
工具系统
有三款金融研究工具基于真实的模拟数据运行。实际应用中会使用真实的API调用，但在开发阶段模拟数据具有显著优势：即可确定的调试能力。当智能体行为异常时，可以立即判断问题出在其逻辑上还是API上。
另一种值得更多关注的设计模式是具备命名空间意识的工具注册表，它取代了为每个节点配备所有工具的做法：
from langchain_core.tools import BaseTool
from typing import Iterable, Mapping, Any
class ToolRegistry:
"""Namespace-aware container for LangChain tools."""
def __init__(self) -> None:
self._tools_by_toolset: dict[str, dict[str, BaseTool]] = {}
def add_tools(self, toolset: str, tools: Iterable[BaseTool]) -> None:
bucket = self._tools_by_toolset.setdefault(toolset, {})
bucket.update({t.name: t for t in tools})
def get_tools(self, toolset: str) -> tuple[BaseTool, ...]:
return tuple(self._tools_by_toolset.get(toolset, {}).values())
def invoke(
self, toolset: str, tool_name: str, tool_args: Mapping[str, Any]
) -> Any:
t = self._tools_by_toolset.get(toolset, {}).get(tool_name)
if t is None:
raise ValueError(
f"Unknown tool '{tool_name}' in toolset '{toolset}'"
)
return t.invoke(dict(tool_args))
注册在 "planner"、"executor" 和 "replanner" 这些命名空间下进行。所有工具都对 Planner 可见，以便其能够利用这些工具进行规划，但只有 Executor 才有权限调用它们。即便 Replanner 因程序错误而发出了工具调用请求，注册表也会予以拒绝，因为该命名空间没有调用权限。这就是针对智能体的基于角色的访问控制，而职责分离同时也作为一种安全机制，在规划过程中防止不必要的状态变化。
Planner 节点
def planner_node(state: StrategyState) -> dict:
"""Create a step-by-step research plan using structured output."""
planner = model.with_structured_output(Plan)
prompt = PLAN_PROMPT.format(
available_tools=AVAILABLE_TOOLS_TEXT,
ticker_choices=ticker_choices_text(),
metric_choices=metric_choices_text(),
query=state["query"],
)
plan: Plan = planner.invoke(prompt)
steps = [s.model_dump() for s in plan.steps]
return {"plan": steps, "current_step": 0}
model.with_structured_output(Plan) 可让大语言模型以兼容 Pydantic 的 JSON 对象形式回复，若输出格式不正确则自动触发重试。这比用正则表达式解析自由文本要高效得多。
计划提示中的约束规则
PLAN_PROMPT = """\
You are a financial research planner. Given a user's analysis request,
create a step-by-step research plan using the available tools.
Available tools:
{available_tools}
Rules:
- Use only the tools listed above.
- Every plan step must be executable with one of those tools.
- When a tool accepts 'ticker' or 'tickers', use only these exact values:
{ticker_choices}
- When a tool accepts 'metric', use one of these exact values:
{metric_choices}
- There are no other tools available. Final synthesis is handled separately.
Create an efficient plan. Group related lookups. Aim for 4-8 steps.
User request: {query}"""
有两个要点值得注意。提示语会列出有效的参数值，这样大语言模型就不必猜测有哪些股票代码或指标存在。这或许是针对工具调用型智能体最有效的防幻觉方法。如果没有这一提示，模型会自信地规划诸如 get_metrics(ticker="BITCOIN", metric="vibes") 这样的步骤，而当这些步骤失败时又会显得十分惊讶。
另一个要点是“没有其他可用工具”这句话。如果缺少这句话，大语言模型就会编造出 analyze_sentiment(ticker="NVDA") 或 generate_report(findings=...) 这类并不存在的工具步骤。
“目标为4到8步”的要求刻意保持了灵活性。诸如“必须恰好5步”这样的严格限制反而适得其反，它会导致简单查询需要多余的步骤，而复杂查询则被迫塞进已超负荷的步骤中。设定范围可以让模型有调整空间，以指导原则代替强制要求通常能使其做出正确的决策。
执行节点
执行节点被设计得极为简单：它负责执行下一步操作，通过注册表调用工具，并存储输出结果。
MAX_STEPS = 12 # Safety limit on total steps
def executor_node(state: StrategyState) -> dict:
"""Execute the next pending step from the plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
if current_step >= len(plan):
return {}
if current_step >= MAX_STEPS:
return {"current_step": len(plan)}
step = plan[current_step]
tool_name = step["tool"]
tool_args = step["args"]
try:
result = str(
TOOL_REGISTRY.invoke(
AgentName.EXECUTOR.value, tool_name, tool_args
)
)
status = "Error" if result.startswith("Error:") else "Success"
except Exception as exc:
result = f"Error: {exc}"
status = "Error"
entry = {
"step": current_step + 1,
"tool": tool_name,
"args": tool_args,
"result": result,
"status": status,
}
return {
"scratchpad": [entry],
"current_step": current_step + 1,
}
MAX_STEPS = 12 是一个硬性上限。如果重规划器不断增加步骤数或陷入僵局，智能体也无法无限消耗 API 信用额度。一旦达到该上限，执行器会将 current_step 设为 len(plan)，然后流程会将目前收集到的数据传递给报告器。
失败情况通过约定方式识别：status = "Error" if result.startswith("Error:") else "Success"。三个工具中的每一个都会在错误信息前加上 "Error:"，这样执行器无需进行 JSON 解析或针对特定异常的处理即可标记失败。重规划器在决定是重新尝试还是调整计划时，会读取此状态值。
重规划器节点
该节点体现了规划代理与简单协调器之间的区别。在每次工具执行后，它会综合所有证据，判断剩余的规划方案是否依然合理：
MAX_REPLANS = 2 # Prevent infinite replanning
def replanner_node(state: StrategyState) -> dict:
"""Review progress and optionally modify the remaining plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
replan_count = state.get("replan_count", 0)
scratchpad = state.get("scratchpad", [])
remaining = plan[current_step:]
if len(remaining) = MAX_REPLANS:
return {}
scratchpad_text = "\n".join(
f"Step {e['step']}: {format_tool_call(e['tool'], e['args'])} "
f"-> [{e['status']}] {e['result'][:150]}..."
for e in scratchpad
)
remaining_text = "\n".join(
f"Step {s['step_id']}: {format_tool_call(s['tool'], s['args'])} "
f"- {s['purpose']}"
for s in remaining
)
replanner = model.with_structured_output(ReplanDecision)
prompt = REPLAN_PROMPT.format(
goal=state["query"],
scratchpad=scratchpad_text,
remaining_steps=remaining_text,
)
decision: ReplanDecision = replanner.invoke(prompt)
if decision.should_replan and decision.updated_steps:
new_steps = plan[:current_step] + [
s.model_dump() for s in decision.updated_steps
]
return {"plan": new_steps, "replan_count": replan_count + 1}
return {"replan_count": replan_count + 1}
查看 e['result'][:150]。Replanner 会将每个工具的返回结果截取为150个字符，因为它需要的是核心内容而非完整的 JSON 数据，而 Reporter 后续会收到未截断的数据。在细节无关紧要的地方会节省令牌，在需要详细信息的地方则使用令牌。
当重新规划时，plan[:current_step] + [s.model_dump() for s in decision.updated_steps] 会保留已完成的步骤，仅替换未完成的步骤，从而不会改动历史记录。
重新规划提示
REPLAN_PROMPT = """\
You are a financial research planner reviewing progress on a research task.
Original goal: {goal}
Completed steps and findings so far:
{scratchpad}
Remaining steps in the plan:
{remaining_steps}
Based on the findings so far, should the remaining plan change?
If an expected tool failed or revealed something unexpected, add a step
to investigate.
If a step is now redundant, remove it.
Use only the available executable tools already shown in the plan.
Do not add recommendation, summary, or report-writing steps."""
最后那行代码起着关键作用。如果去掉“不要添加推荐、总结或报告撰写步骤”这一限制，Replanner就会在每个计划中添加一个综合步骤，尽管实际上应由Reporter来承担这项工作。提示语需要明确界定各个节点之间的边界。
MAX_REPLANS = 2这一上限恰到好处。在早期版本中，一个三步任务需要经过六轮重新规划，而大型语言模型找到的“改进”实际上并无任何实际效果。两轮重新规划就足以实现真正的自我修正，比如处理出错的工具或删除冗余步骤，同时避免分析瘫痪的情况。最终必须有人指示智能体停止犹豫并完成工作。
记者会获取完整的未截断草稿，然后编写结构化的 Markdown 报告。该流程会调用 model.invoke(prompt).content，得到的是纯文本而非结构化输出，这样大型语言模型就可以按自己认为合适的方式呈现分析结果。其中最有效的指令是：“请具体说明，并引用数据中的数字。”如果没有这一要求，报告就会显得含糊不清，只会说“NVDA的基本面很强劲”，而不会提及实际的市盈率与营收数据。
在 LangGraph 中整合这些组件
这里四个节点会相互连接：
from langgraph.graph import StateGraph, END
from enum import Enum
class AgentName(Enum):
PLANNER = "planner"
EXECUTOR = "executor"
REPLANNER = "replanner"
REPORT = "report"
def build_graph():
"""Build and compile the LangGraph planning-agent workflow."""
workflow = StateGraph(StrategyState)
workflow.add_node(AgentName.PLANNER.value, planner_node)
workflow.add_node(AgentName.EXECUTOR.value, executor_node)
workflow.add_node(AgentName.REPLANNER.value, replanner_node)
workflow.add_node(AgentName.REPORT.value, report_node)
workflow.set_entry_point(AgentName.PLANNER.value)
workflow.add_edge(AgentName.PLANNER.value, AgentName.EXECUTOR.value)
workflow.add_edge(AgentName.EXECUTOR.value, AgentName.REPLANNER.value)
workflow.add_conditional_edges(
AgentName.REPLANNER.value,
should_continue_execution,
{
AgentName.EXECUTOR.value: AgentName.EXECUTOR.value,
AgentName.REPORT.value: AgentName.REPORT.value,
},
)
workflow.add_edge(AgentName.REPORT.value, END)
return workflow.compile()
def should_continue_execution(state: StrategyState) -> str:
"""Return the next node name after re-planning."""
if state.get("current_step", 0) >= len(state.get("plan", [])):
return AgentName.REPORT.value
return AgentName.EXECUTOR.value
该流程图几乎就像英文句子一样。计划器将任务交给执行器，执行器再将任务交给重新规划器；如果还有步骤未完成，重新规划器会将任务转回执行器，而一旦所有步骤都处理完毕，则转交给报告器。报告器最终会到达终点。
AgentName 枚举看似简单，但实际上很有用。代码中散布的诸如 "planner" 这样的纯字符串容易引发拼写错误，而 "plannr" 则会悄悄创建出一个孤立节点。使用枚举后，这类错误会在导入时就被发现。
路由逻辑只检查一个条件：所有步骤是否都已完成？当 current_step >= len(plan) 时，控制权会交给报告器，否则则返回执行器。由于每次执行时 current_step 的值都会增加，且 MAX_STEPS = 12 设定了上限，即使重新规划器添加了新步骤，循环也会终止。
一次查询，全程追踪
该示例请求让智能体分析NVDA和AMD，将两者的市盈率及最新新闻并列展示，然后再从中选择一个。
规划阶段。规划器会生成五个步骤：NVDA的完整指标、AMD的完整指标、市盈率的直接对比、NVDA的新闻检索以及AMD的新闻检索。首先呈现各公司的详细资料以获得全面了解，接着进行直接对比，最后是新闻内容，因为新闻属于补充性背景信息而非核心估值数据。
执行阶段。第一步和第二步会返回结构清晰的指标JSON数据，每次执行后Replanner都会确认数据无误且计划未变。第三步则会返回市盈率的对比结果：
{
"metric": "P/E",
"values": {
"NVDA": 58.3,
"AMD": 102.5
}
}
两个数值均存在且有效，因此无需重新规划。第4步和第5步各会返回三篇新闻文章。第5步完成后，current_step的值将等于len(plan)，控制权便会交给报告员。
报告。收到五份完整的草稿记录后，报告员会列出具体数字，例如NVDA的市盈率为58.3，而AMD为102.5，尽管NVDA的市值高出约2.8万亿美元，但其相对价值仍更优。在Groq上完成一次完整处理大约需要8到10秒，而在本地使用40亿参数的Ollama模型则需30到40秒。
最重要的升级
当前版本使用的是模拟数据，但其架构已可投入实际应用。其中三项升级最为重要：
- 真实的财务数据。 可以用调用 Yahoo Finance 或 Alpha Vantage 的方式替代字典查询。由于
@tool装饰器模式的作用，签名保持不变，只有函数体发生改变，因此规划器和执行器永远无法知道这些数据来自模拟的 JSON 文件还是 REST API。 - 持久化内存。 目前，临时存储的数据仅在一次运行中有效。借助 ChromaDB 或 Pinecone 等向量存储工具，代理可以查询之前的分析结果，从而回答“NVDA 与我们上个月看到的情况相比如何？”这样的问题，进而成为能够持续进化的分析工具，而非无状态的工具。
该结构无法提供的功能
计划-执行-重新规划循环能够完成其任务。结构化的输出能让智能体按既定路径行事，工具注册机制可避免混乱，而安全限制则能防止过度支出。这样的设计十分稳健，代码也保持整洁。
即便如此，整个系统的智能仍依赖于其底层的大语言模型。更强大的模型能够制定更优的计划，在重新规划时发现更多错误，并生成更精准的报告；而较弱的模型则只会机械地遵循既定结构，给出平庸的分析结果。图表无法让模型变得更聪明，它所做的只是为模型提供更好的框架，让其能够更好地运用已有的智能。
项目管理层面在很大程度上已经得到解决：状态管理、工具协调、执行流程以及自我修正都遵循明确的模式。但决策环节仍存在不确定性——重新规划器究竟应在何时进行重新规划，又应在何时保持沉默？何时数值属于异常情况，何时只是令人意外？目前这些判断都体现在提示词中，期望大语言模型能将其处理妥当，实际效果往往超出预期，但也常常不尽如人意。
相关阅读
- DS-STAR详解：谷歌的数据科学智能体及其优势所在 — 详细介绍谷歌的DS-STAR数据科学智能体：其七个模块、DS-STAR+报告生成机制、已发布的提示词、消融实验结果以及现存的局限性。
- 用Python构建DevOps AI智能体：LangChain、Claude与安全化的Bash工具 — 构建能够运行Bash命令、读取日志并查询Kubernetes系统的智能体：首先制定安全规则，再选择相应工具，采用ReAct循环机制、内存管理功能、FastAPI封装层，并解决常见故障。