每个LangGraph智能体都遵循的标准七部分布局结构。
LangGraph智能体采用的固定文件结构，涵盖从导入与状态管理到工具、节点、边以及组装与调用的各个部分，同时还提供模板及多智能体扩展功能。
来源: The LangGraph Mental Model: A standardized architecture guide for every agent you’ll ever build
任何学习过 LangGraph 一段时间的人都会有一种奇怪的感觉：概念上明明很清晰，但代码却怎么也写不出来。图由节点构成，节点负责执行操作，边决定下一步的走向，状态则用于存储信息。这些内容五分钟就能在白板上画出来，但一旦打开空白文件，却毫无头绪。该导入的库被忘了，想要的类也找不到合适的放置位置，而且似乎有五种不同的方式可以实现同样的功能。
为何代码比概念更难
根本原因在于 LangGraph 本身并不提供任何固定结构。它只提供功能完备的构建模块，具体的组织方式由使用者自行决定。不同的教程和代码库对内容的编排方式各不相同，因此你的记忆中也永远无法形成统一的框架。
此处提出的解决方案是选定一种标准布局，并将其应用于所有智能体，无论是小型聊天机器人还是由多个协同工作的智能体构成的流程。可以将其视为一份无需费力就能记住并调用的模板，类似于展示智能体已完成操作的图表，只不过它是针对源代码而非运行时历史记录的。
全局文件概览
每个LangGraph应用程序都可以写成一个包含七个部分的单一文件，这些部分从上到下按固定顺序排列：
langgraph_agent.py
│
├── MODULE 1: IMPORTS & CONFIGURATION
│ └── All your libraries, API keys, model setup
│
├── MODULE 2: STATE
│ └── The TypedDict that defines your agent's memory
│
├── MODULE 3: TOOLS (optional, but common)
│ └── Functions decorated with @tool that the LLM can call
│
├── MODULE 4: NODES
│ └── Functions that do the actual work at each graph step
│
├── MODULE 5: EDGES & ROUTING
│ └── Functions that decide what happens next
│
├── MODULE 6: GRAPH ASSEMBLY
│ └── Where you build, wire, and compile the graph
│
└── MODULE 7: ENTRYPOINT
└── The __main__ block or invoke() call that runs everything
这就是框架结构。开始一个项目时，先打开一个文件，将七个章节标题以注释形式输入，然后再逐一填充内容。接下来的部分会详细介绍每个章节的功能、需要了解的名称以及可复用的模板。
第1节：导入与配置
功能说明
可以把这一节视为工具箱。木匠在开始制作任何物品之前都会先准备好所有工具，而在这里你则是告诉Python需要哪些外部功能，并设置智能体将要使用的模型。
需要了解的名称
ChatOpenAI或ChatAnthropic是语言模型类，也就是智能体的“大脑”。只需在这里创建一次，即可在所有地方重复使用。StateGraph是用于构建图结构的核心类，相当于你绘制的蓝图。
END 是用于表示终止点的特殊常量。指向该点的边会终止程序运行。MemorySaver 是一种检查指针，可用于在多轮对话中保留内存状态。模板
# --- Standard Library ---
import os
from typing import TypedDict, Annotated, Literal
# --- LangChain Core ---
from langchain_openai import ChatOpenAI # or ChatAnthropic, ChatGroq, etc.
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langchain_core.tools import tool
# --- LangGraph Core ---
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages # The message reducer
from langgraph.prebuilt import ToolNode # Pre-built node for tool execution
from langgraph.checkpoint.memory import MemorySaver
# --- Configuration ---
# Always name your model variable 'llm' - it's the standard in every node
llm = ChatOpenAI(
model="gpt-4o", # or "claude-3-5-sonnet-20241022", etc.
temperature=0, # 0 = deterministic; raise for creativity
api_key=os.environ.get("OPENAI_API_KEY")
)
为何如此布局
这些导入项被分为四组：标准库、LangChain（模型层）、LangGraph（图层）以及配置文件。这种划分方式与这些库本身的层级结构相对应——LangChain负责提供人工智能基础功能，而LangGraph则负责任务协调。将不同组别分开，便于在出现问题时快速判断各部分的功能归属。
这里最实用的习惯是使用 llm 来称呼模型。后续的所有节点函数都会引用该变量；如果某个项目使用 model，另一个项目又使用 chat，代码的统一性就会被打乱，阅读起来也会更加困难。
第2节：状态
其作用是什么
State是智能体的工作内存，是贯穿整个图结构的唯一事实来源。每个节点都可以读取和写入它，而引擎则负责管理它的生命周期。在LangGraph的所有要素中，正确实现State可能是最重要的。
办公室的白板是个很好的比喻：进入房间的任何人，也就是每个节点，都可以查看上面的内容并添加新内容，而白板在整个会议期间——也就是整个图结构运行期间——都保持不变。
需要了解的名称
TypedDict是Python中的一个类，用于描述State的结构，类似于模式定义。你的State类会继承自它。Annotated是一个类型提示封装类。LangGraph用它来为某个字段附加一个reducer，即决定新值如何与现有值合并的函数。
add_messages 是用于 messages 字段的内置 reducer。每当有节点添加消息时，它不会替换整个列表，而是选择追加新消息。这是你会最常使用的 reducer。operator.add 是另一种常见的 reducer，适用于需要通过加法累积值的字段，比如结果列表或计数器。模板
# ============================================================
# MODULE 2: STATE
# ============================================================
class AgentState(TypedDict):
# 'messages' is the heartbeat of almost every LangGraph agent.
# Annotated[list, add_messages] means: "this is a list, and when
# a node writes to it, append - don't replace."
messages: Annotated[list[BaseMessage], add_messages]
# Add custom fields below for your specific agent's needs.
# Fields without a reducer are REPLACED each time a node writes to them.
# Example: a simple string field (gets replaced each write)
current_task: str
# Example: a list you want to accumulate (use operator.add as reducer)
# results: Annotated[list[str], operator.add]
# Example: a counter
# iteration_count: int
关于 reducer 的思考
Reducer 让许多初学者感到困惑，这里有一个类比。把状态想象成在办公室里传阅的纸质表格。messages 字段就像一叠便签纸：处理表格的每个人都会添加便签，而没有人会删除之前的便签。这就是 add_messages 的作用。
没有 reducer 的字段则相当于一个文本框，每个人都会覆盖前一个人写的内容，这正是 LangGraph 的默认行为——最后写入的内容有效。
有一个简单的规则：每次为 messages 字段指定 Annotated[list, add_messages]，对于那些需要持续增长的列表则选择 Annotated[list, operator.add]，其余字段则保持不变。
第 3 节：工具
其用途
工具是智能体的双手。大语言模型则是大脑，负责推理和决策，但仅凭自身无法在现实中采取行动。借助工具，它能够浏览网页、调用 API、执行代码、在数据库中查询信息，或是运行用户定义的任何 Python 函数。
本部分为可选内容。仅生成文本而不接触外部系统的智能体可以跳过它，不过大多数实际智能体都需要工具。
需要了解的名称
@tool是 LangChain 中的一个装饰器，可将普通的 Python 函数转换为 LangGraph 能识别的工具。该函数的文档字符串会成为大语言模型在决定是否使用它时所阅读的描述。ToolNode是一个现成的节点，位于langgraph.prebuilt中，它会执行大语言模型选定的工具功能，因此无需用户自行编写执行逻辑。
bind_tools 是您 LLM 实例上的一种方法，用于将工具绑定到该实例上。之后模型就能知晓有哪些工具存在，并且可以通过发送特殊的 tool_calls 消息而非普通文本来请求调用这些工具。模板
# ============================================================
# MODULE 3: TOOLS
# ============================================================
@tool
def search_web(query: str) -> str:
"""Search the web for current information about a topic.
Use this when you need real-time information that is not in
your training data, such as recent news or live prices.
Args:
query: The search query string.
Returns:
A string containing search results.
"""
# Your actual implementation here (e.g., Tavily, SerpAPI, etc.)
# Placeholder for illustration:
return f"Search results for: {query}"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression and return the result.
Use this for any arithmetic, algebra, or numerical computation.
Args:
expression: A valid Python math expression as a string, e.g. '2 + 2 * 10'
Returns:
The computed result as a string.
"""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
# Collect all tools into a list - this is the pattern you always follow
tools = [search_web, calculate]
# Bind tools to the LLM so it knows they exist and can choose to call them
llm_with_tools = llm.bind_tools(tools)
# Create the pre-built ToolNode that will execute tool calls automatically
tool_node = ToolNode(tools)
文档字符串的重要性超出想象
初学者常常忽视这一点：模型会依据你的文档字符串来选择工具。它会读取三引号内的文本。如果文档字符串含糊不清或缺失，大型语言模型就无从判断你的工具是否合适，因此请撰写清晰具体的描述，说明该工具的功能以及使用时机。
第4节：节点
其作用
节点即执行任务的单元。每个节点都是一个Python函数，它接收当前状态，执行诸如调用大型语言模型、处理数据或调用工具等操作，然后返回一个包含状态更新内容的字典。这就是完整的任务描述。
节点从不决定后续会发生什么。它只需完成自己的工作并更新状态，后续的路由处理则由第5节说明。
需要了解的名称
state是每个节点都会接收的参数，它是一个结构类似AgentState的字典。你可以从中读取诸如state["messages"]或state["current_task"]这样的值。HumanMessage、AIMessage和SystemMessage是你会频繁使用的三种消息类型。前者是用户输入的内容，后者是模型的回复，而后者则用于向模型传递你的指令，即系统提示。invoke是用来调用 LLM 或llm_with_tools以获取答案的方法。你需要传入一组消息，随后会收到一个AIMessage。
AgentState中的字段名称对应。其值为更新内容，而非完整状态，LangGraph会使用您定义的归约函数来合并这些更新内容。模板
# ============================================================
# MODULE 4: NODES
# ============================================================
# ── Node: Agent (the reasoning brain) ───────────────────────
def agent_node(state: AgentState) -> dict:
"""The central reasoning node. Calls the LLM and decides
whether to respond or call a tool."""
# Build the message list to send to the LLM.
# Always include a system message to set behavior.
system_prompt = SystemMessage(content=(
"You are a helpful assistant. Use the available tools "
"when you need real-time information or computation. "
"Respond clearly and concisely."
))
# The LLM receives the system prompt + all previous messages in state
messages_to_send = [system_prompt] + state["messages"]
# Call the LLM. Use llm_with_tools if you have tools; plain llm if not.
response = llm_with_tools.invoke(messages_to_send)
# Return the LLM's response as a state update.
# add_messages will APPEND this AIMessage to state["messages"].
return {"messages": [response]}
# ── Node: Summarizer (example of a non-LLM processing node) ─
def summarize_node(state: AgentState) -> dict:
"""Summarizes the conversation so far to keep context short.
This shows that nodes don't have to call an LLM - they can
do any Python processing."""
all_messages = state["messages"]
# Summarize with the LLM (a different prompt, same LLM)
summary_prompt = [
SystemMessage(content="Summarize the following conversation in 2-3 sentences."),
HumanMessage(content=str(all_messages))
]
summary_response = llm.invoke(summary_prompt)
# Replace messages with a fresh start containing just the summary
return {
"messages": [AIMessage(content=f"[Summary] {summary_response.content}")]
}
节点的图像表示
可以将节点想象成接力赛跑者。它接过代表状态的接力棒，完成自己的跑段即执行相应操作，然后将更新后的接力棒传递给下一位跑者。它不会挑选下一位跑者，那是负责指挥比赛的边所承担的任务。
每个节点的函数签名都相同：def my_node(state: AgentState) -> dict。输入是整个状态，输出则是一个仅包含需要修改的字段的字典。
第5节：边与路由
其作用
边是图中的决策逻辑，它们将节点连接起来并决定执行流程。LangGraph中有两种类型的边。
静态边会在没有任何条件限制的情况下将一个节点与另一个节点相连，例如“在摘要器之后，转到代理节点”。无需决策，只是直接的连接。
条件边会根据当前状态返回一个字符串，用于指定下一个节点的名称。这正是智能体了解自身流程逻辑的关键所在——它应该调用某个工具、结束当前流程，还是重新循环尝试？
需要了解的名称
add_edge(A, B)用于创建静态边，意味着控制流始终从A指向B。add_conditional_edges(source, routing_function, mapping)用于从source节点创建分支。路由函数接收当前状态并返回一个字符串，而mapping字典则将该字符串转换为下一个节点的实际名称。
tools_condition 是来自 langgraph.prebuilt 的现成路由函数。它会检查状态中的最后一条消息，判断是否存在工具调用。如果有，则将路径路由到 "tools"，否则路由到 END。这是使用工具的智能体的标准处理方式，值得牢记。START 是表示图起始点的常量。设置入口点意味着需要从 START 添加一条边连接到第一个节点。模板
# ============================================================
# MODULE 5: EDGES & ROUTING
# ============================================================
# Import the pre-built tool routing function
from langgraph.prebuilt import tools_condition
# ── Custom Routing Function Example ─────────────────────────
def should_continue(state: AgentState) -> Literal["tools", "summarize", "__end__"]:
"""Custom router for the agent node.
Routing functions always:
1. Receive the current state as input
2. Return a string that maps to the next node (or END)
The return values must match the keys in add_conditional_edges' mapping.
"""
last_message = state["messages"][-1] # Look at what the LLM just said
# Case 1: The LLM decided to call a tool
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
# Case 2: The conversation is getting long - summarize before continuing
if len(state["messages"]) > 20:
return "summarize"
# Case 3: The LLM gave a direct answer - we're done
return "__end__" # LangGraph's internal name for END
用于路由的图像
可以将路由函数想象成铁路信号员。当某个节点的处理完成时，列车就会抵达车站。信号员会查看反映当前状态的清单，然后切换相应的道岔，让列车驶入正确的轨道。信号员不会移动列车，那是LangGraph负责的工作。
需要记住的一条规则是：路由函数绝不会修改状态，它们仅负责读取状态。任何更改都应体现在节点中，因此如果在路由过程中想要更新状态，就把相关逻辑放到节点里。
第6节：图结构构建
其作用
所有内容都在这一节整合起来。可以把这一节视为构建阶段。第1至5节属于准备阶段，负责定义材料、切割并摆放各个部件，而现在则通过连接这些部件并进行编译来完成整体结构的构建。
组装过程始终遵循相同的五个步骤，值得牢记：初始化、注册节点、设置入口点、连接边、编译。
需要了解的名称
StateGraph(AgentState)会以你的State类作为架构来创建构建器，因此每个节点都必须返回与其兼容的更新结果。add_node(name, function)用于注册节点。名称是在边中使用的字符串，函数则对应第4节中提到的节点。set_entry_point(name)用于指定调用图时首先运行的节点。add_edge(A, B)用于在A和B之间建立直接的、无条件的连接。add_conditional_edges(source, router, mapping)则用于从源节点添加条件分支，它会使用你定义的路由函数以及将返回的字符串映射到目标节点名称的映射表。
compile(checkpointer=...) 会将图结构封装为可运行的对象（即 CompiledGraph）。之后无法再添加节点或边。可选的 checkpointer 参数可启用持久内存功能。模板
# ============================================================
# MODULE 6: GRAPH ASSEMBLY
# ============================================================
# ── Step 1: Initialize ──────────────────────────────────────
# Always pass your State class to StateGraph
graph_builder = StateGraph(AgentState)
# ── Step 2: Register All Nodes ──────────────────────────────
# Format: add_node("node_name_as_string", node_function)
# The string name is what you use in ALL edge definitions
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node) # The pre-built ToolNode from Module 3
graph_builder.add_node("summarize", summarize_node)
# ── Step 3: Set Entry Point ─────────────────────────────────
# Which node runs first when we invoke the graph?
graph_builder.set_entry_point("agent")
# ── Step 4: Wire the Edges ──────────────────────────────────
# Conditional edge from agent: check if we need tools, a summary, or we're done
graph_builder.add_conditional_edges(
"agent", # Source node
should_continue, # Routing function from Module 5
{
"tools": "tools", # If router returns "tools" → go to tools node
"summarize": "summarize", # If router returns "summarize" → go to summarize node
"__end__": END, # If router returns "__end__" → stop the graph
}
)
# Static edge: after tools run, always go back to agent (the ReAct loop)
graph_builder.add_edge("tools", "agent")
# Static edge: after summarization, always return to agent
graph_builder.add_edge("summarize", "agent")
# ── Step 5: Compile ─────────────────────────────────────────
# Without checkpointer: no persistent memory (stateless per invocation)
# With checkpointer: memory persists across turns (stateful conversations)
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
用于理解的示意图
电路板是最恰当的类比。每次调用add_node都会放置一个组件，而add_edge和add_conditional_edges则负责在组件之间连接导线。compile()会执行最终检查并完成电路板的封装，之后它就准备好通电运行，即通过invoke来执行任务。
该模板包含了基本的ReAct循环：agent → tools → agent → tools，如此反复，直到代理决定任务完成并转向END。这种“推理-行动”循环是几乎所有使用工具的LangGraph代理的底层机制，理解这一机制就能解释你所见的大多数图表结构。
第7节：入口点与调用
其作用
这就是启动平台。之前的所有内容都属于定义阶段，而这一部分则是执行阶段：即运行图结构的代码，负责输入数据的提供与输出结果的收集。
运行它有两种方式：invoke用于单次处理，而流式处理则可实现逐步输出。此外还需要config，即包含运行时参数的字典，其中最重要的就是用于内存管理的thread_id。
需要了解的名称
graph.invoke(input, config)会执行整个图结构并返回最终状态。由于是同步操作，直到所有步骤完成之前都不会有任何反馈。graph.stream(input, config)同样会运行图结构，但会在状态更新发生时立即返回部分结果，非常适合需要展示实时进度的用户界面应用。
config 是传递给任一调用的字典。其键值为 {"configurable": {"thread_id": "some_id"}}。线程 ID 的作用类似于会话 ID，用于告知检查点程序应恢复和存储哪些内存。具有不同线程 ID 的两个用户完全不会共享内存。HumanMessage(content=...) 是在将用户输入交给图结构之前对其进行封装的方式。模板
# ============================================================
# MODULE 7: ENTRYPOINT & INVOCATION
# ============================================================
if __name__ == "__main__":
# ── Config: defines this conversation's memory session ──
# Change thread_id to start a fresh conversation.
# Keep the same thread_id to continue an existing one.
config = {"configurable": {"thread_id": "user-session-001"}}
# ── Single Invocation (synchronous) ─────────────────────
user_input = "What is the current price of Bitcoin?"
result = graph.invoke(
input={"messages": [HumanMessage(content=user_input)]},
config=config
)
# The result is the final state dictionary.
# Access the last message to get the agent's final answer.
final_answer = result["messages"][-1].content
print(f"Agent: {final_answer}")
# ── Streaming Invocation (for real-time output) ──────────
for chunk in graph.stream(
input={"messages": [HumanMessage(content=user_input)]},
config=config,
stream_mode="values" # Yields the full state after each node runs
):
# Each chunk is a state snapshot. The last message shows progress.
latest = chunk["messages"][-1]
if hasattr(latest, "content") and latest.content:
print(f"[Streaming] {latest.content}")
# ── Multi-turn Conversation Loop ─────────────────────────
print("\n--- Starting Interactive Session ---")
while True:
user_text = input("You: ").strip()
if user_text.lower() in ("exit", "quit", "bye"):
break
response = graph.invoke(
input={"messages": [HumanMessage(content=user_text)]},
config=config # Same config = same memory thread
)
print(f"Agent: {response['messages'][-1].content}\n")
完整模板
下方是可直接复制到新项目中的完整最小结构框架。这是一个值得收藏的单一文件：只需用您自己的逻辑替换其中的占位实现，即可得到任何 LangGraph 智能体的基础框架。
# ============================================================
# LANGGRAPH CANONICAL AGENT TEMPLATE
# Modules: Imports → State → Tools → Nodes → Edges → Assembly → Entrypoint
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
from typing import TypedDict, Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# ── MODULE 2: STATE ─────────────────────────────────────────
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# Add your custom fields here
# ── MODULE 3: TOOLS ─────────────────────────────────────────
@tool
def my_tool(input: str) -> str:
"""Describe clearly what this tool does and when the LLM should use it."""
return f"Result for: {input}"
tools = [my_tool]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
# ── MODULE 4: NODES ─────────────────────────────────────────
def agent_node(state: AgentState) -> dict:
"""The reasoning node. Calls the LLM, optionally triggers tool calls."""
messages = [SystemMessage(content="You are a helpful assistant.")] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# ── MODULE 5: EDGES & ROUTING ───────────────────────────────
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
"""Decide: did the LLM call a tool, or did it give a final answer?"""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "__end__"
# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node)
graph_builder.set_entry_point("agent")
graph_builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "__end__": END}
)
graph_builder.add_edge("tools", "agent")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
# ── MODULE 7: ENTRYPOINT ────────────────────────────────────
if __name__ == "__main__":
config = {"configurable": {"thread_id": "session-001"}}
while True:
user_text = input("You: ").strip()
if not user_text or user_text.lower() in ("exit", "quit"):
break
response = graph.invoke(
{"messages": [HumanMessage(content=user_text)]},
config=config
)
print(f"Agent: {response['messages'][-1].content}\n")
更进一步：多智能体系统
当单智能体架构显得自然之后，多智能体系统也就不再令人畏惧。其核心思想是每个智能体都只是一个编译后的图，而这个编译后的图本身又可以成为更大监督图中的一个节点。
多智能体相关的重要概念
- 监督节点是指那些并非直接回应用户需求，而是负责决定“应由哪位专家处理任务”的智能体。它负责在各个子智能体之间进行路由。
- 子图是指作为节点注册在另一张图中的编译后的图。它像普通节点一样接收并返回状态信息。
Command是一种较新的模式，从LangGraph 0.2版本开始支持，该模式下一个节点会在单个值中同时返回状态更新信息与路由指令。在某些多智能体设计中，这种方式无需再使用独立的条件边。结构模板
# ── MULTI-AGENT PATTERN ─────────────────────────────────────
# Each specialist is a compiled graph (a subgraph)
# Sub-agent 1: A researcher
researcher_graph = StateGraph(AgentState)
# ... (built with its own nodes, edges, and tools)
researcher = researcher_graph.compile()
# Sub-agent 2: A writer
writer_graph = StateGraph(AgentState)
# ... (built with its own nodes, edges, and tools)
writer = writer_graph.compile()
# ── SUPERVISOR NODE ─────────────────────────────────────────
def supervisor_node(state: AgentState) -> dict:
"""Decides which sub-agent should handle the current task."""
# The supervisor LLM decides: "researcher" or "writer" or "FINISH"
response = supervisor_llm.invoke(state["messages"])
return {"messages": [response], "next_agent": response.content}
def route_to_agent(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
"""Routes to the appropriate sub-agent based on supervisor's decision."""
return state.get("next_agent", "__end__")
# ── SUPERVISOR GRAPH ────────────────────────────────────────
supervisor_builder = StateGraph(AgentState)
supervisor_builder.add_node("supervisor", supervisor_node)
supervisor_builder.add_node("researcher", researcher) # Subgraph as a node!
supervisor_builder.add_node("writer", writer) # Subgraph as a node!
supervisor_builder.set_entry_point("supervisor")
supervisor_builder.add_conditional_edges(
"supervisor",
route_to_agent,
{"researcher": "researcher", "writer": "writer", "__end__": END}
)
supervisor_builder.add_edge("researcher", "supervisor")
supervisor_builder.add_edge("writer", "supervisor")
supervisor_graph = supervisor_builder.compile(checkpointer=MemorySaver())
多智能体版本与单智能体版本结构相同，也包含相同的七个部分。唯一的区别在于某些“节点”实际上是完整的编译后的图结构。其心智模型可以无缝扩展。
快速参考
这里有一张简化的卡片，您可以将其打印或贴在墙上，以便集中查看核心要点。
- 状态：使用
TypedDict定义状态结构。Annotated[T, reducer]用于为某个字段关联一个归约器。add_messages是标准的消息列表归约器，采用追加而非替换的方式处理数据。operator.add则用于累积列表或数值。 - LLM：
ChatOpenAI或ChatAnthropic代表该类别，始终与名称llm相关联。bind_tools(tools)用于向模型告知可用的工具。invoke(messages)会调用模型并返回响应。 - 消息：
HumanMessage用于传递用户输入，AIMessage用于传递模型输出，而SystemMessage用于传递用户的指令。这些消息始终以列表形式传递给invoke函数。 - 工具：
@tool用于将 Python 函数标记为模型可调用的函数，其文档字符串将作为该函数的描述。ToolNode(tools)会自动执行工具调用。tools_condition用于判断模型是否请求使用某个工具。 - 图结构：
StateGraph(State)用于创建图结构，add_node(name, fn)用于注册节点，set_entry_point(name)用于选择第一个节点，add_edge(A, B)用于建立无条件连接，add_conditional_edges(source, router, mapping)用于根据路由器的返回值建立连接，compile(checkpointer=...)用于完成编译，而END用于标记终点。 - 调用方式：
graph.invoke(input, config)用于返回最终状态，graph.stream(input, config)用于逐步输出更新结果，config = {"configurable": {"thread_id": "..."}}用于标识内存会话，而MemorySaver()则是一种内存中的检查点机制。
使其成为条件反射
重点并非从零开始教授 LangGraph，而是为大脑提供一个可以自动掌握的稳定框架，就如同音乐家会直接使用熟悉的和弦进行，或厨师在开始烹饪前先准备好所有食材。
七个部分——导入、状态、工具、节点、边、组装和入口点——构成了这一框架。每个部分中的名称对应音符，而整个模板则相当于乐谱。
一个简单的聊天机器人和包含数十个节点的流程都可以采用相同的结构。虽然复杂度会增加，但结构本身保持不变，正是这种稳定性使其成为标准。
一种合理的入门方式是从最简单的 ReAct 循环开始：一个智能体节点、一个工具节点，以及一条使用 tools_condition 的条件边。先让这个基础功能正常运行，然后再添加另一个节点，之后再增加另一个智能体。无论规模如何，这种结构都能保持不变。
关于持久化后端、人工干预模式以及部署选项，可查阅 LangGraph 的官方文档。