作者：CSDN博客
LangGraph 是LangChain 生态系统中的多智能体框架，它允许创建多个智能体组成的网络，这些智能体可以相互协作完成复杂任务。LangGraph 提供ReAct 、Plan-and-Execute 、Reflection & Critique 、RAG 以及多智能体协作（如代理切换、群体智能）等多种模式。LangGraph 定位为LangChain 的扩展库，用于构建有状态的、多智能体协作系统。LangGraph 的关键特性包括：
l 基于图的拓扑结构定义工作流。
l 支持循环和条件分支。
l 内置持久化状态管理。
l 兼容LangChain 原有工具链。
多智能体框架LangGraph 中的图
LangGraph 中的图（Graph ）可定义工作流，通过添加节点（智能体）和边（工作流转移）来构建，例如定义Supervisor 、Researcher 、Emailer 等节点，并添加边以协调任务执行。在LangGraph 中，“图”表示智能体网络的结构，由节点（智能体/ 工具）和边（通信路径）组成。图的结构可以是：
l 线性图：任务按顺序执行。
l 树状图：任务分解为子任务。
l 有向无环图（DAG ）：支持复杂的依赖关系。
l 循环图：支持迭代或反馈机制。
【示例 8.17 】LangGraph 基础图创建。
from langgraph.graph import StateGraph, END
# 定义状态结构（可选，但推荐）
from typing import TypedDict
class State(TypedDict):
start: bool
processed: bool
# 创建图
workflow = StateGraph(State)
# 定义节点函数
def start _ node(state):
print(" 开始执行 ")
return {"start": True}
def process _ node(state):
print(" 处理中 ...")
return {"processed": True}
# 添加节点
workflow.add _ node("start", start _ node)
workflow.add _ node("process", process _ node)
# 设置入口
workflow.set _ entry _ point("start")
# 添加边
workflow.add _ edge("start", "process")
workflow.add _ edge("process", END) # 使用 END 表示结束
# 编译
app = workflow.compile()
# 执行
result = app.invoke({})
print(" 执行结果 :", result)
示例代码执行结果如下：
开始执行
处理中 ...
执行结果 : {'start': True, 'processed': True}
多智能体框架LangGraph 中的状态
状态在LangGraph 中表示智能体执行过程中的中间结果。每个节点（智能体）可以：
l 读取当前状态。
l 修改状态并传递给下一个节点。
状态可以包含：
l 中间计算结果。
l 上下文信息。
l 历史决策记录。
状态管理是LangGraph 的核心功能之一，确保数据在整个智能体网络中正确流动。状态管理机制示例如下：
from typing import TypedDict
class AgentState(TypedDict):
input: str
history: list[str]
intermediate _ steps: list[tuple]
多智能体框架图中的节点
节点表示智能体（如research_agent 、chart_agent ），每个节点有特定功能（如研究、生成图表），通过函数定义其行为，并与其他节点协作。节点是LangGraph 中的基本计算单元，可以是：
l 智能体节点：包含完整的智能体逻辑。
l 工具节点：封装特定功能的工具。
l 决策节点：基于条件选择路径。
l 聚合节点：合并多个输入。
节点通过定义明确的接口与其他节点通信，接收输入状态并产生输出状态。
节点类型包括函数式节点和工具调用节点，示例如下：
# 函数式节点定义
def agent _ node(state):
response = agent.invoke(state)
return {"response": response}
# 工具调用节点
def tool _ node(state):
tool _ output = tools[state["selected _ tool"]](state["input"])
return {"output": tool _ output}
【示例 8.18 】 多节点工作流。
from langgraph.graph import StateGraph
# 定义状态类型，这在新版本中是必需的
from typing import TypedDict
class State(TypedDict):
value: int
workflow = StateGraph(State)
def start(state: State) -> State:
print(" 开始工作流 ")
return {"value": 0}
def increment(state: State) -> State:
new _ value = state["value"] + 1
print(f" 增加值到 {new _ value}")
return {"value": new _ value}
def end(state: State) -> State:
print(f" 最终值 : {state['value']}")
return state
workflow.add _ node("start", start)
workflow.add _ node("increment", increment)
workflow.add _ node("end", end)
workflow.set _ entry _ point("start")
workflow.add _ edge("start", "increment")
workflow.add _ edge("increment", "end")
workflow.set _ finish _ point("end")
app = workflow.compile()
app.invoke({"value": 0}) # 初始状态需要符合 State 类型定义
输出：
开始工作流
增加值到 1
最终值 : 1
多智能体框架LangGraph 中的边
边表示智能体之间的任务转移，可通过条件边（如根据状态决定下一个节点）或直接边实现，例如Researcher 完成后返回Supervisor 。
（1 ）边表示节点之间的连接关系，它定义了：数据流动的方向、触发条件、可选的数据转换、权重（在有多个路径时使用）。边的配置决定了智能体网络如何响应不同的输入和状态变化。
（2 ）边控制逻辑：
from langgraph.graph import ConditionalEdge
def should _ continue(state):
if state["needs _ more _ info"]:
return "human _ help"
return "end"
workflow.add _ conditional _ edges(
"decision _ node",
should _ continue,
{"human _ help": "human _ input _ node", "end": END}
)
（3 ）边类型：
l 固定边：无条件转移。
l 条件边：基于谓词函数路由。
l 动态边：运行时计算目标节点。
【示例 8.19 】 条件边示例。
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 定义状态结构
class State(TypedDict):
value: int
def start(state: State) -> State:
print(" 启动节点 ")
return {"value": 5}
def process _ a(state: State) -> State:
print(" 执行路径 A")
return state
def process _ b(state: State) -> State:
print(" 执行路径 B")
return state
def decide _ next(state: State) -> str:
if state["value"] > 3:
return "process _ a"
return "process _ b"
# 创建 StateGraph 实例
workflow = StateGraph(State)
# 添加节点
workflow.add _ node("start", start)
workflow.add _ node("process _ a", process _ a)
workflow.add _ node("process _ b", process _ b)
# 设置入口
workflow.set _ entry _ point("start")
# 添加条件边（注意：现在直接返回目标节点名）
workflow.add _ conditional _ edges(
"start",
decide _ next,
{ # 可选：显式映射，但如果你的 decide _ next 直接返回节点名，可以省略这个字典
"process _ a": "process _ a",
"process _ b": "process _ b"
}
)
# 添加普通边
workflow.add _ edge("process _ a", "process _ b")
workflow.add _ edge("process _ b", END) # 设置结束点
# 编译并运行
app = workflow.compile()
app.invoke({})
执行代码，输出如下：
执行路径 A
执行路径 B
多智能体框架LangChain 中的工具
LangChain 提供工具（如Google 搜索、Python 函数等），可通过继承BaseTool 类或@tool 装饰器定义自定义工具，智能体根据工具描述调用相应的方法。在LangChain 中，工具是智能体可以调用的外部 功能。
（1 ）常见的工具类型包括：
l API 工具：调用外部API ，如天气API 、股票API 、定位API 等。
l Python 函数工具：封装Python 函数。
l LLM 工具：使用语言模型完成特定任务。
l 搜索工具：如Google Serp API 、Wolfram Alpha 等。
（2 ）工具通过Tool 类封装，需要定义：
l 名称（name ）。
l 功能描述（description ）。
l 执行函数（func ）。
（3 ）工具生态系统示例：
from langchain.tools import tool
@tool
def semantic _ search(query: str) -> str:
""" 向量数据库检索工具 """
return vectorstore.similarity _ search(query)
# 工具注册
tools = [semantic _ search]
（4 ）高级功能：
l 工具组合：通过Toolkit 封装工具集。
l 权限控制：工具访问权限管理。
l IO 校验：基于Pydantic 的输入输出验证。
【示例 8.20 】 自定义工具实现（LangChain+Qwen ）。
# 安装依赖包 :pip install langchain langchain-community langchainhub dashscope
from langchain.tools import tool
from langchain.agents import AgentExecutor, create _ react _ agent
from langchain import hub
from langchain _ community.chat _ models import ChatTongyi
# |
广告