大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。

 我们正在讨论使用 LangGraph 构建 Multi - Agent架构。根据提供的引用，我们知道 LangGraph利用图结构来定义代理，其中每个代理是图中的一个节点，边表示代理之间的连接。代理通过接收其他代理的输入并将控制权传递给下一个代理来执行操作。

引用[1]提到，在LangGraph框架中，主要通过几种方法建立各个子代理之间的通信连接。引用[2]则给出了一个代码片段，展示了如何创建一个研究代理（research_agent），该代理使用了一个工具（tavily_tool）和一个系统消息。

因此，为了构建Multi-Agent架构，我们需要以下步骤：

1. 创建多个代理（每个代理是一个节点）。
2. 定义这些代理之间的连接（边），即控制流。
3. 定义整个图的结构，包括入口点、状态传递等。

下面我们将逐步说明构建方法，并提供代码示例。

步骤1：创建代理 每个代理通常是一个函数或一个可调用对象，它接收当前状态，执行一些操作（如调用语言模型、使用工具等），然后返回更新后的状态。我们可以使用create_agent函数（如引用[2]所示）来创建代理，但具体实现可能因项目而异。

假设我们有一个create_agent函数，它接受一个语言模型（llm）、一个工具列表和一个系统消息，并返回一个代理函数。例如：

步骤2：定义图 LangGraph提供了StateGraph类来构建图。我们首先定义状态的结构，然后添加节点（每个节点对应一个代理），再添加边（定义节点之间的转移），最后设置入口点和条件边（用于分支）。

步骤3：编译图 最后，我们将图编译成一个可执行的对象。

下面是一个示例代码，展示如何使用LangGraph构建一个包含两个代理（研究代理和写作代理）的Multi-Agent系统：

注意：以下代码示例需要安装LangGraph和必要的依赖（如langchain, openai等）。同时，我们假设已经定义了所需的工具和语言模型。

首先，我们导入必要的模块：

from langgraph.graph import StateGraph from typing import TypedDict, List, Annotated import operator 

定义状态（State）： 状态是一个共享的数据结构，在代理之间传递。我们使用TypedDict来定义状态，其中包含我们需要的字段。

class AgentState(TypedDict): # 输入问题 question: str # 研究代理得到的结果 research_result: str # 写作代理生成的最终答案 final_answer: str 

接下来，创建两个代理： 假设我们已经有一个create_agent函数，它返回一个函数，该函数接收状态并返回更新后的状态（或一个包含更新字段的字典）。由于引用[2]中提到了create_agent，我们可以参考其用法。

这里我们简化代理的创建，直接定义两个代理函数（在实际中，代理内部可能会调用LLM和工具）：

def research_agent(state: AgentState): # 这里模拟研究代理的行为：使用工具搜索信息 # 在实际应用中，这里可能调用Tavily搜索工具 # 假设我们有一个函数tavily_search(question)返回结果 research_result = tavily_search(state['question']) # 返回更新后的状态（只更新research_result字段） return {"research_result": research_result} def writing_agent(state: AgentState): # 写作代理根据研究结果生成最终答案 # 这里我们调用LLM来生成回答 prompt = f"基于以下研究结果，回答问题：{state['question']} 研究结果：{state['research_result']}" final_answer = llm.invoke(prompt).content return {"final_answer": final_answer} 

注意：上述代码中的tavily_search和llm需要事先定义。在实际项目中，代理的创建可能更复杂，需要包装成与LangGraph兼容的节点。

然后，我们构建图：

# 初始化图，指定状态类型 graph = StateGraph(AgentState) # 添加节点 graph.add_node("research", research_agent) graph.add_node("write", writing_agent) # 设置入口点：从研究代理开始 graph.set_entry_point("research") # 添加边：从research节点到write节点 graph.add_edge("research", "write") graph.add_edge("write", END) # END是一个特殊的结束节点 # 编译图 app = graph.compile() 

以上是一个简单的线性流程：研究代理 -> 写作代理 -> 结束。

但是，在实际应用中，我们可能需要条件转移。例如，如果研究代理得到的结果已经足够回答，那么可能不需要写作代理再加工（或者需要其他代理处理）。这时我们可以使用条件边。

修改状态定义，增加一个需要判断的字段（例如，是否需要进一步写作）？或者我们可以通过节点函数的返回值来判断。

