LangGraph多智能体系统：监督者、子图与发送机制
LangGraph多智能体系统：从单个过载智能体扩展到多个智能体，包括监督者路由、辐射与汇聚结构、子图、发送映射归约以及研究助手功能。
前提条件：这些示例假设你已能使用相关工具和内存构建单智能体LangGraph图，并且知道如何通过
interrupt()函数暂停和恢复运行。后续内容均以此为基础，从探讨为何需要多个智能体，逐步发展到构建完整的多智能体研究助手。
为何要将一个智能体拆分为多个
假设单智能体助手表现良好，但用户开始要求它在同一对话中处理查收邮件、研究主题、撰写报告以及管理任务等多项任务。常见的解决方式是添加更多工具并将更多指令放入系统提示词中。不久之后，系统中就会出现15种工具以及长达2,000字的提示词，而助手的质量则逐渐下降。模型开始混淆不同任务，执行研究任务时却使用邮件工具，执行邮件任务时又使用研究工具。
这就是认知过载，也是多智能体设计存在的主要原因。解决这一问题的方法正是软件工程师们数十年来一直采用的：划分职责。在LangGraph中，这样做可以解决三个独立的问题：
- 认知过载。一个模型需要同时处理太多互不相关的任务。而各司其职的专用智能体则能减轻这种负担。
- 顺序瓶颈。原本可以同时进行的工作不得不依次执行。并行智能体则能降低延迟。
- 复杂性。没人能够理解包含40个节点的图结构。较小的子图可以逐个被理解和测试。
以下的每种模式都是针对上述三个问题中的某一个。
五种架构层级
可以把这种发展过程想象成一段楼梯，每上升一步就会增加一个新的理念。
第一级：监督者。一名监督者代理负责协调各个工作者。虽然仍为单一决策者，但它会分配任务而非亲力亲为，这已使其超越了单纯的工具型聊天机器人。
第二级：并行处理。独立的任务被拆分并同时执行，之后再合并结果。其核心思路是分支扩展与汇聚：多个分支同时运行并将结果传递给后续步骤。这种模式适用于需要从多个来源或子任务中重复处理的场景。
第三级：子图结构。不再使用庞大的单一工作流，而是将每个分支转化为具有独立内部步骤及自身输出结构的较小图。这样一来，复杂部分就更容易构建、测试和理解了。
第4级：动态并行性。采用类映射归约风格的Send模式时，分支数量无法提前确定。工作节点会在运行时动态创建，每个任务、主题或分析人员对应一个节点。这些代理并非预先定义，它们会同时工作，且每个节点负责特定的任务。
第5级：所有方式结合。一个实际的系统会同时运用上述所有方法，正如文末的研究助手示例所示。
一个简单的记忆方法：第1级的工作流程通过一个主管节点处理，第2级让固定数量的工作节点协同工作，第3级允许每个分支构成完整的图结构，而第4级则根据需求动态生成工作节点。
第1级：主管节点模式
这是最简单的多智能体模式，也是最佳的起点。监督者接收用户的请求，然后决定由哪位专家来处理。专家负责执行任务并反馈结果，之后由监督者决定后续行动。
以律师事务所为例。高级合伙人会与客户沟通，了解需求后将工作分配给不同的初级律师——一位负责合同事务，一位负责诉讼，还有一位负责合规事宜。初级律师负责具体工作，而合伙人则会对这些工作进行审核并回复客户。
其流程如下：
User Input
↓
[Supervisor Node] ← decides who handles what
↓ ↓
[Specialist A] [Specialist B] ← do the actual work
↓ ↓
[Supervisor Node] ← reviews results, responds or routes again
↓
Response
状态
关键的设计决策在于next_agent字段。监督节点负责向其中写入数据，而路由器则负责读取这些数据。
# ── MODULE 2: STATE ─────────────────────────────────────────
from typing import TypedDict, Annotated, Literal
from langgraph.graph import MessagesState
class SupervisorState(MessagesState):
# The supervisor writes this field to indicate who should act next.
# "researcher", "writer", "FINISH" are the possible values.
next_agent: str
专业工具
每位专家都有其专属的专用工具，这些工具之间不存在重叠。
# ── MODULE 3: TOOLS ─────────────────────────────────────────
from langchain_core.tools import tool
# Researcher's tools
@tool
def search_web(query: str) -> str:
"""Search the web for factual information, news, or research."""
return f"[Web results for: {query}]"
@tool
def search_academic(query: str) -> str:
"""Search academic papers and journals for scholarly sources."""
return f"[Academic results for: {query}]"
# Writer's tools
@tool
def format_as_report(content: str, title: str) -> str:
"""Format raw content into a structured report with headings."""
return f"# {title}\n\n{content}"
@tool
def check_grammar(text: str) -> str:
"""Check and correct grammar in the provided text."""
return f"[Grammar-checked]: {text}"
# Each specialist gets only its own tools
researcher_tools = [search_web, search_academic]
writer_tools = [format_as_report, check_grammar]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
researcher_llm = llm.bind_tools(researcher_tools)
writer_llm = llm.bind_tools(writer_tools)
节点
接下来是监控节点和专用节点。监控节点的纯文本回复本身即为路由决策。
# ── MODULE 4: NODES ─────────────────────────────────────────
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import ToolNode
# ── Supervisor ────────────────────────────────────────────
SUPERVISOR_PROMPT = """You are a supervisor coordinating a team of specialists.
Given the conversation, decide who should act next.
Your team:
- researcher: finds information, searches web and academic sources
- writer: formats content, writes reports, checks grammar
- FINISH: use this when the task is complete and you have a final answer
Respond with ONLY the name of the next agent: researcher, writer, or FINISH."""
def supervisor_node(state: SupervisorState) -> dict:
"""Reads the conversation history and routes to the right specialist."""
messages = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
response = llm.invoke(messages)
# The supervisor's plain text response IS the routing decision
return {"next_agent": response.content.strip()}
# ── Researcher Specialist ──────────────────────────────────
def researcher_node(state: SupervisorState) -> dict:
"""Specialist that handles all research and information-gathering tasks."""
messages = [
SystemMessage(content="You are a research specialist. Use your search tools to find accurate, thorough information.")
] + state["messages"]
response = researcher_llm.invoke(messages)
return {"messages": [response]}
# ── Writer Specialist ──────────────────────────────────────
def writer_node(state: SupervisorState) -> dict:
"""Specialist that handles all writing, formatting, and editing tasks."""
messages = [
SystemMessage(content="You are a writing specialist. Use your tools to produce well-structured, polished output.")
] + state["messages"]
response = writer_llm.invoke(messages)
return {"messages": [response]}
# Tool execution nodes for each specialist
researcher_tool_node = ToolNode(researcher_tools)
writer_tool_node = ToolNode(writer_tools)
路由
# ── MODULE 5: ROUTING ────────────────────────────────────────
def route_after_supervisor(state: SupervisorState) -> Literal["researcher", "writer", "__end__"]:
"""Read the supervisor's decision and route accordingly."""
decision = state["next_agent"]
if decision == "FINISH":
return "__end__"
return decision # "researcher" or "writer"
def route_researcher(state: SupervisorState) -> Literal["researcher_tools", "supervisor"]:
"""After researcher acts: did it call a tool, or produce a final answer?"""
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "researcher_tools"
return "supervisor" # Research done - report back to supervisor
def route_writer(state: SupervisorState) -> Literal["writer_tools", "supervisor"]:
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "writer_tools"
return "supervisor"
图组装
# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
graph_builder = StateGraph(SupervisorState)
# Register all nodes
graph_builder.add_node("supervisor", supervisor_node)
graph_builder.add_node("researcher", researcher_node)
graph_builder.add_node("researcher_tools", researcher_tool_node)
graph_builder.add_node("writer", writer_node)
graph_builder.add_node("writer_tools", writer_tool_node)
# Entry: always start at supervisor
graph_builder.add_edge(START, "supervisor")
# Supervisor routes to specialist or ends
graph_builder.add_conditional_edges(
"supervisor",
route_after_supervisor,
{"researcher": "researcher", "writer": "writer", "__end__": END}
)
# Researcher loop: tools → researcher → supervisor
graph_builder.add_conditional_edges(
"researcher", route_researcher,
{"researcher_tools": "researcher_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("researcher_tools", "researcher")
# Writer loop: tools → writer → supervisor
graph_builder.add_conditional_edges(
"writer", route_writer,
{"writer_tools": "writer_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("writer_tools", "writer")
graph = graph_builder.compile(checkpointer=MemorySaver())
在调度过程中，关键在于返回给监督者的信息。当某个专家完成任务后，会向监督者汇报。监督者会查看整个对话过程，然后从三种操作中选择一种：将任务交给另一位专家、让同一位专家继续处理，或直接结束任务。这其实就是调度的核心所在。
第二级：具有分支与汇聚特性的并行化
监督者模式是顺序式的，一次只有一名专家在处理任务。而许多实际任务都包含可以同时运行的独立子任务，并行化正是为了解决延迟问题。
烹饪是个很好的类比。没人会先做好沙拉，再开始煮意大利面，最后才做酱料。我们会同时准备沙拉、烧水以及制作酱料，这样晚餐就能更快做好。
在 LangGraph 中，具有多条出边 的节点会形成并行分支。LangGraph 会检测到这种分支结构，并以它所称的超级步骤同时处理所有目标节点。在后续任何节点开始处理之前，这些结果会被收集起来，这就是所谓的入边过程。
有一条明确的规则：当同一个超级步骤中的多个节点向同一个状态字段写入数据时，LangGraph 必须知道如何合并这些写入操作。如果没有合并器，系统会抛出错误；而使用 operator.add 则可以安全地拼接这些结果。
# ── MODULE 2: STATE FOR PARALLEL PATTERN ────────────────────
import operator
from typing import Annotated
class ResearchState(TypedDict):
query: str
# Without Annotated + operator.add, parallel writes to 'context' would CRASH.
# With it, LangGraph concatenates results from all parallel nodes.
context: Annotated[list[str], operator.add]
final_answer: str
一个并行研究示例使用了两个数据源：
# ── MODULE 4: PARALLEL WORKER NODES ─────────────────────────
def web_search_node(state: ResearchState) -> dict:
"""Runs in parallel with academic_search_node."""
result = f"[WEB] Results for: {state['query']}"
return {"context": [result]} # Returns a LIST - operator.add will concatenate
def academic_search_node(state: ResearchState) -> dict:
"""Runs in parallel with web_search_node."""
result = f"[ACADEMIC] Papers about: {state['query']}"
return {"context": [result]} # Also a LIST
def synthesize_node(state: ResearchState) -> dict:
"""Runs AFTER both parallel nodes complete (fan-in point).
state['context'] already contains results from BOTH parallel nodes."""
combined = "\n".join(state["context"])
prompt = [
SystemMessage(content="Synthesize the following research into a clear answer."),
HumanMessage(content=f"Research:\n{combined}\n\nQuery: {state['query']}")
]
response = llm.invoke(prompt)
return {"final_answer": response.content}
# ── MODULE 6: PARALLEL GRAPH ASSEMBLY ───────────────────────
from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(ResearchState)
graph_builder.add_node("web_search", web_search_node)
graph_builder.add_node("academic_search", academic_search_node)
graph_builder.add_node("synthesize", synthesize_node)
# Fan-out: START branches to BOTH search nodes simultaneously
graph_builder.add_edge(START, "web_search")
graph_builder.add_edge(START, "academic_search")
# Fan-in: BOTH search nodes must complete before synthesize runs
graph_builder.add_edge("web_search", "synthesize")
graph_builder.add_edge("academic_search", "synthesize")
graph_builder.add_edge("synthesize", END)
parallel_graph = graph_builder.compile()
“扇出”指的是从同一个起点 START 出发的两条边。LangGraph 会同时处理这两条边的目标节点。而“扇入”也会自动发生，因为只有当两条输入边都满足条件，即两个并行节点都处理完成后，synthesize 才会执行一次。
第三级：用于封装的子图
扁平的并行节点适用于简单场景。当每个分支都很复杂，包含多个步骤、各自的工具和逻辑时，子图就是更合适的工具。子图是一个完全编译好的 StateGraph，作为父图中的一个节点来运行。
可以将其想象成公司中的各个部门。CEO（即父图）会将任务分配给工程部和市场部。每个部门都有内部的流程和会议，但 CEO 只负责分配任务并接收成果。
父节点与子图通过共享的状态键进行通信。当子图启动时，会传递父节点状态与子图状态中都存在的所有键，而当子图结束时会将这些键传回。
Parent State: {topic, cleaned_data, report_A, report_B, processed_ids}
↓
Subgraph A State: {cleaned_data, internal_step_1, report_A, processed_ids}
← reads 'cleaned_data' from parent
→ writes 'report_A' and 'processed_ids' back to parent
Subgraph B State: {cleaned_data, internal_step_2, report_B, processed_ids}
← reads 'cleaned_data' from parent
→ writes 'report_B' and 'processed_ids' back to parent
仅存在于子图中的键，如上面的internal_step_1，会保持私密状态，其父节点永远无法看到它们。
完整的实现如下：
# ── MODULE 2: STATE SCHEMAS (Three Levels) ──────────────────
# Parent state - what the orchestrating graph tracks
class ParentState(TypedDict):
raw_input: str
cleaned_data: str # passed INTO both subgraphs
failure_summary: str # written back by subgraph A
performance_report: str # written back by subgraph B
processed_ids: Annotated[list[int], operator.add] # BOTH subgraphs write this
# Subgraph A's internal state - includes its private fields + shared keys
class FailureAnalysisState(TypedDict):
cleaned_data: str # shared input from parent
failures: list[str] # private intermediate state - parent never sees this
failure_summary: str # shared output to parent
processed_ids: Annotated[list[int], operator.add] # shared output (accumulated)
# Subgraph A's output schema - filters which keys return to parent
# Without this, ALL keys return, which can cause conflicts
class FailureAnalysisOutput(TypedDict):
failure_summary: str # only these two keys go back to parent
processed_ids: Annotated[list[int], operator.add]
# Subgraph B's internal state
class PerformanceAnalysisState(TypedDict):
cleaned_data: str
metrics: list[str] # private
performance_report: str # shared output
processed_ids: Annotated[list[int], operator.add]
class PerformanceAnalysisOutput(TypedDict):
performance_report: str
processed_ids: Annotated[list[int], operator.add]
# ── SUBGRAPH A: Failure Analysis ─────────────────────────────
def extract_failures(state: FailureAnalysisState) -> dict:
failures = [f"Error in {state['cleaned_data']}: timeout"]
return {"failures": failures, "processed_ids": [1]}
def summarize_failures(state: FailureAnalysisState) -> dict:
summary = f"Found {len(state['failures'])} failure(s): {state['failures']}"
return {"failure_summary": summary}
failure_builder = StateGraph(FailureAnalysisState, output=FailureAnalysisOutput)
failure_builder.add_node("extract_failures", extract_failures)
failure_builder.add_node("summarize_failures", summarize_failures)
failure_builder.add_edge(START, "extract_failures")
failure_builder.add_edge("extract_failures", "summarize_failures")
failure_builder.add_edge("summarize_failures", END)
failure_subgraph = failure_builder.compile() # compiled → can be used as a node
# ── SUBGRAPH B: Performance Analysis ────────────────────────
def collect_metrics(state: PerformanceAnalysisState) -> dict:
metrics = [f"Latency: 120ms", "Uptime: 99.8%"]
return {"metrics": metrics, "processed_ids": [2]}
def write_report(state: PerformanceAnalysisState) -> dict:
report = f"Performance Report:\n" + "\n".join(state["metrics"])
return {"performance_report": report}
performance_builder = StateGraph(PerformanceAnalysisState, output=PerformanceAnalysisOutput)
performance_builder.add_node("collect_metrics", collect_metrics)
performance_builder.add_node("write_report", write_report)
performance_builder.add_edge(START, "collect_metrics")
performance_builder.add_edge("collect_metrics", "write_report")
performance_builder.add_edge("write_report", END)
performance_subgraph = performance_builder.compile()
# ── PARENT GRAPH ─────────────────────────────────────────────
def clean_data_node(state: ParentState) -> dict:
return {"cleaned_data": state["raw_input"].strip().lower()}
def combine_reports_node(state: ParentState) -> dict:
combined = f"{state['failure_summary']}\n\n{state['performance_report']}"
print(f"Final combined report:\n{combined}")
return {}
parent_builder = StateGraph(ParentState)
parent_builder.add_node("clean_data", clean_data_node)
parent_builder.add_node("failure_analysis", failure_subgraph) # subgraph as node
parent_builder.add_node("performance_analysis", performance_subgraph) # subgraph as node
parent_builder.add_node("combine_reports", combine_reports_node)
parent_builder.add_edge(START, "clean_data")
# Fan-out to both subgraphs (parallel execution)
parent_builder.add_edge("clean_data", "failure_analysis")
parent_builder.add_edge("clean_data", "performance_analysis")
# Fan-in: combine only after both subgraphs complete
parent_builder.add_edge("failure_analysis", "combine_reports")
parent_builder.add_edge("performance_analysis", "combine_reports")
parent_builder.add_edge("combine_reports", END)
parent_graph = parent_builder.compile()
本例中的输出架构 FailureAnalysisOutput 和 PerformanceAnalysisOutput 是许多教程都会跳过的部分，也正是因此人们才会遇到状态键冲突的问题。它能够精确筛选子图向父图暴露的字段，未包含在内的内容则被视为私有的内部状态。
第4级：Send API与map-reduce
并行分支和子图适用于在设计阶段就已知道分支数量的情况。但如果用户今天希望研究5个主题，明天又希望研究20个呢？硬编码20个分支显然是不可行的。这时就需要借助由 Send 支持的map-reduce模式了。
想象一下由出版商来划分章节的情况。一名编辑收到手稿后，将其拆分成若干章节，并同时将每一章交给不同的校对员。无论章节是两个还是二十个，就需要雇佣相应数量的校对员。所有校对工作完成后，这些成果会被整合成最终的书籍。并行工作的数量是在运行时决定的，而非设计阶段。
有三个关键点：
Send(node_name, state)函数来自langgraph.constants，它会创建一个带有自身状态数据的节点实例。一个返回这些节点列表的路由函数会让 LangGraph 同时启动所有节点，每个节点拥有独立的状态。- 收集字段上的
operator.add函数就是归约步骤。随着各个工作节点完成任务，它们的结果会逐步累积到父状态中的列表中。
以下是完整的映射还原实现：
# ── MODULE 2: MAP-REDUCE STATE ──────────────────────────────
from langgraph.constants import Send
class OverallState(TypedDict):
topic: str
subjects: list[str] # populated by the map step
jokes: Annotated[list[str], operator.add] # accumulated by the reduce step
best_joke: str
class JokeState(TypedDict):
"""The private state each parallel worker gets.
Notice: this does NOT need to match OverallState."""
subject: str
# ── MODULE 4: MAP-REDUCE NODES ──────────────────────────────
def generate_subjects(state: OverallState) -> dict:
"""MAP step 1: Expands the topic into a list of subjects to process."""
prompt = [
SystemMessage(content="Generate a list of 3 sub-topics related to the given topic. Return as comma-separated values only."),
HumanMessage(content=state["topic"])
]
response = llm.invoke(prompt)
subjects = [s.strip() for s in response.content.split(",")]
return {"subjects": subjects}
def generate_joke(state: JokeState) -> dict:
"""MAP step 2: Worker node. Each instance handles exactly ONE subject.
Receives its own isolated JokeState - not the full OverallState."""
prompt = [HumanMessage(content=f"Write a short, funny joke about: {state['subject']}")]
response = llm.invoke(prompt)
# Returns a LIST - operator.add in OverallState will accumulate these
return {"jokes": [response.content]}
def pick_best_joke(state: OverallState) -> dict:
"""REDUCE step: All jokes are now in state['jokes']. Pick the winner."""
jokes_text = "\n".join([f"{i}. {j}" for i, j in enumerate(state["jokes"])])
prompt = [
SystemMessage(content="You are a comedy judge. Pick the funniest joke from the list below. Reply with ONLY the number."),
HumanMessage(content=jokes_text)
]
response = llm.invoke(prompt)
winning_index = int(response.content.strip())
return {"best_joke": state["jokes"][winning_index]}
# ── MODULE 5: THE SEND ROUTER ────────────────────────────────
def continue_to_jokes(state: OverallState):
"""The routing function that LAUNCHES parallel workers.
Returning a LIST of Send objects (not a string) is what triggers map-reduce.
Each Send creates an independent, parallel execution of 'generate_joke'
with its own state payload. The number of parallel workers is determined
at runtime by how many subjects exist in state.
"""
return [
Send("generate_joke", {"subject": subject})
for subject in state["subjects"]
]
# ── MODULE 6: MAP-REDUCE GRAPH ASSEMBLY ─────────────────────
graph_builder = StateGraph(OverallState)
graph_builder.add_node("generate_subjects", generate_subjects)
graph_builder.add_node("generate_joke", generate_joke)
graph_builder.add_node("pick_best_joke", pick_best_joke)
graph_builder.add_edge(START, "generate_subjects")
# This conditional edge uses the Send router - it fans out dynamically
graph_builder.add_conditional_edges(
"generate_subjects",
continue_to_jokes, # returns a LIST of Send objects
["generate_joke"] # list of possible destination nodes (for graph validation)
)
# Fan-in: pick_best_joke runs after ALL parallel generate_joke instances complete
graph_builder.add_edge("generate_joke", "pick_best_joke")
graph_builder.add_edge("pick_best_joke", END)
map_reduce_graph = graph_builder.compile()
关键在于 continue_to_jokes。它返回的是 Send 对象的列表而非字符串，当 LangGraph 从路由函数接收到这样的列表时，会立即启动列表中的每一个任务。state["subjects"] 中的三个条目意味着有三个并行的 generate_joke 节点，二十个条目则意味着有二十个节点。整个图的结构会随着运行时产生的任务数量而扩展。
第5级：多智能体研究助手
现在，监督式协调、人工审核、并行子图以及 Send API 等所有功能都整合在一个生产级系统中。这是 LangGraph Academy 课程的最终项目，相关说明见此处。
工作流程如下：用户提出研究主题，系统会生成一组人工智能分析师角色，由人工进行审核或优化。随后每个分析师作为并行访谈子代理，通过网络搜索向人工智能“专家”提问，并生成报告的其中一个部分。最后，由写作节点将这些部分整合成完整的文档。
User provides topic
↓
[create_analysts] → LLM generates N analyst personas
↓
[human_feedback] ← INTERRUPT: human approves or sends feedback
↓ ↑
[should_continue] → if feedback, loop back to create_analysts
↓
[initiate_research] → Send API: spawns N parallel interview subgraphs
↓↓↓ (N parallel workers running simultaneously)
[interview_subgraph × N] → each analyst interviews an "expert" via web search
↓
[write_report] → LLM compiles all sections into a final report
三级状态
from pydantic import BaseModel, Field
# ── Analyst persona - a Pydantic model for structured LLM output ──
class Analyst(BaseModel):
name: str = Field(description="The analyst's full name")
role: str = Field(description="Their professional role (e.g. 'Financial Analyst')")
focus: str = Field(description="Their specific analytical focus area")
@property
def persona(self) -> str:
return f"Name: {self.name}\nRole: {self.role}\nFocus: {self.focus}"
class Perspectives(BaseModel):
analysts: list[Analyst]
# ── Outer graph state ─────────────────────────────────────────
class ResearchGraphState(TypedDict):
topic: str
max_analysts: int
human_analyst_feedback: str # human writes here during HITL pause
analysts: list[Analyst]
sections: Annotated[list, operator.add] # accumulated from all parallel interviews
final_report: str
# ── Interview subgraph state ──────────────────────────────────
class InterviewState(MessagesState):
# inherits: messages: Annotated[list[BaseMessage], add_messages]
max_num_turns: int
context: Annotated[list, operator.add] # search results accumulate here
analyst: Analyst
interview: str # formatted transcript
sections: list # completed section (sent back to outer graph)
生成分析结果，并进行人工核查
# ── ANALYST GENERATION NODE ───────────────────────────────────
ANALYST_PROMPT = """You are generating a team of {max_analysts} research analysts
for the topic: {topic}.
{feedback_section}
Each analyst should have a unique perspective and area of focus.
Generate analysts with diverse viewpoints - financial, technical, social, etc."""
def create_analysts(state: ResearchGraphState) -> dict:
"""Generates analyst personas using structured LLM output."""
feedback = state.get("human_analyst_feedback", "")
feedback_section = (
f"User feedback to incorporate:\n{feedback}" if feedback
else "No feedback yet - generate your best initial set."
)
prompt = ANALYST_PROMPT.format(
max_analysts=state["max_analysts"],
topic=state["topic"],
feedback_section=feedback_section
)
# with_structured_output: forces LLM to return a Pydantic model
# instead of free-form text - ensures clean, usable data
structured_llm = llm.with_structured_output(Perspectives)
result = structured_llm.invoke([SystemMessage(content=prompt)])
return {"analysts": result.analysts}
def human_feedback_node(state: ResearchGraphState) -> dict:
"""No-op node. Its only purpose is to be a pause point.
The graph is compiled with interrupt_before=['human_feedback'].
See Part 3 for the full HITL pattern."""
pass # Does nothing - the pause happens BEFORE this node via interrupt_before
def should_continue_to_research(state: ResearchGraphState) -> Literal["create_analysts", "initiate_research"]:
"""After human feedback: loop back to refine, or proceed to interviews."""
if state.get("human_analyst_feedback"):
return "create_analysts"
return "initiate_research"
访谈子图
每个分析器都作为独立的子图运行，即一个小型ReAct智能体，它会对“专家”（另一个承担该角色的大语言模型）进行访谈，并在网络上搜索信息。
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
@tool
def web_search_tool(query: str) -> str:
"""Search the web for information on a topic."""
return f"[Search results for '{query}']" # Replace with real search API
interview_tools = [web_search_tool]
interview_llm = llm.bind_tools(interview_tools)
def generate_question(state: InterviewState) -> dict:
"""The analyst node: generates the next interview question."""
analyst = state["analyst"]
messages = [
SystemMessage(content=(
f"You are {analyst.name}, a {analyst.role} focused on {analyst.focus}. "
f"You are interviewing an expert about the research topic. "
f"Ask insightful questions that align with your analytical focus. "
f"When you have enough information, say 'Thank you, that is all I needed.'"
))
] + state["messages"]
response = interview_llm.invoke(messages)
return {"messages": [response]}
def generate_answer(state: InterviewState) -> dict:
"""The expert node: answers the analyst's question, using web search."""
messages = [
SystemMessage(content=(
"You are an expert being interviewed. Answer thoroughly and factually. "
"Use the web search tool when you need current data."
))
] + state["messages"]
response = interview_llm.invoke(messages)
return {"messages": [response]}
def save_interview(state: InterviewState) -> dict:
"""Formats the full Q&A exchange into a clean transcript string."""
transcript = []
for msg in state["messages"]:
if hasattr(msg, "content"):
role = "Analyst" if type(msg).__name__ == "HumanMessage" else "Expert"
transcript.append(f"{role}: {msg.content}")
return {"interview": "\n\n".join(transcript)}
def write_section(state: InterviewState) -> dict:
"""The final node: uses the interview transcript to write a report section."""
analyst = state["analyst"]
prompt = [
SystemMessage(content=(
f"Based on the following interview conducted by {analyst.name} ({analyst.role}), "
f"write a structured report section focused on {analyst.focus}. "
f"Be concise, factual, and cite specific points from the interview."
)),
HumanMessage(content=state["interview"])
]
response = llm.invoke(prompt)
# sections is the key that Send passes back to the outer graph
return {"sections": [response.content]}
def route_interview(state: InterviewState) -> Literal["generate_answer", "save_interview"]:
"""Continue interviewing OR wrap up when analyst is satisfied or max turns hit."""
last_message = state["messages"][-1]
if (
"thank you, that is all" in last_message.content.lower()
or len(state["messages"]) >= state.get("max_num_turns", 6) * 2
):
return "save_interview"
# Check if analyst made a tool call (web search)
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "generate_answer"
return "generate_answer"
# ── INTERVIEW SUBGRAPH ASSEMBLY ───────────────────────────────
interview_builder = StateGraph(InterviewState)
interview_tool_node = ToolNode(interview_tools)
interview_builder.add_node("generate_question", generate_question)
interview_builder.add_node("generate_answer", generate_answer)
interview_builder.add_node("interview_tools", interview_tool_node)
interview_builder.add_node("save_interview", save_interview)
interview_builder.add_node("write_section", write_section)
interview_builder.add_edge(START, "generate_question")
interview_builder.add_conditional_edges(
"generate_question",
route_interview,
{"generate_answer": "generate_answer", "save_interview": "save_interview"}
)
interview_builder.add_conditional_edges(
"generate_answer",
lambda s: "interview_tools" if (hasattr(s["messages"][-1], "tool_calls") and s["messages"][-1].tool_calls) else "generate_question",
{"interview_tools": "interview_tools", "generate_question": "generate_question"}
)
interview_builder.add_edge("interview_tools", "generate_answer")
interview_builder.add_edge("save_interview", "write_section")
interview_builder.add_edge("write_section", END)
interview_graph = interview_builder.compile()
外部编排图
# ── SEND ROUTER: Spawn one interview per analyst ─────────────
def initiate_research(state: ResearchGraphState):
"""Launches N parallel interview subgraphs via Send — one per analyst.
This is the map-reduce Send pattern from Level 4, applied to subgraphs."""
return [
Send("interview", {
"analyst": analyst,
"messages": [HumanMessage(content=f"Research topic: {state['topic']}")],
"context": [],
"max_num_turns": 3,
"interview": "",
"sections": []
})
for analyst in state["analysts"]
]
# ── REPORT COMPILATION NODE ───────────────────────────────────
def write_report(state: ResearchGraphState) -> dict:
"""Fan-in point: all sections have arrived from parallel interviews.
The LLM compiles them into a final report."""
sections_text = "\n\n---\n\n".join(state["sections"])
prompt = [
SystemMessage(content=(
"You are a senior editor. Compile the following report sections written by "
"different analysts into one cohesive, well-structured final report. "
"Add a proper introduction and conclusion. Do not repeat content."
)),
HumanMessage(content=f"Topic: {state['topic']}\n\nSections:\n{sections_text}")
]
response = llm.invoke(prompt)
return {"final_report": response.content}
# ── OUTER GRAPH ASSEMBLY ─────────────────────────────────────
outer_builder = StateGraph(ResearchGraphState)
outer_builder.add_node("create_analysts", create_analysts)
outer_builder.add_node("human_feedback", human_feedback_node)
outer_builder.add_node("interview", interview_graph) # subgraph as node
outer_builder.add_node("write_report", write_report)
outer_builder.add_edge(START, "create_analysts")
outer_builder.add_edge("create_analysts", "human_feedback")
outer_builder.add_conditional_edges(
"human_feedback",
should_continue_to_research,
{"create_analysts": "create_analysts", "initiate_research": "initiate_research"}
)
# initiate_research is a ROUTING FUNCTION (returns Send objects), not a node
outer_builder.add_conditional_edges(
"human_feedback",
initiate_research, # returns list of Send objects
["interview"] # validation: lists possible destinations
)
outer_builder.add_edge("interview", "write_report")
outer_builder.add_edge("write_report", END)
research_graph = outer_builder.compile(
checkpointer=MemorySaver(),
interrupt_before=["human_feedback"] # HITL pause for analyst approval
)
运行整个程序
if __name__ == "__main__":
config = {"configurable": {"thread_id": "research-001"}}
# Step 1: Generate analysts
result = research_graph.invoke(
{
"topic": "The impact of AI on the future of software engineering",
"max_analysts": 3,
"human_analyst_feedback": "",
"analysts": [],
"sections": [],
"final_report": ""
},
config=config
)
# Step 2: HITL — graph paused at human_feedback node
snapshot = research_graph.get_state(config)
analysts = snapshot.values["analysts"]
print("Generated analysts:")
for a in analysts:
print(f" - {a.name} ({a.role}): {a.focus}")
feedback = input("\nProvide feedback to refine analysts (or press Enter to approve): ")
if feedback.strip():
# Human has feedback — inject it and let supervisor regenerate
research_graph.update_state(
config,
{"human_analyst_feedback": feedback},
as_node="human_feedback"
)
else:
# Approved — clear feedback and proceed to interviews
research_graph.update_state(
config,
{"human_analyst_feedback": None},
as_node="human_feedback"
)
# Step 3: Resume — N parallel interviews launch via Send, then report is compiled
print("\nRunning parallel interviews...")
final_result = research_graph.invoke(None, config=config)
print("\n" + "="*60)
print("FINAL RESEARCH REPORT")
print("="*60)
print(final_result["final_report"])
选择模式
- 会在任务之间混淆的单个代理需要使用监督者模式。监督者负责路由，而专家节点负责执行任务。
- 独立且处理速度较慢的任务适合使用扇出和扇入模式。这些任务可同时运行，对于共享字段可使用
operator.add还原器来处理。 - 本身较为复杂的并行分支需要使用子图模式。将每个分支编译为独立的图，将其作为节点嵌入，并通过输出架构来控制返回给父节点的内容。
- 运行时决定工作节点数量的情况适合使用
Send模式。路由器会生成一系列Send对象，每个对象都会创建一个拥有独立状态的独立工作节点。
关键词参考
监督者结构。 supervisor_node是协调器，它负责读取状态、写入next_agent，而不执行任何专业性工作。specialist_node则使用自身的工具和提示语来处理某一特定任务，处理完成后再将结果反馈给监督者。next_agent: str是用于路由的传统状态字段：监督者写入名称，路由器则读取该名称。
并行性。“扇出”指的是从一个源节点发出多个add_edge调用，LangGraph会同时执行这些调用。“扇入”则是指多条边指向同一个目标节点，此时需要等待所有源节点的处理完成。对于有多个并行节点写入的字段，必须使用Annotated[list, operator.add]，因为否则并行写入会导致程序崩溃，而operator.add则是常用的归约函数，用于合并来自不同节点的列表。
子图。 StateGraph(InternalState, output=OutputSchema) 是一种带有两个参数的实现方式，第二个参数用于筛选哪些键会返回给父节点。输出模式是一个仅包含需要公开键值的 TypedDict，其余内容均保持私有。重叠的键值充当父节点与子节点之间的通道，因为两个模式中的任何公共键值都会自动被共享。subgraph.compile() 会将子图封装为一个可调用对象，随后将其传递给 parent.add_node("name", compiled_subgraph)。调用 graph.get_graph(xray=1).draw_mermaid() 即可在图表中显示内部的子图结构。
Map-reduce。 Send(node_name, state_dict) 表示一次并行工作进程的调用，其传递的数据无需与父节点的状态一致。路由器返回 [Send(...), Send(...)] 时会同时启动这些调用。使用 Send 时，add_conditional_edges 的第三个参数是一个有效目标节点名称的列表而非字典，仅用于图结构验证。最后，with_structured_output(PydanticModel) 会在模型调用后使用，以确保生成的 JSON 符合 Pydantic 模式要求。这是获取可靠的管理决策结果与分析人员列表的常用方法。
组合能力才是关键
从单个智能体发展到多个智能体其实并非质的飞跃，只不过是相同的构建模块被反复运用并相互连接而已。多智能体系统中的每个智能体都是一张图，即一个节点或子图，而这些基本元素本身从未改变。关键在于掌握如何将它们组合起来。
这里的设计思路是经过精心规划的：首先是一个监督节点，接着是扁平的并行节点，对于复杂分支则使用并行子图，再通过Send指令驱动动态工作节点，最后将所有部分整合在一起。如果某个步骤使用起来感觉自然顺畅，那就说明设计达到了预期效果，因为每一层都清晰地建立在上一层的基础之上。
5级的研究助手功能已接近实际生产环境中的多智能体代码库所具备的功能：人工审核流程、并行处理的专家模块、结构化的模型输出、动态的任务分配机制以及最终的整合步骤。再加上标准的项目结构、内存管理以及人工干预的安全机制，这些要素共同构成了大多数真实LangGraph应用背后的架构。
相关阅读
- LangGraph中的自修复多智能体团队：架构师、测试员、调试器 — 用Python构建一个LangGraph状态机，让架构师、开发人员、pytest沙箱环境、调试器和审核员协同工作并循环迭代，直至测试通过。