假设我们修改研究代理，使其返回一个包含两个字段的字典：research_result和direct_answer（如果可以直接回答则设置direct_answer为答案，否则为None）。然后我们修改状态：

由于时间关系，我们不再展开条件边的设置。但基本思路是：在节点之间添加条件边，然后定义条件函数（根据当前状态决定下一个节点）。

另一个常见的需求是循环（例如，一个代理的输出作为另一个代理的输入，然后循环直到满足条件）。LangGraph支持循环图。

总结一下，使用LangGraph构建Multi-Agent架构的关键步骤：

1. 定义状态（一个字典类型，包含所有代理需要共享的数据）
2. 创建代理（节点），每个代理是一个函数，接收状态并返回更新后的状态（部分字段）
3. 构建图：添加节点，设置入口点，添加边（包括条件边）和循环
4. 编译图，得到一个可执行的应用程序

注意：上面的代码示例是简化版，实际应用中需要根据具体需求调整。例如，LangGraph官方文档中有更详细的示例。

由于引用[1]和[2]提供的信息有限，我们只能给出一个基本框架。建议参考LangGraph官方文档获取更详细的代码示例和高级特性。

最后，我们生成相关问题：

§§相关问题§§

1. 在LangGraph中如何添加条件分支？
2. 如何定义LangGraph的状态（State）以支持多个代理之间的数据传递？
3. 在Multi-Agent架构中，如何处理代理之间的循环调用？
4. LangGraph支持哪些类型的边（Edge）？

   使用 LangGraph 构建 Multi-Agent 架构的方法及代码示例

   LangGraph 通过图结构实现 Multi-Agent 协作，核心流程包括：节点定义（代理）+ 边连接（通信）+ 状态管理。以下是具体方法和代码示例：

---

1. 核心架构方法

- 节点设计
每个代理封装为独立节点，包含： - LLM 实例：如 OpenAI GPT-4 - 工具集：搜索、代码执行等工具 - 系统提示：定义代理的角色和任务

 research_agent = create_agent( llm=OpenAI(model="gpt-4"), tools=[tavily_search_tool], # 搜索工具 system_message="你是一个研究代理，负责收集信息" ) 

- 通信机制
通过边连接代理，传递控制权和数据： - 显式边：直接指定下一个节点 - 条件边：根据输出动态路由

 graph.add_edge("research_agent", "writing_agent") # 研究代理 → 写作代理 graph.add_conditional_edges("writing_agent", decide_next_node) # 动态路由 

- 状态管理
使用共享状态对象传递数据：

 class AgentState(TypedDict): question: str research_data: dict final_answer: str 

---

2. 完整代码示例

from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, Dict, Union from langchain_core.agents import AgentAction, AgentFinish # 定义共享状态 class AgentState(TypedDict): input: str research_result: str draft: str final_output: str # 创建代理函数（简化版） def research_agent(state: AgentState): result = tavily_search(state["input"]) # 调用搜索工具 return {"research_result": result} # 更新状态 def writing_agent(state: AgentState): prompt = f"基于研究结果撰写报告：{state['research_result']}" draft = llm.invoke(prompt) return {"draft": draft} # 构建图 graph = StateGraph(AgentState) graph.add_node("research", research_agent) graph.add_node("write", writing_agent) graph.add_edge("research", "write") # 固定路由 graph.add_edge("write", END) # 结束流程 graph.set_entry_point("research") # 入口节点 # 编译并运行 app = graph.compile() result = app.invoke({"input": "量子计算最新进展"}) print(result["final_output"]) 

---

3. 关键技巧

1. 动态路由
   通过条件边实现分支逻辑：
   
   

   def router(state: AgentState) -> str: if "需要专家评审" in state["draft"]: return "expert_agent" # 跳转到专家节点 return "end" # 否则结束 graph.add_conditional_edges("write", router) 

   
   
   
   
2. 工具集成
   代理通过工具增强能力：
   
   

   tools = [TavilySearchResults(max_results=3), PythonREPLTool()] agent = create_agent(llm, tools, "你是一个数据分析代理") 

   
   
   
   
3. 循环处理
   支持迭代优化：
   
   

   graph.add_edge("expert_agent", "write") # 专家反馈 → 重新写作 

   
   
   
   

---

架构优势

- 解耦设计：代理独立开发/替换 - 可控流程：通过图结构显式定义协作逻辑 - 状态透明：所有数据流可见可追踪 [^1][^2]