LangGraph智能体设计模式与多智能体开发（人工智能技术丛书）【行情 报价 价格 评测】-京东
王晓华LangGraph开发入门书《LangGraph智能体设计模式与多智能体开发》全文试读~_langgraph智能体设计模式与多智能体开发 pdf 下载-CSDN博客
LangGraph作为多智能体编排框架，专注于管理集成了大语言模型的应用程序的控制，尤其适用于构建复杂、有状态的代理系统。其核心设计理念是提供灵活的基础设施支持，允许开发者构建长期运行的工作流或代理，同时不抽象底层提示工程或模型架构，保留对执行过程的精细控制。LangGraph图标如图15-1所示。
图15-1 LangGraph图标
与传统代理框架不同，LangGraph通过图结构定义代理间的交互逻辑，支持循环流程和动态决策，能够实现多代理协作、任务自动化及复杂状态管理。例如，在需要多角色参与或持续状态更新的场景（如对话代理、业务流程自动化）中，开发者可通过节点和边的连接逻辑显式定义状态转移规则，而非依赖固定模板。
此外，LangGraph强调可靠性与可扩展性，提供工具链支持从开发、部署到运维的全生命周期管理，适用于需要高可控性的生产级AI应用。其灵活性使其既可作为独立框架构建定制化代理，也能与生态的其他工具（如提示词管理、模型集成）协同使用，形成完整的AI工作流解决方案。
15.1 初识LangGraph与工作流搭建
LangGraph是一个基于图结构的开源框架，专为构建状态化、多智能体系统而设计，尤其适合与大语言模型结合使用，用于创建动态代理及协作式工作流。它突破了传统有向无环图（Directed Acyclic Graph，DAG）的限制，通过节点（执行步骤）和边（执行顺序与条件）定义流程，支持循环和条件分支，使代理能迭代优化任务（如改进RAG管道的检索质量）。基于LangGraph的智能体构建如图15-2所示。[f1]
图15-2 基于LangGraph的智能体构建
15.1.1 用LangGraph把大象装进冰箱
从前面的讲解中可以看到，使用LangGraph的目标就是使用其状态、节点以及边构建一个可用于协同工作的智能体。接下来，我们将通过一个例子，具体展示基于LangGraph的智能体的搭建过程。
相信读者都知道这样一个段子“如何把大象装进冰箱”，那么问题来了，如何才能做到把大象装进冰箱（见图15-3）。步骤很简单，总共分三步：第一步，把冰箱门打开；第二步，把大象装进去；第三步，把冰箱门带上。
图15-3 把大象装进冰箱
我们首先看一下“把大象装进冰箱”的代码实现，读者可以直接运行代码查看结果。这个代码实现很简单，按照流程阅读即可，代码如下：
from IPython.display import Image, display
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.graph.state import CompiledStateGraph
from typing_extensions import TypedDict
# 1. 定义状态
class ElephantInFridgeState(TypedDict):
"""
定义状态—大象在冰箱中
状态内的参数可以在多个节点间保存和流转
"""
fridge_open: bool # 冰箱门是否打开
elephant_inside: bool # 大象是否在冰箱内
# 2. 定义每个步骤的处理节点
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第一步: 把冰箱门打开"""
print("正在打开冰箱门…")
# 执行当前节点业务逻辑
print("冰箱门已打开！")
# 更新状态
state["fridge_open"] = True
# 当前节点处理完成,更新状态
return state
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第二步: 把大象放进去"""
# 获取当前状态
if not state["fridge_open"]:
raise ValueError("冰箱门未打开，无法放入大象！")
# 执行当前节点业务逻辑
print("正在把大象放入冰箱…")
print("大象已放入冰箱！")
# 更新状态
state["elephant_inside"] = True
# 当前节点处理完成,更新状态
return state
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第三步: 把冰箱门带上"""
# 获取当前状态
if not state["elephant_inside"]:
print("警告：冰箱内没有大象，是否确认关闭？")
# 执行当前节点业务逻辑
print("正在关闭冰箱门…")
print("冰箱门已关闭！")
# 更新状态
state["fridge_open"] = False
# 当前节点处理完成,更新状态
return state
# 3. 构建Graph工作流图
def build_graph() -> CompiledStateGraph:
"""构建"把大象装冰箱"的工作流图"""
# 创建图对象StateGraph
graph = StateGraph(ElephantInFridgeState)
# 添加节点
graph.add_node("open_fridge", open_fridge)
graph.add_node("put_elephant", put_elephant)
graph.add_node("close_fridge", close_fridge)
# 添加起始边
graph.add_edge(START, "open_fridge")
# 添加边，将节点连接起来，形成工作流
graph.add_edge("open_fridge", "put_elephant")
graph.add_edge("put_elephant", "close_fridge")
# 添加结束边
graph.add_edge("close_fridge", END)
# 返回编译好的图
return graph.compile()
# 4. 运行Graph工作流
def run_workflow():
"""运行"把大象装冰箱"的工作流"""
# 构建图
graph = build_graph()
# 定义初始状态
initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)
# 运行图
result = graph.invoke(initial_state)
# 输出最终状态
print("\\n工作流执行完毕，最终状态:")
print(f"冰箱门状态: {'打开' if result['fridge_open'] else '关闭'}")
print(f"大象是否在冰箱内: {'是' if result['elephant_inside'] else '否'}")
# 可视化工作流图
display(Image(graph.get_graph().draw_mermaid_png()))
return result
# 执行入口
if __name__ == "__main__":
final_state = run_workflow()
下面我们分步骤对这个示例代码进行讲解。
1. 状态定义（ElephantInFridgeState）
class ElephantInFridgeState(TypedDict):
fridge_open: bool # 冰箱门开关状态 (True=开启，False=关闭)
elephant_inside: bool # 大象位置状态 (True=在冰箱内，False=在冰箱外)
（1）核心作用：作为工作流的共享内存空间，跨节点传递和持久化关键状态。
（2）状态属性：
- fridge_open：实时记录冰箱门的物理状态。
- elephant_inside：实时追踪大象的空间位置。
（3）设计特点：
- 继承自LangGraph的TypedDict，强制类型安全。
- 所有节点均可读写，实现跨节点数据共享。
- 初始状态：冰箱门关闭（False），大象在冰箱外（False）。
2. 节点功能实现
每个节点都是独立的状态处理器，遵循“获取状态→执行业务→更新状态→返回状态”的工作流范式。
1）开门节点（open_fridge）
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
print("正在打开冰箱门…")
print("冰箱门已打开！")
state["fridge_open"] = True # 更新门状态
return state
- 前置条件：无（可直接从初始状态启动）。
- 核心操作：变更fridge_open状态为True。
- 执行效果：输出操作日志，确保冰箱门开启。
2）放入大象节点（put_elephant）
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:
if not state["fridge_open"]: # 状态检查
raise ValueError("冰箱门未打开，无法放入大象！")
print("正在把大象放入冰箱…")
print("大象已放入冰箱！")
state["elephant_inside"] = True # 更新大象位置
return state
- 前置校验：强制检查fridge_open=True，否则抛出异常。
- 核心操作：变更elephant_inside状态为True。
- 关键机制：通过状态依赖确保工作流顺序逻辑。
3）关门节点（close_fridge）
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
if not state["elephant_inside"]: # 安全警告
print("警告：冰箱内没有大象，是否确认关闭？")
print("正在关闭冰箱门…")
print("冰箱门已关闭！")
state["fridge_open"] = False # 更新门状态
return state
- 安全机制：当elephant_inside=False时输出警告。
- 核心操作：重置fridge_open为False。
- 容错设计：允许在无大象时关门，但给出风险提示。
3. 工作流图构建（build_graph）
graph = StateGraph(ElephantInFridgeState)
# 节点注册
graph.add_node("open_fridge", open_fridge)
graph.add_node("put_elephant", put_elephant)
graph.add_node("close_fridge", close_fridge)
# 线性工作流定义
graph.add_edge(START, "open_fridge") # 入口 → 开门
graph.add_edge("open_fridge", "put_elephant") # 开门 → 放入大象
graph.add_edge("put_elephant", "close_fridge") # 放入大象 → 关门
graph.add_edge("close_fridge", END) # 关门 → 出口
1）图结构
创建有向无环图控制流。
2）节点关系
- START：LangGraph内置起始节点。
- 线性执行链：开门→放入大象→关门。
- END：LangGraph内置终止节点。
3）编译输出
返回可执行的CompiledStateGraph对象。
4）工作流执行（run_workflow）
def run_workflow():
graph = build_graph() # 编译图结构
initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)
result = graph.invoke(initial_state) # 执行工作流
# 状态结果输出
print(f"冰箱门状态: {'打开' if result['fridge_open'] else '关闭'}")
print(f"大象是否在冰箱内: {'是' if result['elephant_inside'] else '否'}")
# 可视化输出
display(Image(graph.get_graph().draw_mermaid_png()))
5）执行流程
（1）编译工作流图。
（2）初始化状态：fridge_open=False, elephant_inside=False。
（3）注入初始状态并触发执行。
6）输出内容
- 实时打印各节点操作日志。
- 终态数据报告。
- Mermaid格式可视化流程图（PNG渲染）。
完整的流程如图15-4所示。
图15-4 大象放进冰箱的完整流程
这里我们通过边严格定义顺序：开门→放入大象→关门，并加载了前置条件校验（开门检查）、异常安全机制（ValueError）以及风险预警（空冰箱关门警告）。
此实现完整演示了LangGraph在顺序工作流建模中的核心能力，通过状态对象实现节点间解耦，同时确保业务流程严格按顺序执行。
15.1.2 LangGraph特性与技术原理
作为LangChain生态系统的一部分，LangGraph也可以独立运行，其核心优势在于支持循环流程、细粒度可控性和内置持久性，解决了传统有向无环图框架无法处理循环逻辑的局限性。
1. 核心特性与技术原理
- 循环与分支：LangGraph允许定义包含循环和条件分支的工作流，例如代理可多次调用工具优化结果（如改进RAG管道的检索质量），适应复杂任务需求。
- 状态管理：通过节点（执行步骤）和边（执行顺序与条件）构建图结构，状态在节点间传递并动态更新。例如，节点可调用LLM或工具，条件边根据状态决定下一步操作。
- 持久性与人工干预：自动保存每一步状态，支持暂停/恢复执行（如错误恢复或时间回溯），并允许人类审核或修改代理的下一步行动，实现“人机回环（Human-in-the-Loop）”协作。
- 流式支持与集成：实时流式传输每个节点的输出（包括令牌流），并与LangChain/LangSmith无缝集成，提升开发效率。
2. 应用场景
LangGraph适用于自动化复杂业务流程：
- 客户服务：构建多轮对话机器人，提供个性化支持。
- 数据分析：代理协同检索多源信息并生成报告。
- 流程自动化：如订单处理、库存管理等需状态跟踪的场景。
- 教育/推荐系统：AI教师动态调整教学策略，或基于用户行为生成推荐。
3. 示例与部署
用户可通过Python代码定义图结构（如分类查询→生成响应→循环优化），并利用内置MemorySaver保存对话上下文，确保长期任务的连续性。LangGraph的商业平台支持生产环境部署，适合构建高可控、高可靠的智能系统。
LangGraph通过图计算模型（灵感源自Pregel和Apache Beam）实现了灵活的任务编排，为开发动态AI应用提供了底层支持，推动多代理协作与自动化演进。
15.1.3 LangGraph组件详解
前面讲解的基于“链”的模型调用与流程设计，虽然能够较好地完成任务，但是简单的链不具备循环能力，对于多分支任务，则需要一个具备更精细控制能力的框架来支持复杂场景的LLM应用。
LangGraph的出现宣布智能体的使用进入多智能体框架领域。LangGraph是基于图论运作的，提供了一种状态机的技术，可以驱动循环代理调用，实现有向有环图。因此，LangGraph有3个关键元素：
- State（状态）：一个共享的数据结构，表示应用程序的当前快照。它可以是任何Python类型，但通常是TypedDict或Pydantic BaseModel。
- Nodes（节点）：编码代理逻辑的Python函数。它们接收当前的State作为输入，执行一些计算或起到辅助作用，并返回一个更新后的State。
- Edges（边）：Python函数，基于当前的State决定接下来执行哪个Node。它们可以是条件分支或固定转换。
通过组合Nodes和Edges，可以创建复杂、循环的工作流，随着时间的推移演化State。不过，真正的强大之处在于LangGraph如何管理State。需要强调的是，Nodes和Edges仅仅是Python函数—它们可以包含LLM，也可以只是普通的Python代码。
简而言之，节点是用于工作的模块，边告诉我们接下来该做什么。
LangGraph的底层图算法使用消息传递来定义一个通用程序。当一个Node完成其操作时，它会沿着一条或多条边发送消息给其他节点。这些接收节点随后执行它们的函数，将结果消息传递给下一组节点，过程继续，且程序以离散的“超级步骤”进行。
一个超级步骤可以被视为对图节点的一次迭代。并行运行的节点属于同一超级步骤，而顺序运行的节点属于不同的超级步骤。在图执行开始时，所有节点都处于非活动状态。当一个节点在其任何输入边（或“通道”）上收到新消息（状态）时，会变为活动状态。活跃节点运行其函数并响应更新。
在每个超级步骤结束时，没有收到消息的节点会通过标记自己为非活动状态来停止投票。图执行在所有节点都非活动且没有消息传递时终止。
在定义图形时，首先要做的是定义图形的State。State包含图形的模式以及指定如何应用更新到State的reducer函数。State的模式将是所有Nodes和Edges在图中的输入模式，可以是TypedDict或Pydantic模型。所有Nodes节点将会发出对State的更新，然后通过指定的reducer函数应用这些更新。
15.1.4 LangGraph中的StateGraph
StateGraph是LangChain中的一个类，用于表示图的数据结构，并反映系统的当前状态。它类似于状态机，通过节点对状态的更新来驱动流程。
- 初始化时需定义state_schema，指定状态的结构，确保节点可以访问和更新共享状态。
- 状态图的核心作用是维护全局状态，所有节点通过读写该状态进行通信。
StateGraph的特点：
- 支持循环代理调用，适用于复杂任务（如对话代理、自动化流程）。
- 状态会随着节点的执行逐步更新，形成完整的执行轨迹。
示例代码如下：
from langgraph.graph import StateGraph
from typing import TypedDict, List, Annotated
import operator
# 定义一个State
class State(TypedDict):
input: str
all_actions: Annotated[List[str], operator.add]
# 初始化一个StateGraph对象
graph = StateGraph(State)
StateGraph是LangGraph的核心，是graph的一个类。可以通过定义一个State来初始化一个StateGraph对象。定义的State可以随时更新StateGraph对象的状态。StateGraph对象可以通过Nodes更新状态。
reducer函数是理解如何将来自节点的更新应用到State的关键。State中的每个键都有其独立的reducer函数。如果没有显式指定reducer函数，则假定对该键的所有更新都应覆盖它。
以下两个示例将展示如何使用默认的reducer函数：
from typing import TypedDict
class State(TypedDict):
foo: int
bar: list[str]
在此示例中，没有为任何键指定reducer函数。假设图的输入是{"foo": 1, "bar": ["hi"]}，再假设第一个节点返回{"foo": 2}。这被视为对状态的更新。注意，节点不需要返回整个State模式—只需要一个更新。在应用此更新后，State将变为{"foo": 2, "bar": ["hi"]}。如果第二个节点返回{"bar": ["bye"]}，那么State将变为{"foo": 2, "bar": ["bye"]}。
from typing import TypedDict, Annotated
from operator import add
class State(TypedDict):
foo: int
bar: Annotated[list[str], add]
在此示例中，我们使用Annotated类型为第二个键（bar）指定了一个reducer函数（operator.add）。注意，第一个键保持不变。假设图的输入是{"foo": 1, "bar": ["hi"]}，再假设第一个节点返回{"foo": 2}。这被视为对状态的更新。在应用此更新后，State将变为{"foo": 2, "bar": ["hi"]}。如果第二个节点返回{"bar": ["bye"]}，那么State将变为{"foo": 2, "bar": ["hi", "bye"]}。需要注意，这里的bar键是通过将两个列表加在一起更新的。
15.1.5 LangGraph中的节点
节点是图中的核心执行单元，每个节点具有唯一名称，且可以是一个函数、可运行对象（Runnable）或LangChain表达式。
- 节点接收字典类型的状态数据，返回更新后的相同结构的状态。
- 特殊节点"END"表示状态机的终止点，流程执行到此即结束。
节点的行为规则：
- 节点的输入和输出需严格遵循state_schema定义的状态结构。
- 节点可通过调用模型（如LLM）生成结果，并更新状态中的特定字段。
示例代码如下：
# 添加节点
# graph.add_node(name, value)，
# name是节点的名称str
# value是节点的值，是一个函数或者是LCEL runnable
# 一个model，暂时理解为调用大模型Reasoning的过程
def model():
return "model"
# 一个tool_executor，理解为Acting的过程
def tool_executor():
return "tool_executor"
graph.add_node("model", model)
graph.add_node("tools", tool_executor)
# 还有一个特殊的节点，叫End，表示StateGraph的结束状态
from langgraph.graph import END
在LangGraph中，节点通常是Python函数（同步或异步），其中第一个位置参数是state；第二个位置参数（可选）是config，包含可选的可配置参数（例如thread_id）。下面是添加不同的自定义节点的代码：
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph
builder = StateGraph(dict)
def my_node(state: dict, config: RunnableConfig):
print("In node: ", config["configurable"]["user_id"])
return {"results": f"Hello, {state['input']}!"}
# 第二个参数是可选的
def my_other_node(state: dict):
return state
builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)
在节点中，有以下两个特殊节点。
1. START Node（开始节点）
START Node是一个开始节点，表示将用户输入发送到图的节点。引用此节点的主要目的是确定哪些节点应首先被调用。
from langgraph.graph import START
graph.add_edge(START, "node_a")
2. END Node（结束节点）
END Node表示一个结束节点。当想表示某些边在完成后没有进一步操作时，可以引用此节点。
from langgraph.graph import END
graph.add_edge("node_a", END)
15.1.6 LangGraph中的边
边（Edges）定义了逻辑，包括如何路由以及图形如何停止。这是Agent如何工作以及不同节点如何相互通信的一个重要部分。
一个节点可以有多个传出边，此时所有这些目标节点将在下一个超级步骤中并行执行。
条件边的三要素如下：
- 上游节点：边的起点（如节点A）。
- 路由函数：根据状态值返回条件结果（如route(state)→"next_node"）。
- 状态映射：将路由函数的返回值与具体下游节点绑定。
边有以下几种关键类型：
- Normal Edges（普通边）：直接从一个节点转到下一个节点。
- Conditional Edges（条件边）：调用一个函数来确定接下来要转到的节点。
- Entry Point（入口点）：当用户输入到达时首先调用的节点。
- Conditional Entry Point（条件入口点）：调用一个函数来确定当用户输入到达时首先调用的节点。
1. Normal Edges
如果希望从节点A转到节点B，可以直接使用add_edge()方法。
graph.add_edge("node_a", "node_b")
2. Conditional Edges
如果希望有条件地路由到一条或多条边（或可选地终止），可以使用add_conditional_edges()方法。此方法接收节点的名称和routing_function（路由函数）来调用该节点后执行：
graph.add_conditional_edges("node_a", routing_function)
路由函数类似于节点，接收图的当前状态并返回一个值。默认情况下，路由函数的返回值用作状态接下来要发送到的节点的名称（或列表）。所有这些节点将在下一个超级步骤中并行运行。
graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})
3. Entry Point
Entry Point是图形启动时首先运行的节点。可以使用add_edge()方法从虚拟的START节点到要执行的第一个节点来指定进入图的位置。
from langgraph.graph import START
graph.add_edge(START, "node_a")
4. Conditional Entry Point
Conditional Entry Point允许根据自定义逻辑在不同节点之间开始。读者可以使用add_conditional_edges从虚拟的START节点来实现这一点。
from langgraph.graph import START
graph.add_conditional_edges(START, routing_function)
读者可以选择提供一个字典，将路由函数的输出映射到下一个节点的名称。
graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})
下面给出完整的使用节点的示例：
# 在创建了上面两个节点之后，可以通过边将两个节点连接起来
# LangGraph提供了3种边的类型
# Starting Edge，开始边，表示从StateGraph的开始状态到第一个节点的边
# 开始边到model节点
graph.set_entry_point("model")
# Normal Edge，普通边，表示从一个节点到另一个节点的边
# model节点到tools节点
graph.add_edge("model", "tools")
# Conditional Edges，条件边，表示从一个节点到另一个节点的边，但是这个边是有条件的
# 一个条件函数
def condition():
return "我是一个判定节点"
# 添加条件边，从model节点开始，根据condition函数的返回值，如果是end就结束，如果不是end就继续到tools节点
graph.add_conditional_edges(
"model",
condition, # 判定函数
{
"end": END,
"continue": "tools"
}
)
# 最后返回一个runnable的chain
app = graph.compile()
15.2 基于LangGraph的聊天助手搭建
在深入探讨智能体架构的演进路径时，我们需进一步解析大语言模型与流程编排框架的协同机制。基于前期构建的LangGraph智能体框架，当前研究聚焦于模型能力与流程控制的深度融合。通过将Qwen3大模型嵌入为独立计算节点，可实现语言生成能力与既有功能模块的解耦式集成。这种设计模式既保留了流程编排的灵活性，又赋予智能体即时接入先进语言模型的通道。
具体实现层面，LangGraph的节点化加载机制提供了两种典型接入范式：独立节点模式和原生接口集成。在独立节点模式中，Qwen3可作为专用处理单元挂载至流程任意环节，通过定义标准化的输入/输出接口，使复杂任务可拆解为“意图解析→工具调用→结果润色”的多阶段流水线。例如，在客服场景中，原始查询首先经由Qwen3模型进行需求分类，再由知识库检索节点执行数据查询，在返回结果前再次调用模型生成自然语言回复。这种分层处理架构显著提升了复杂任务的处理鲁棒性。
更值得关注的是LangGraph原生接口集成。通过实现框架预置的接口，我们可直接将Qwen3的推理服务封装为框架原生组件。这种集成方式不仅能支持同步/异步调用模式的灵活配置，还能利用框架内置的缓存机制和批处理优化策略。在实际部署时，可针对对话历史管理、上下文长度控制等场景进行定制化开发，例如通过状态管理模块自动维护多轮对话的记忆轨迹，或利用向量数据库实现历史记录的语义化检索。
两种接入方式在工程实践中呈现互补特性：独立节点模式适合快速验证模型能力边界，其可视化编排特性便于业务人员理解流程逻辑；而原生接口集成则更适用于生产环境，通过深度耦合框架特性可获得更优的性能表现。需要注意，随着模型迭代速度加快，框架的插件化设计使得技术升级无须改动核心流程，这为构建持续演进的智能体系统提供了基础设施保障。
15.2.1 LangGraph构建的聊天助手
在上一节中，我们演示了“把大象装进冰箱”这一完整流程，通过这个完整的工作流我们知道了LangGraph的基本使用方法。本小节将使用LangGraph完成一个聊天助手智能体的搭建工作。
1. 基于Qwen3的智能体搭建
为了适配LangGraph，我们首先需要提供可用于在智能体对话中作为对话模块的大模型Qwen3。相对于经典的Qwen3的对话框架，我们需要修正新的问答模块，同时为了简便起见，还需要将其通过类的方式进行重新组建。代码如下：
from openai import OpenAI
from langchain_core.messages import HumanMessage, AIMessage
#QwenLLM 类
class QwenLLM:
def __init__(self):
self.client = OpenAI(
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
self.system_prompt = "你是一个有用的智能助手！"
def invoke(self, messages: list):
# 将 LangChain 消息对象转换为 OpenAI 格式
openai_messages = [
{"role": "system", "content": self.system_prompt}
]
for msg in messages:
if isinstance(msg, HumanMessage):
openai_messages.append({"role": "user", "content": msg.content})
elif isinstance(msg, AIMessage):
openai_messages.append({"role": "assistant", "content": msg.content})
completion = self.client.chat.completions.create(
model="qwen-plus",
messages=openai_messages
)
return AIMessage(content=completion.choices[0].message.content)
上面这段代码定义了一个名为QwenLLM的类，用于通过OpenAI兼容接口调用阿里云的Qwen大模型。该类在初始化时配置了API密钥和基础URL，并设定了一个系统提示语。Invoke()方法用于接收一个消息列表，并将其转换为适合OpenAI接口调用的消息格式，其中用户消息和助手消息分别对应到相应的角色。最后，调用Qwen-Plus模型，返回生成的响应内容并封装成AIMessage对象。
AIMessage是langchain_core.messages模块中的一个类，用于表示由AI助手生成的消息。它通常在使用LangChain框架开发应用时出现，用来统一管理和传递对话历史中助手一方的消息。
在上面代码中，AIMessage用来封装调用Qwen大模型后返回的AI回复内容。这样做的好处是可以将用户输入（HumanMessage）和AI输出（AIMessage）统一成一致的数据结构，便于后续处理、记录或传递给其他模块继续处理，比如构建对话链、记忆管理、日志记录等。
2. 结合Qwen3和LangGraph的聊天助手智能体构建
我们沿用先前构建的大语言模型，并将其作为节点集成到LangGraph聊天助手中。此处以Qwen3作为节点进行添加，并完成相应的配置，具体代码如下：
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
from langgraph.graph import StateGraph
graph_builder = StateGraph(State)
from qwen import QwenLLM
llm = QwenLLM()
def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
for event in graph.stream({"messages": ("user", user_input)}):
for value in event.values():
print("Assistant:", value["messages"][-1].content)
上面这段代码实现了一个基于状态图的简单聊天机器人，首先定义了一个State类型，使用TypedDict和Annotated来描述包含消息列表的状态结构；然后创建了一个StateGraph状态图，初始化了一个QwenLLM模型实例，并定义了一个chatbot节点函数，该函数接收当前状态，调用LLM生成回复并返回更新后的消息列表。接着将chatbot节点设为入口点并编译成可执行图。最后进入交互循环，持续接收用户输入（输入“quit”“exit”“q”可退出），将用户消息添加到状态中，并通过graph.stream流式处理状态图，实时打印助手的回复内容。整个流程实现了用户与基于大语言模型的聊天机器人的持续对话功能。
15.2.2 原生LangGraph接口对Qwen3的调用
前面我们通过使用Qwen3自带的接口来连接与使用LangGraph。此外，我们还可以使用LangGraph接口来调用Qwen3，代码如下：
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # 假设的 API 地址
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
agent = create_react_agent(
model=llm,
tools=[get_weather],
prompt="你是一个个人助手"
)
# Run the agent
response = agent.invoke(
{"messages": [{"role": "user", "content": "今天shanghai的天气怎么样"}]})
for message in response['messages']:
print(message)
print("****")
print(message.content)
print("—————-")
运行上面代码，结果如下：
content='今天shanghai的天气怎么样' additional_kwargs={} response_metadata={} id='39679bab-bf29-488a-b4d1-c0d198844df7'
****
今天shanghai的天气怎么样
—————-
content='' additional_kwargs={'tool_calls': [{'id': 'call_191569dd5ac74905a22816', 'function': {'arguments': '{"city": "Shanghai"}', 'name': 'get_weather'}, 'type': 'function', 'index': 0}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 18, 'prompt_tokens': 171, 'total_tokens': 189, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None} id='run–6e3a2c4f-e1a0-48c1-82b7-2474df8c5aca-0' tool_calls=[{'name': 'get_weather', 'args': {'city': 'Shanghai'}, 'id': 'call_191569dd5ac74905a22816', 'type': 'tool_call'}] usage_metadata={'input_tokens': 171, 'output_tokens': 18, 'total_tokens': 189, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
****
—————-
content='Shanghai 天气晴朗!' name='get_weather' id='e1a73c6c-a071-49b4-920a-fd1170ff0e98' tool_call_id='call_191569dd5ac74905a22816'
****
Shanghai 天气晴朗!
—————-
content='今天上海的天气是晴朗的！希望你有美好的一天！' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 203, 'total_tokens': 220, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run–6879e139-14c0-4e8a-a22e-ae52f4b7740d-0' usage_metadata={'input_tokens': 203, 'output_tokens': 17, 'total_tokens': 220, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
****
今天上海的天气是晴朗的！希望你有美好的一天！
—————-
上面结果展示了模型在运行过程中对内容的处理流程，其中content表示交互的文本内容；tool_calls表示使用工具调用方法完成对应的工具使用。
15.2.3 一般智能体运行范式详解
接下来，我们对LangGraph文本处理部分进行拆解。一个简单的文本处理过程如下：
（1）读取用户输入。
（2）把输入发给LLM，让它生成回复。
（3）如果LLM给出了工具调用的指令，就执行这些工具（比如tool_calls命令）。
（4）把工具执行结果反馈给LLM。
（5）重复这个循环。
LangGraph的Agent架构本质是构建了一个自迭代的认知增强回路：当LLM遇到知识边界时，通过工具调用扩展能力边界，工具执行结果作为新输入反哺LLM，形成“感知-决策-执行-反馈”的闭环。这种设计巧妙地将LLM的符号推理能力与确定性工具相结合，解决了纯LLM的问题。
我们使用代码实现这个处理过程，示例如下：
def loop(llm):
msg = user_input()
while True:
output, tool_calls = llm(msg)
print("Agent: ", output)
if tool_calls:
msg = [ handle_tool_call(tc) for tc in tool_calls ]
else:
msg = user_input()
上面代码成功地将Agent转换成一个“LLM+工具执行+反馈调整”的小闭环，虽然简单但非常高效。
我们使用这个模型模式来完成Agent工具调用的范式，代码如下：
from langchain.tools import tool
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
# 工具函数定义（使用LangChain的@tool装饰器）
@tool
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
).bind_tools([get_weather])
# 系统提示初始化
system_prompt = "你是一个个人助手"
messages = [{"role": "system", "content": system_prompt}]
def user_input():
"""获取用户输入"""
return input("User: ")
def handle_tool_call(tool_call):
"""执行工具调用并返回结果消息"""
func_name = tool_call['name']
args = tool_call['args']
# 查找匹配的工具
if func_name == "get_weather":
result = get_weather.invoke(args)
return ToolMessage(
content=result,
tool_call_id=tool_call['id'],
name=func_name
)
return ToolMessage(
content=f"未知工具: {func_name}",
tool_call_id=tool_call['id'],
name=func_name
)
def loop(llm):
"""主循环处理交互逻辑"""
msg = user_input()
messages.append(HumanMessage(content=msg))
while True:
# 调用模型获取响应
response = llm.invoke(messages)
# 提取模型输出和工具调用
output = response.content
tool_calls = response.tool_calls if hasattr(response, 'tool_calls') else []
print(f"Agent: {output}")
if tool_calls:
# 处理所有工具调用
tool_responses = [handle_tool_call(tc) for tc in tool_calls]
messages.append(response) # 添加AI消息
messages.extend(tool_responses) # 添加工具结果
print(f"工具调用结果: {[r.content for r in tool_responses]}")
else:
# 显示最终回复并等待新输入
messages.append(response)
msg = user_input()
messages.append(HumanMessage(content=msg))
# 启动交互循环
loop(llm)
这种架构标志着从单次问答到持续认知进程的转变：
- 传统Chat：input→LLM→output。
- LangGraph范式：(input+state)→[LLM+Tools]→(output+new_state)。
最终形成具备自我演进能力的对话实体，其核心价值在于将离散的交互转换为持续的价值累积过程，每次工具调用都在扩展Agent的认知边界。当循环终止时，系统输出的不仅是答案，也是经过工具验证的知识轨迹。输出结果如下：
User: hi
Agent: Hello! How can I assist you today?
User: 今天shanghai的天气如怎么样
Agent:
工具调用结果: ['Shanghai 天气晴朗!']
Agent: 今天上海的天气是晴朗的！
User:
在解析这段LangGraph实现逻辑时，核心在于理解消息传递链与工具调用的协同机制。代码中的loop()函数构建了交互主循环，其设计精髓体现在消息队列的动态维护上。每次的用户输入都会被封装为HumanMessage对象追加到messages列表，这个列表本质上是会话上下文的完整记录。当调用llm.invoke(messages)时，模型会基于完整对话历史生成响应。这种设计确保了上下文连贯性，避免了传统对话系统常见的上下文丢失问题。
这种工具调用处理流程展现了典型的链式调用模式。当模型返回包含tool_calls的响应时，系统会遍历每个工具调用指令，通过handle_tool_call()函数进行解析执行。这里值得注意的是工具路由机制：通过检查tool_call['name']与已注册工具的匹配关系，实现了灵活的工具扩展能力。在实际开发中，这种模式可轻松扩展支持数据库查询、API调用等复杂操作，只需按规范注册新的工具函数即可。
下面是使用预创建方式完成的LangGraph智能体代码：
from langchain.tools import tool
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
def prompt(state):
user_name = "xiaohua"
system_msg = f"You are a helpful assistant. Address the user as {user_name}."
return [{"role": "system", "content": system_msg}] + state["messages"]
# 工具函数定义（使用LangChain的@tool装饰器）
@tool
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
).bind_tools([get_weather])
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(
model=llm, # 直接传入 LLM 实例
tools=[get_weather],
prompt=prompt
)
result = agent.invoke(HumanMessage(content="你好"))
代码运行结果请读者可以自行验证。
15.2.4 基于LangGraph与Qwen3完成聊天助手
接下来，我们使用LangGraph结合Qwen3来完成聊天助手。为了加深读者对实现过程的理解，我们将在聊天助手中额外添加天气查询工具，并通过条件节点的判断进行工具调用。代码如下：
from langchain.tools import tool
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
tools = [get_weather]
# 工具节点
tool_node = ToolNode(tools=tools)
# 定义状态类型
from typing import Annotated
from typing_extensions import TypedDict
class State(TypedDict):
messages: Annotated[list, lambda old, new: old + new]
# Chatbot 节点：LLM 响应（不绑定工具）
def chatbot(state: State):
response = llm.invoke(state["messages"])
return {"messages": [response]}
# 构建图
from langgraph.graph import StateGraph,END
from langgraph.prebuilt import ToolNode, tools_condition
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 路由逻辑：LLM 判断是否调用工具
graph_builder.set_entry_point("chatbot")
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("chatbot", END)
# 编译
graph = graph_builder.compile()
# 执行对话
config = {"configurable": {"thread_id": "1"}}
def stream_graph_updates(user_input: str):
events = graph.stream(
{"messages": [{"role": "user", "content": user_input}]},
config,
#stream_mode="values",
)
for event in events:
for value in event.values():
print("Assistant:", value["messages"][-1].content)
# 运行主循环
while True:
try:
user_input = input("User: ")
if user_input.lower() in ["exit", "quit", "q"]:
print("Goodbye!")
break
stream_graph_updates(user_input)
except Exception as e:
print(f"错误：{e}")
break
读者可以自行运行代码验证结果。
15.2.5 进阶智能体构建
在上一小节中，我们使用基础的LangGraph组件完成了智能体的构建，其中自定义条件选择的实现方法如下：
# 更精确的工具调用检测
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state['messages']
last_message = messages[-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_message, AIMessage) and hasattr(last_message, 'tool_calls') and last_message.tool_calls:
print("检测到工具调用，转到工具节点")
return "tools"
# 其他情况结束流程
return END
在上面代码中，我们额外添加了对消息进行检测的方法，并返回调用工具。
完整的实现代码如下：
from typing import Literal
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""这是一个对天气进行搜索并得到返回结果的工具."""
return f"{city} 天气晴朗!"
tools = [get_weather]
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
# 工具节点
tool_node = ToolNode(tools=tools)
# 更精确的工具调用检测
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state['messages']
last_message = messages[-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_message, AIMessage) and hasattr(last_message, 'tool_calls') and last_message.tool_calls:
print("检测到工具调用，转到工具节点")
return "tools"
# 其他情况结束流程
return END
# 使用字典键访问消息
def chatbot(state: MessagesState):
# 绑定工具到LLM
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke(state['messages'])
return {"messages": [response]}
# 构建图
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 设置入口点
graph_builder.set_entry_point("chatbot")
# 正确设置条件边
graph_builder.add_conditional_edges(
"chatbot",
should_continue,
{
"tools": "tools", # 如果调用工具，转到工具节点
END: END # 否则结束
}
)
# 工具节点执行后回到chatbot
graph_builder.add_edge("tools", "chatbot")
# 编译图
graph = graph_builder.compile()
# 执行图
final_state = graph.invoke(
{"messages": [HumanMessage(content="上海的天气怎么样?")]},
config={"configurable": {"thread_id": 929}}
)
# 输出结果
result = final_state["messages"][-1].content
print("最终结果:", result)
运行上面代码，结果如下：
检测到工具调用，转到工具节点
最终结果: 上海的天气目前是晴朗的。
此外，我们还可以通过LangGraph添加额外的记忆力组件，从而使得智能体在运行时能够记忆原有的内容。添加记忆力组件的代码如下：
# 初始化内存以在图运行之间持久化状态
checkpointer = MemorySaver()
# 编译图
# 这将编译成一个LangChain可运行对象
# 这意味着你可以像使用其他可运行对象一样使用它
# 注意，我们（可选地）在编译图时传递内存
graph = graph_builder.compile(checkpointer=checkpointer)
完整代码如下：
from typing import Literal
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""这是一个对天气进行搜索并得到返回结果的工具."""
return f"{city} 天气晴朗!"
tools = [get_weather]
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
# 工具节点
tool_node = ToolNode(tools=tools)
# 更精确的工具调用检测
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state['messages']
last_message = messages[-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_message, AIMessage) and hasattr(last_message, 'tool_calls') and last_message.tool_calls:
print("检测到工具调用，转到工具节点")
return "tools"
# 其他情况结束流程
return END
# 使用字典键访问消息
def chatbot(state: MessagesState):
# 绑定工具到LLM
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke(state['messages'])
return {"messages": [response]}
# 构建图
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 设置入口点
graph_builder.set_entry_point("chatbot")
# 正确设置条件边
graph_builder.add_conditional_edges(
"chatbot",
should_continue,
{
"tools": "tools", # 如果调用工具，转到工具节点
END: END # 否则结束
}
)
# 工具节点执行后回到chatbot
graph_builder.add_edge("tools", "chatbot")
# 初始化内存以在图运行之间持久化状态
checkpointer = MemorySaver()
# 编译图
# 这将编译成一个LangChain可运行对象
# 这意味着你可以像使用其他可运行对象一样使用它
# 注意，我们（可选地）在编译图时传递内存
graph = graph_builder.compile(checkpointer=checkpointer)
# 执行图，使用可运行对象
final_state = graph.invoke(
{"messages": [HumanMessage(content="上海的天气怎么样?")]},
config={"configurable": {"thread_id": 929}}
)
# 从 final_state 中获取最后一条消息的内容
result = final_state["messages"][-1].content
print(result)
final_state = graph.invoke(
{"messages": [HumanMessage(content="我问的那个城市?")]},
config={"configurable": {"thread_id": 929}}
)
result = final_state["messages"][-1].content
print(result)
读者可以自行运行代码验证结果。
15.3 本章小结
在本章中，我们深入探索了LangGraph的核心应用，通过智能体案例展示了其基本组成结构和运行机制。案例演示清晰呈现了LangGraph如何驱动智能体完成文本交互和工具调用，展现了其将复杂流程转换为可编程图的独特价值。在Qwen3模型的集成方面，我们既可通过其原生API接口直接调用，也能利用LangGraph的统一调度框架实现更灵活的流程控制，这种双重接入方式为开发者提供了充分的自由度。
在工具调用方面，我们不仅实现了基础功能，而且通过自定义条件判断和工具节点设计，创建了智能化的触发机制—当用户输入包含“天气”等关键词时，系统能自动生成工具调用请求，这种设计显著提升了交互的自然度和效率。整个过程展示了LangGraph如何将多步骤、多工具的工作流转换为可管理的状态转移过程，其中状态对象的动态更新机制确保了对话上下文的连贯性。
通过实践验证，LangGraph的核心优势在于其将复杂智能体工作流抽象为节点和边的能力。这种范式转换使得开发者能够像设计电路图一样编辑AI行为，将LLM的认知能力与确定性工具无缝结合。工具调用不再只是功能扩展手段，更成为智能体与环境交互的感知延伸。当天气查询工具返回“晴朗”时，这不仅是数据反馈，更是智能体对现实世界认知的更新。
[f1]本节介绍LangChain，此处智能体构造与组件图似乎与LangChain不相关，可否删除图15-2？或者替换为LangChain相关的图
LangGraph作为多智能体编排框架，专注于管理集成了大语言模型的应用程序的控制，尤其适用于构建复杂、有状态的代理系统。其核心设计理念是提供灵活的基础设施支持，允许开发者构建长期运行的工作流或代理，同时不抽象底层提示工程或模型架构，保留对执行过程的精细控制。LangGraph图标如图15-1所示。
图15-1 LangGraph图标
与传统代理框架不同，LangGraph通过图结构定义代理间的交互逻辑，支持循环流程和动态决策，能够实现多代理协作、任务自动化及复杂状态管理。例如，在需要多角色参与或持续状态更新的场景（如对话代理、业务流程自动化）中，开发者可通过节点和边的连接逻辑显式定义状态转移规则，而非依赖固定模板。
此外，LangGraph强调可靠性与可扩展性，提供工具链支持从开发、部署到运维的全生命周期管理，适用于需要高可控性的生产级AI应用。其灵活性使其既可作为独立框架构建定制化代理，也能与生态的其他工具（如提示词管理、模型集成）协同使用，形成完整的AI工作流解决方案。
15.1 初识LangGraph与工作流搭建
LangGraph是一个基于图结构的开源框架，专为构建状态化、多智能体系统而设计，尤其适合与大语言模型结合使用，用于创建动态代理及协作式工作流。它突破了传统有向无环图（Directed Acyclic Graph，DAG）的限制，通过节点（执行步骤）和边（执行顺序与条件）定义流程，支持循环和条件分支，使代理能迭代优化任务（如改进RAG管道的检索质量）。基于LangGraph的智能体构建如图15-2所示。[f1]
图15-2 基于LangGraph的智能体构建
15.1.1 用LangGraph把大象装进冰箱
从前面的讲解中可以看到，使用LangGraph的目标就是使用其状态、节点以及边构建一个可用于协同工作的智能体。接下来，我们将通过一个例子，具体展示基于LangGraph的智能体的搭建过程。
相信读者都知道这样一个段子“如何把大象装进冰箱”，那么问题来了，如何才能做到把大象装进冰箱（见图15-3）。步骤很简单，总共分三步：第一步，把冰箱门打开；第二步，把大象装进去；第三步，把冰箱门带上。
图15-3 把大象装进冰箱
我们首先看一下“把大象装进冰箱”的代码实现，读者可以直接运行代码查看结果。这个代码实现很简单，按照流程阅读即可，代码如下：
from IPython.display import Image, display
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.graph.state import CompiledStateGraph
from typing_extensions import TypedDict
# 1. 定义状态
class ElephantInFridgeState(TypedDict):
"""
定义状态—大象在冰箱中
状态内的参数可以在多个节点间保存和流转
"""
fridge_open: bool # 冰箱门是否打开
elephant_inside: bool # 大象是否在冰箱内
# 2. 定义每个步骤的处理节点
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第一步: 把冰箱门打开"""
print("正在打开冰箱门…")
# 执行当前节点业务逻辑
print("冰箱门已打开！")
# 更新状态
state["fridge_open"] = True
# 当前节点处理完成,更新状态
return state
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第二步: 把大象放进去"""
# 获取当前状态
if not state["fridge_open"]:
raise ValueError("冰箱门未打开，无法放入大象！")
# 执行当前节点业务逻辑
print("正在把大象放入冰箱…")
print("大象已放入冰箱！")
# 更新状态
state["elephant_inside"] = True
# 当前节点处理完成,更新状态
return state
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第三步: 把冰箱门带上"""
# 获取当前状态
if not state["elephant_inside"]:
print("警告：冰箱内没有大象，是否确认关闭？")
# 执行当前节点业务逻辑
print("正在关闭冰箱门…")
print("冰箱门已关闭！")
# 更新状态
state["fridge_open"] = False
# 当前节点处理完成,更新状态
return state
# 3. 构建Graph工作流图
def build_graph() -> CompiledStateGraph:
"""构建"把大象装冰箱"的工作流图"""
# 创建图对象StateGraph
graph = StateGraph(ElephantInFridgeState)
# 添加节点
graph.add_node("open_fridge", open_fridge)
graph.add_node("put_elephant", put_elephant)
graph.add_node("close_fridge", close_fridge)
# 添加起始边
graph.add_edge(START, "open_fridge")
# 添加边，将节点连接起来，形成工作流
graph.add_edge("open_fridge", "put_elephant")
graph.add_edge("put_elephant", "close_fridge")
# 添加结束边
graph.add_edge("close_fridge", END)
# 返回编译好的图
return graph.compile()
# 4. 运行Graph工作流
def run_workflow():
"""运行"把大象装冰箱"的工作流"""
# 构建图
graph = build_graph()
# 定义初始状态
initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)
# 运行图
result = graph.invoke(initial_state)
# 输出最终状态
print("\\n工作流执行完毕，最终状态:")
print(f"冰箱门状态: {'打开' if result['fridge_open'] else '关闭'}")
print(f"大象是否在冰箱内: {'是' if result['elephant_inside'] else '否'}")
# 可视化工作流图
display(Image(graph.get_graph().draw_mermaid_png()))
return result
# 执行入口
if __name__ == "__main__":
final_state = run_workflow()
下面我们分步骤对这个示例代码进行讲解。
1. 状态定义（ElephantInFridgeState）
class ElephantInFridgeState(TypedDict):
fridge_open: bool # 冰箱门开关状态 (True=开启，False=关闭)
elephant_inside: bool # 大象位置状态 (True=在冰箱内，False=在冰箱外)
（1）核心作用：作为工作流的共享内存空间，跨节点传递和持久化关键状态。
（2）状态属性：
- fridge_open：实时记录冰箱门的物理状态。
- elephant_inside：实时追踪大象的空间位置。
（3）设计特点：
- 继承自LangGraph的TypedDict，强制类型安全。
- 所有节点均可读写，实现跨节点数据共享。
- 初始状态：冰箱门关闭（False），大象在冰箱外（False）。
2. 节点功能实现
每个节点都是独立的状态处理器，遵循“获取状态→执行业务→更新状态→返回状态”的工作流范式。
1）开门节点（open_fridge）
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
print("正在打开冰箱门…")
print("冰箱门已打开！")
state["fridge_open"] = True # 更新门状态
return state
- 前置条件：无（可直接从初始状态启动）。
- 核心操作：变更fridge_open状态为True。
- 执行效果：输出操作日志，确保冰箱门开启。
2）放入大象节点（put_elephant）
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:
if not state["fridge_open"]: # 状态检查
raise ValueError("冰箱门未打开，无法放入大象！")
print("正在把大象放入冰箱…")
print("大象已放入冰箱！")
state["elephant_inside"] = True # 更新大象位置
return state
- 前置校验：强制检查fridge_open=True，否则抛出异常。
- 核心操作：变更elephant_inside状态为True。
- 关键机制：通过状态依赖确保工作流顺序逻辑。
3）关门节点（close_fridge）
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
if not state["elephant_inside"]: # 安全警告
print("警告：冰箱内没有大象，是否确认关闭？")
print("正在关闭冰箱门…")
print("冰箱门已关闭！")
state["fridge_open"] = False # 更新门状态
return state
- 安全机制：当elephant_inside=False时输出警告。
- 核心操作：重置fridge_open为False。
- 容错设计：允许在无大象时关门，但给出风险提示。
3. 工作流图构建（build_graph）
graph = StateGraph(ElephantInFridgeState)
# 节点注册
graph.add_node("open_fridge", open_fridge)
graph.add_node("put_elephant", put_elephant)
graph.add_node("close_fridge", close_fridge)
# 线性工作流定义
graph.add_edge(START, "open_fridge") # 入口 → 开门
graph.add_edge("open_fridge", "put_elephant") # 开门 → 放入大象
graph.add_edge("put_elephant", "close_fridge") # 放入大象 → 关门
graph.add_edge("close_fridge", END) # 关门 → 出口
1）图结构
创建有向无环图控制流。
2）节点关系
- START：LangGraph内置起始节点。
- 线性执行链：开门→放入大象→关门。
- END：LangGraph内置终止节点。
3）编译输出
返回可执行的CompiledStateGraph对象。
4）工作流执行（run_workflow）
def run_workflow():
graph = build_graph() # 编译图结构
initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)
result = graph.invoke(initial_state) # 执行工作流
# 状态结果输出
print(f"冰箱门状态: {'打开' if result['fridge_open'] else '关闭'}")
print(f"大象是否在冰箱内: {'是' if result['elephant_inside'] else '否'}")
# 可视化输出
display(Image(graph.get_graph().draw_mermaid_png()))
5）执行流程
（1）编译工作流图。
（2）初始化状态：fridge_open=False, elephant_inside=False。
（3）注入初始状态并触发执行。
6）输出内容
- 实时打印各节点操作日志。
- 终态数据报告。
- Mermaid格式可视化流程图（PNG渲染）。
完整的流程如图15-4所示。
图15-4 大象放进冰箱的完整流程
这里我们通过边严格定义顺序：开门→放入大象→关门，并加载了前置条件校验（开门检查）、异常安全机制（ValueError）以及风险预警（空冰箱关门警告）。
此实现完整演示了LangGraph在顺序工作流建模中的核心能力，通过状态对象实现节点间解耦，同时确保业务流程严格按顺序执行。
15.1.2 LangGraph特性与技术原理
作为LangChain生态系统的一部分，LangGraph也可以独立运行，其核心优势在于支持循环流程、细粒度可控性和内置持久性，解决了传统有向无环图框架无法处理循环逻辑的局限性。
1. 核心特性与技术原理
- 循环与分支：LangGraph允许定义包含循环和条件分支的工作流，例如代理可多次调用工具优化结果（如改进RAG管道的检索质量），适应复杂任务需求。
- 状态管理：通过节点（执行步骤）和边（执行顺序与条件）构建图结构，状态在节点间传递并动态更新。例如，节点可调用LLM或工具，条件边根据状态决定下一步操作。
- 持久性与人工干预：自动保存每一步状态，支持暂停/恢复执行（如错误恢复或时间回溯），并允许人类审核或修改代理的下一步行动，实现“人机回环（Human-in-the-Loop）”协作。
- 流式支持与集成：实时流式传输每个节点的输出（包括令牌流），并与LangChain/LangSmith无缝集成，提升开发效率。
2. 应用场景
LangGraph适用于自动化复杂业务流程：
- 客户服务：构建多轮对话机器人，提供个性化支持。
- 数据分析：代理协同检索多源信息并生成报告。
- 流程自动化：如订单处理、库存管理等需状态跟踪的场景。
- 教育/推荐系统：AI教师动态调整教学策略，或基于用户行为生成推荐。
3. 示例与部署
用户可通过Python代码定义图结构（如分类查询→生成响应→循环优化），并利用内置MemorySaver保存对话上下文，确保长期任务的连续性。LangGraph的商业平台支持生产环境部署，适合构建高可控、高可靠的智能系统。
LangGraph通过图计算模型（灵感源自Pregel和Apache Beam）实现了灵活的任务编排，为开发动态AI应用提供了底层支持，推动多代理协作与自动化演进。
15.1.3 LangGraph组件详解
前面讲解的基于“链”的模型调用与流程设计，虽然能够较好地完成任务，但是简单的链不具备循环能力，对于多分支任务，则需要一个具备更精细控制能力的框架来支持复杂场景的LLM应用。
LangGraph的出现宣布智能体的使用进入多智能体框架领域。LangGraph是基于图论运作的，提供了一种状态机的技术，可以驱动循环代理调用，实现有向有环图。因此，LangGraph有3个关键元素：
- State（状态）：一个共享的数据结构，表示应用程序的当前快照。它可以是任何Python类型，但通常是TypedDict或Pydantic BaseModel。
- Nodes（节点）：编码代理逻辑的Python函数。它们接收当前的State作为输入，执行一些计算或起到辅助作用，并返回一个更新后的State。
- Edges（边）：Python函数，基于当前的State决定接下来执行哪个Node。它们可以是条件分支或固定转换。
通过组合Nodes和Edges，可以创建复杂、循环的工作流，随着时间的推移演化State。不过，真正的强大之处在于LangGraph如何管理State。需要强调的是，Nodes和Edges仅仅是Python函数—它们可以包含LLM，也可以只是普通的Python代码。
简而言之，节点是用于工作的模块，边告诉我们接下来该做什么。
LangGraph的底层图算法使用消息传递来定义一个通用程序。当一个Node完成其操作时，它会沿着一条或多条边发送消息给其他节点。这些接收节点随后执行它们的函数，将结果消息传递给下一组节点，过程继续，且程序以离散的“超级步骤”进行。
一个超级步骤可以被视为对图节点的一次迭代。并行运行的节点属于同一超级步骤，而顺序运行的节点属于不同的超级步骤。在图执行开始时，所有节点都处于非活动状态。当一个节点在其任何输入边（或“通道”）上收到新消息（状态）时，会变为活动状态。活跃节点运行其函数并响应更新。
在每个超级步骤结束时，没有收到消息的节点会通过标记自己为非活动状态来停止投票。图执行在所有节点都非活动且没有消息传递时终止。
在定义图形时，首先要做的是定义图形的State。State包含图形的模式以及指定如何应用更新到State的reducer函数。State的模式将是所有Nodes和Edges在图中的输入模式，可以是TypedDict或Pydantic模型。所有Nodes节点将会发出对State的更新，然后通过指定的reducer函数应用这些更新。
15.1.4 LangGraph中的StateGraph
StateGraph是LangChain中的一个类，用于表示图的数据结构，并反映系统的当前状态。它类似于状态机，通过节点对状态的更新来驱动流程。
- 初始化时需定义state_schema，指定状态的结构，确保节点可以访问和更新共享状态。
- 状态图的核心作用是维护全局状态，所有节点通过读写该状态进行通信。
StateGraph的特点：
- 支持循环代理调用，适用于复杂任务（如对话代理、自动化流程）。
- 状态会随着节点的执行逐步更新，形成完整的执行轨迹。
示例代码如下：
from langgraph.graph import StateGraph
from typing import TypedDict, List, Annotated
import operator
# 定义一个State
class State(TypedDict):
input: str
all_actions: Annotated[List[str], operator.add]
# 初始化一个StateGraph对象
graph = StateGraph(State)
StateGraph是LangGraph的核心，是graph的一个类。可以通过定义一个State来初始化一个StateGraph对象。定义的State可以随时更新StateGraph对象的状态。StateGraph对象可以通过Nodes更新状态。
reducer函数是理解如何将来自节点的更新应用到State的关键。State中的每个键都有其独立的reducer函数。如果没有显式指定reducer函数，则假定对该键的所有更新都应覆盖它。
以下两个示例将展示如何使用默认的reducer函数：
from typing import TypedDict
class State(TypedDict):
foo: int
bar: list[str]
在此示例中，没有为任何键指定reducer函数。假设图的输入是{"foo": 1, "bar": ["hi"]}，再假设第一个节点返回{"foo": 2}。这被视为对状态的更新。注意，节点不需要返回整个State模式—只需要一个更新。在应用此更新后，State将变为{"foo": 2, "bar": ["hi"]}。如果第二个节点返回{"bar": ["bye"]}，那么State将变为{"foo": 2, "bar": ["bye"]}。
from typing import TypedDict, Annotated
from operator import add
class State(TypedDict):
foo: int
bar: Annotated[list[str], add]
在此示例中，我们使用Annotated类型为第二个键（bar）指定了一个reducer函数（operator.add）。注意，第一个键保持不变。假设图的输入是{"foo": 1, "bar": ["hi"]}，再假设第一个节点返回{"foo": 2}。这被视为对状态的更新。在应用此更新后，State将变为{"foo": 2, "bar": ["hi"]}。如果第二个节点返回{"bar": ["bye"]}，那么State将变为{"foo": 2, "bar": ["hi", "bye"]}。需要注意，这里的bar键是通过将两个列表加在一起更新的。
15.1.5 LangGraph中的节点
节点是图中的核心执行单元，每个节点具有唯一名称，且可以是一个函数、可运行对象（Runnable）或LangChain表达式。
- 节点接收字典类型的状态数据，返回更新后的相同结构的状态。
- 特殊节点"END"表示状态机的终止点，流程执行到此即结束。
节点的行为规则：
- 节点的输入和输出需严格遵循state_schema定义的状态结构。
- 节点可通过调用模型（如LLM）生成结果，并更新状态中的特定字段。
示例代码如下：
# 添加节点
# graph.add_node(name, value)，
# name是节点的名称str
# value是节点的值，是一个函数或者是LCEL runnable
# 一个model，暂时理解为调用大模型Reasoning的过程
def model():
return "model"
# 一个tool_executor，理解为Acting的过程
def tool_executor():
return "tool_executor"
graph.add_node("model", model)
graph.add_node("tools", tool_executor)
# 还有一个特殊的节点，叫End，表示StateGraph的结束状态
from langgraph.graph import END
在LangGraph中，节点通常是Python函数（同步或异步），其中第一个位置参数是state；第二个位置参数（可选）是config，包含可选的可配置参数（例如thread_id）。下面是添加不同的自定义节点的代码：
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph
builder = StateGraph(dict)
def my_node(state: dict, config: RunnableConfig):
print("In node: ", config["configurable"]["user_id"])
return {"results": f"Hello, {state['input']}!"}
# 第二个参数是可选的
def my_other_node(state: dict):
return state
builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)
在节点中，有以下两个特殊节点。
1. START Node（开始节点）
START Node是一个开始节点，表示将用户输入发送到图的节点。引用此节点的主要目的是确定哪些节点应首先被调用。
from langgraph.graph import START
graph.add_edge(START, "node_a")
2. END Node（结束节点）
END Node表示一个结束节点。当想表示某些边在完成后没有进一步操作时，可以引用此节点。
from langgraph.graph import END
graph.add_edge("node_a", END)
15.1.6 LangGraph中的边
边（Edges）定义了逻辑，包括如何路由以及图形如何停止。这是Agent如何工作以及不同节点如何相互通信的一个重要部分。
一个节点可以有多个传出边，此时所有这些目标节点将在下一个超级步骤中并行执行。
条件边的三要素如下：
- 上游节点：边的起点（如节点A）。
- 路由函数：根据状态值返回条件结果（如route(state)→"next_node"）。
- 状态映射：将路由函数的返回值与具体下游节点绑定。
边有以下几种关键类型：
- Normal Edges（普通边）：直接从一个节点转到下一个节点。
- Conditional Edges（条件边）：调用一个函数来确定接下来要转到的节点。
- Entry Point（入口点）：当用户输入到达时首先调用的节点。
- Conditional Entry Point（条件入口点）：调用一个函数来确定当用户输入到达时首先调用的节点。
1. Normal Edges
如果希望从节点A转到节点B，可以直接使用add_edge()方法。
graph.add_edge("node_a", "node_b")
2. Conditional Edges
如果希望有条件地路由到一条或多条边（或可选地终止），可以使用add_conditional_edges()方法。此方法接收节点的名称和routing_function（路由函数）来调用该节点后执行：
graph.add_conditional_edges("node_a", routing_function)
路由函数类似于节点，接收图的当前状态并返回一个值。默认情况下，路由函数的返回值用作状态接下来要发送到的节点的名称（或列表）。所有这些节点将在下一个超级步骤中并行运行。
graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})
3. Entry Point
Entry Point是图形启动时首先运行的节点。可以使用add_edge()方法从虚拟的START节点到要执行的第一个节点来指定进入图的位置。
from langgraph.graph import START
graph.add_edge(START, "node_a")
4. Conditional Entry Point
Conditional Entry Point允许根据自定义逻辑在不同节点之间开始。读者可以使用add_conditional_edges从虚拟的START节点来实现这一点。
from langgraph.graph import START
graph.add_conditional_edges(START, routing_function)
读者可以选择提供一个字典，将路由函数的输出映射到下一个节点的名称。
graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})
下面给出完整的使用节点的示例：
# 在创建了上面两个节点之后，可以通过边将两个节点连接起来
# LangGraph提供了3种边的类型
# Starting Edge，开始边，表示从StateGraph的开始状态到第一个节点的边
# 开始边到model节点
graph.set_entry_point("model")
# Normal Edge，普通边，表示从一个节点到另一个节点的边
# model节点到tools节点
graph.add_edge("model", "tools")
# Conditional Edges，条件边，表示从一个节点到另一个节点的边，但是这个边是有条件的
# 一个条件函数
def condition():
return "我是一个判定节点"
# 添加条件边，从model节点开始，根据condition函数的返回值，如果是end就结束，如果不是end就继续到tools节点
graph.add_conditional_edges(
"model",
condition, # 判定函数
{
"end": END,
"continue": "tools"
}
)
# 最后返回一个runnable的chain
app = graph.compile()
15.2 基于LangGraph的聊天助手搭建
在深入探讨智能体架构的演进路径时，我们需进一步解析大语言模型与流程编排框架的协同机制。基于前期构建的LangGraph智能体框架，当前研究聚焦于模型能力与流程控制的深度融合。通过将Qwen3大模型嵌入为独立计算节点，可实现语言生成能力与既有功能模块的解耦式集成。这种设计模式既保留了流程编排的灵活性，又赋予智能体即时接入先进语言模型的通道。
具体实现层面，LangGraph的节点化加载机制提供了两种典型接入范式：独立节点模式和原生接口集成。在独立节点模式中，Qwen3可作为专用处理单元挂载至流程任意环节，通过定义标准化的输入/输出接口，使复杂任务可拆解为“意图解析→工具调用→结果润色”的多阶段流水线。例如，在客服场景中，原始查询首先经由Qwen3模型进行需求分类，再由知识库检索节点执行数据查询，在返回结果前再次调用模型生成自然语言回复。这种分层处理架构显著提升了复杂任务的处理鲁棒性。
更值得关注的是LangGraph原生接口集成。通过实现框架预置的接口，我们可直接将Qwen3的推理服务封装为框架原生组件。这种集成方式不仅能支持同步/异步调用模式的灵活配置，还能利用框架内置的缓存机制和批处理优化策略。在实际部署时，可针对对话历史管理、上下文长度控制等场景进行定制化开发，例如通过状态管理模块自动维护多轮对话的记忆轨迹，或利用向量数据库实现历史记录的语义化检索。
两种接入方式在工程实践中呈现互补特性：独立节点模式适合快速验证模型能力边界，其可视化编排特性便于业务人员理解流程逻辑；而原生接口集成则更适用于生产环境，通过深度耦合框架特性可获得更优的性能表现。需要注意，随着模型迭代速度加快，框架的插件化设计使得技术升级无须改动核心流程，这为构建持续演进的智能体系统提供了基础设施保障。
15.2.1 LangGraph构建的聊天助手
在上一节中，我们演示了“把大象装进冰箱”这一完整流程，通过这个完整的工作流我们知道了LangGraph的基本使用方法。本小节将使用LangGraph完成一个聊天助手智能体的搭建工作。
1. 基于Qwen3的智能体搭建
为了适配LangGraph，我们首先需要提供可用于在智能体对话中作为对话模块的大模型Qwen3。相对于经典的Qwen3的对话框架，我们需要修正新的问答模块，同时为了简便起见，还需要将其通过类的方式进行重新组建。代码如下：
from openai import OpenAI
from langchain_core.messages import HumanMessage, AIMessage
#QwenLLM 类
class QwenLLM:
def __init__(self):
self.client = OpenAI(
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
self.system_prompt = "你是一个有用的智能助手！"
def invoke(self, messages: list):
# 将 LangChain 消息对象转换为 OpenAI 格式
openai_messages = [
{"role": "system", "content": self.system_prompt}
]
for msg in messages:
if isinstance(msg, HumanMessage):
openai_messages.append({"role": "user", "content": msg.content})
elif isinstance(msg, AIMessage):
openai_messages.append({"role": "assistant", "content": msg.content})
completion = self.client.chat.completions.create(
model="qwen-plus",
messages=openai_messages
)
return AIMessage(content=completion.choices[0].message.content)
上面这段代码定义了一个名为QwenLLM的类，用于通过OpenAI兼容接口调用阿里云的Qwen大模型。该类在初始化时配置了API密钥和基础URL，并设定了一个系统提示语。Invoke()方法用于接收一个消息列表，并将其转换为适合OpenAI接口调用的消息格式，其中用户消息和助手消息分别对应到相应的角色。最后，调用Qwen-Plus模型，返回生成的响应内容并封装成AIMessage对象。
AIMessage是langchain_core.messages模块中的一个类，用于表示由AI助手生成的消息。它通常在使用LangChain框架开发应用时出现，用来统一管理和传递对话历史中助手一方的消息。
在上面代码中，AIMessage用来封装调用Qwen大模型后返回的AI回复内容。这样做的好处是可以将用户输入（HumanMessage）和AI输出（AIMessage）统一成一致的数据结构，便于后续处理、记录或传递给其他模块继续处理，比如构建对话链、记忆管理、日志记录等。
2. 结合Qwen3和LangGraph的聊天助手智能体构建
我们沿用先前构建的大语言模型，并将其作为节点集成到LangGraph聊天助手中。此处以Qwen3作为节点进行添加，并完成相应的配置，具体代码如下：
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
from langgraph.graph import StateGraph
graph_builder = StateGraph(State)
from qwen import QwenLLM
llm = QwenLLM()
def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
for event in graph.stream({"messages": ("user", user_input)}):
for value in event.values():
print("Assistant:", value["messages"][-1].content)
上面这段代码实现了一个基于状态图的简单聊天机器人，首先定义了一个State类型，使用TypedDict和Annotated来描述包含消息列表的状态结构；然后创建了一个StateGraph状态图，初始化了一个QwenLLM模型实例，并定义了一个chatbot节点函数，该函数接收当前状态，调用LLM生成回复并返回更新后的消息列表。接着将chatbot节点设为入口点并编译成可执行图。最后进入交互循环，持续接收用户输入（输入“quit”“exit”“q”可退出），将用户消息添加到状态中，并通过graph.stream流式处理状态图，实时打印助手的回复内容。整个流程实现了用户与基于大语言模型的聊天机器人的持续对话功能。
15.2.2 原生LangGraph接口对Qwen3的调用
前面我们通过使用Qwen3自带的接口来连接与使用LangGraph。此外，我们还可以使用LangGraph接口来调用Qwen3，代码如下：
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # 假设的 API 地址
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
agent = create_react_agent(
model=llm,
tools=[get_weather],
prompt="你是一个个人助手"
)
# Run the agent
response = agent.invoke(
{"messages": [{"role": "user", "content": "今天shanghai的天气怎么样"}]})
for message in response['messages']:
print(message)
print("****")
print(message.content)
print("—————-")
运行上面代码，结果如下：
content='今天shanghai的天气怎么样' additional_kwargs={} response_metadata={} id='39679bab-bf29-488a-b4d1-c0d198844df7'
****
今天shanghai的天气怎么样
—————-
content='' additional_kwargs={'tool_calls': [{'id': 'call_191569dd5ac74905a22816', 'function': {'arguments': '{"city": "Shanghai"}', 'name': 'get_weather'}, 'type': 'function', 'index': 0}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 18, 'prompt_tokens': 171, 'total_tokens': 189, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None} id='run–6e3a2c4f-e1a0-48c1-82b7-2474df8c5aca-0' tool_calls=[{'name': 'get_weather', 'args': {'city': 'Shanghai'}, 'id': 'call_191569dd5ac74905a22816', 'type': 'tool_call'}] usage_metadata={'input_tokens': 171, 'output_tokens': 18, 'total_tokens': 189, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
****
—————-
content='Shanghai 天气晴朗!' name='get_weather' id='e1a73c6c-a071-49b4-920a-fd1170ff0e98' tool_call_id='call_191569dd5ac74905a22816'
****
Shanghai 天气晴朗!
—————-
content='今天上海的天气是晴朗的！希望你有美好的一天！' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 203, 'total_tokens': 220, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run–6879e139-14c0-4e8a-a22e-ae52f4b7740d-0' usage_metadata={'input_tokens': 203, 'output_tokens': 17, 'total_tokens': 220, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
****
今天上海的天气是晴朗的！希望你有美好的一天！
—————-
上面结果展示了模型在运行过程中对内容的处理流程，其中content表示交互的文本内容；tool_calls表示使用工具调用方法完成对应的工具使用。
15.2.3 一般智能体运行范式详解
接下来，我们对LangGraph文本处理部分进行拆解。一个简单的文本处理过程如下：
（1）读取用户输入。
（2）把输入发给LLM，让它生成回复。
（3）如果LLM给出了工具调用的指令，就执行这些工具（比如tool_calls命令）。
（4）把工具执行结果反馈给LLM。
（5）重复这个循环。
LangGraph的Agent架构本质是构建了一个自迭代的认知增强回路：当LLM遇到知识边界时，通过工具调用扩展能力边界，工具执行结果作为新输入反哺LLM，形成“感知-决策-执行-反馈”的闭环。这种设计巧妙地将LLM的符号推理能力与确定性工具相结合，解决了纯LLM的问题。
我们使用代码实现这个处理过程，示例如下：
def loop(llm):
msg = user_input()
while True:
output, tool_calls = llm(msg)
print("Agent: ", output)
if tool_calls:
msg = [ handle_tool_call(tc) for tc in tool_calls ]
else:
msg = user_input()
上面代码成功地将Agent转换成一个“LLM+工具执行+反馈调整”的小闭环，虽然简单但非常高效。
我们使用这个模型模式来完成Agent工具调用的范式，代码如下：
from langchain.tools import tool
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
# 工具函数定义（使用LangChain的@tool装饰器）
@tool
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
).bind_tools([get_weather])
# 系统提示初始化
system_prompt = "你是一个个人助手"
messages = [{"role": "system", "content": system_prompt}]
def user_input():
"""获取用户输入"""
return input("User: ")
def handle_tool_call(tool_call):
"""执行工具调用并返回结果消息"""
func_name = tool_call['name']
args = tool_call['args']
# 查找匹配的工具
if func_name == "get_weather":
result = get_weather.invoke(args)
return ToolMessage(
content=result,
tool_call_id=tool_call['id'],
name=func_name
)
return ToolMessage(
content=f"未知工具: {func_name}",
tool_call_id=tool_call['id'],
name=func_name
)
def loop(llm):
"""主循环处理交互逻辑"""
msg = user_input()
messages.append(HumanMessage(content=msg))
while True:
# 调用模型获取响应
response = llm.invoke(messages)
# 提取模型输出和工具调用
output = response.content
tool_calls = response.tool_calls if hasattr(response, 'tool_calls') else []
print(f"Agent: {output}")
if tool_calls:
# 处理所有工具调用
tool_responses = [handle_tool_call(tc) for tc in tool_calls]
messages.append(response) # 添加AI消息
messages.extend(tool_responses) # 添加工具结果
print(f"工具调用结果: {[r.content for r in tool_responses]}")
else:
# 显示最终回复并等待新输入
messages.append(response)
msg = user_input()
messages.append(HumanMessage(content=msg))
# 启动交互循环
loop(llm)
这种架构标志着从单次问答到持续认知进程的转变：
- 传统Chat：input→LLM→output。
- LangGraph范式：(input+state)→[LLM+Tools]→(output+new_state)。
最终形成具备自我演进能力的对话实体，其核心价值在于将离散的交互转换为持续的价值累积过程，每次工具调用都在扩展Agent的认知边界。当循环终止时，系统输出的不仅是答案，也是经过工具验证的知识轨迹。输出结果如下：
User: hi
Agent: Hello! How can I assist you today?
User: 今天shanghai的天气如怎么样
Agent:
工具调用结果: ['Shanghai 天气晴朗!']
Agent: 今天上海的天气是晴朗的！
User:
在解析这段LangGraph实现逻辑时，核心在于理解消息传递链与工具调用的协同机制。代码中的loop()函数构建了交互主循环，其设计精髓体现在消息队列的动态维护上。每次的用户输入都会被封装为HumanMessage对象追加到messages列表，这个列表本质上是会话上下文的完整记录。当调用llm.invoke(messages)时，模型会基于完整对话历史生成响应。这种设计确保了上下文连贯性，避免了传统对话系统常见的上下文丢失问题。
这种工具调用处理流程展现了典型的链式调用模式。当模型返回包含tool_calls的响应时，系统会遍历每个工具调用指令，通过handle_tool_call()函数进行解析执行。这里值得注意的是工具路由机制：通过检查tool_call['name']与已注册工具的匹配关系，实现了灵活的工具扩展能力。在实际开发中，这种模式可轻松扩展支持数据库查询、API调用等复杂操作，只需按规范注册新的工具函数即可。
下面是使用预创建方式完成的LangGraph智能体代码：
from langchain.tools import tool
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
def prompt(state):
user_name = "xiaohua"
system_msg = f"You are a helpful assistant. Address the user as {user_name}."
return [{"role": "system", "content": system_msg}] + state["messages"]
# 工具函数定义（使用LangChain的@tool装饰器）
@tool
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
).bind_tools([get_weather])
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(
model=llm, # 直接传入 LLM 实例
tools=[get_weather],
prompt=prompt
)
result = agent.invoke(HumanMessage(content="你好"))
代码运行结果请读者可以自行验证。
15.2.4 基于LangGraph与Qwen3完成聊天助手
接下来，我们使用LangGraph结合Qwen3来完成聊天助手。为了加深读者对实现过程的理解，我们将在聊天助手中额外添加天气查询工具，并通过条件节点的判断进行工具调用。代码如下：
from langchain.tools import tool
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"{city} 天气晴朗!"
tools = [get_weather]
# 工具节点
tool_node = ToolNode(tools=tools)
# 定义状态类型
from typing import Annotated
from typing_extensions import TypedDict
class State(TypedDict):
messages: Annotated[list, lambda old, new: old + new]
# Chatbot 节点：LLM 响应（不绑定工具）
def chatbot(state: State):
response = llm.invoke(state["messages"])
return {"messages": [response]}
# 构建图
from langgraph.graph import StateGraph,END
from langgraph.prebuilt import ToolNode, tools_condition
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 路由逻辑：LLM 判断是否调用工具
graph_builder.set_entry_point("chatbot")
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("chatbot", END)
# 编译
graph = graph_builder.compile()
# 执行对话
config = {"configurable": {"thread_id": "1"}}
def stream_graph_updates(user_input: str):
events = graph.stream(
{"messages": [{"role": "user", "content": user_input}]},
config,
#stream_mode="values",
)
for event in events:
for value in event.values():
print("Assistant:", value["messages"][-1].content)
# 运行主循环
while True:
try:
user_input = input("User: ")
if user_input.lower() in ["exit", "quit", "q"]:
print("Goodbye!")
break
stream_graph_updates(user_input)
except Exception as e:
print(f"错误：{e}")
break
读者可以自行运行代码验证结果。
15.2.5 进阶智能体构建
在上一小节中，我们使用基础的LangGraph组件完成了智能体的构建，其中自定义条件选择的实现方法如下：
# 更精确的工具调用检测
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state['messages']
last_message = messages[-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_message, AIMessage) and hasattr(last_message, 'tool_calls') and last_message.tool_calls:
print("检测到工具调用，转到工具节点")
return "tools"
# 其他情况结束流程
return END
在上面代码中，我们额外添加了对消息进行检测的方法，并返回调用工具。
完整的实现代码如下：
from typing import Literal
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""这是一个对天气进行搜索并得到返回结果的工具."""
return f"{city} 天气晴朗!"
tools = [get_weather]
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
# 工具节点
tool_node = ToolNode(tools=tools)
# 更精确的工具调用检测
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state['messages']
last_message = messages[-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_message, AIMessage) and hasattr(last_message, 'tool_calls') and last_message.tool_calls:
print("检测到工具调用，转到工具节点")
return "tools"
# 其他情况结束流程
return END
# 使用字典键访问消息
def chatbot(state: MessagesState):
# 绑定工具到LLM
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke(state['messages'])
return {"messages": [response]}
# 构建图
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 设置入口点
graph_builder.set_entry_point("chatbot")
# 正确设置条件边
graph_builder.add_conditional_edges(
"chatbot",
should_continue,
{
"tools": "tools", # 如果调用工具，转到工具节点
END: END # 否则结束
}
)
# 工具节点执行后回到chatbot
graph_builder.add_edge("tools", "chatbot")
# 编译图
graph = graph_builder.compile()
# 执行图
final_state = graph.invoke(
{"messages": [HumanMessage(content="上海的天气怎么样?")]},
config={"configurable": {"thread_id": 929}}
)
# 输出结果
result = final_state["messages"][-1].content
print("最终结果:", result)
运行上面代码，结果如下：
检测到工具调用，转到工具节点
最终结果: 上海的天气目前是晴朗的。
此外，我们还可以通过LangGraph添加额外的记忆力组件，从而使得智能体在运行时能够记忆原有的内容。添加记忆力组件的代码如下：
# 初始化内存以在图运行之间持久化状态
checkpointer = MemorySaver()
# 编译图
# 这将编译成一个LangChain可运行对象
# 这意味着你可以像使用其他可运行对象一样使用它
# 注意，我们（可选地）在编译图时传递内存
graph = graph_builder.compile(checkpointer=checkpointer)
完整代码如下：
from typing import Literal
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""这是一个对天气进行搜索并得到返回结果的工具."""
return f"{city} 天气晴朗!"
tools = [get_weather]
# 初始化模型（绑定工具）
llm = ChatOpenAI(
model_name="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
)
# 工具节点
tool_node = ToolNode(tools=tools)
# 更精确的工具调用检测
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state['messages']
last_message = messages[-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_message, AIMessage) and hasattr(last_message, 'tool_calls') and last_message.tool_calls:
print("检测到工具调用，转到工具节点")
return "tools"
# 其他情况结束流程
return END
# 使用字典键访问消息
def chatbot(state: MessagesState):
# 绑定工具到LLM
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke(state['messages'])
return {"messages": [response]}
# 构建图
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 设置入口点
graph_builder.set_entry_point("chatbot")
# 正确设置条件边
graph_builder.add_conditional_edges(
"chatbot",
should_continue,
{
"tools": "tools", # 如果调用工具，转到工具节点
END: END # 否则结束
}
)
# 工具节点执行后回到chatbot
graph_builder.add_edge("tools", "chatbot")
# 初始化内存以在图运行之间持久化状态
checkpointer = MemorySaver()
# 编译图
# 这将编译成一个LangChain可运行对象
# 这意味着你可以像使用其他可运行对象一样使用它
# 注意，我们（可选地）在编译图时传递内存
graph = graph_builder.compile(checkpointer=checkpointer)
# 执行图，使用可运行对象
final_state = graph.invoke(
{"messages": [HumanMessage(content="上海的天气怎么样?")]},
config={"configurable": {"thread_id": 929}}
)
# 从 final_state 中获取最后一条消息的内容
result = final_state["messages"][-1].content
print(result)
final_state = graph.invoke(
{"messages": [HumanMessage(content="我问的那个城市?")]},
config={"configurable": {"thread_id": 929}}
)
result = final_state["messages"][-1].content
print(result)
读者可以自行运行代码验证结果。
15.3 本章小结
在本章中，我们深入探索了LangGraph的核心应用，通过智能体案例展示了其基本组成结构和运行机制。案例演示清晰呈现了LangGraph如何驱动智能体完成文本交互和工具调用，展现了其将复杂流程转换为可编程图的独特价值。在Qwen3模型的集成方面，我们既可通过其原生API接口直接调用，也能利用LangGraph的统一调度框架实现更灵活的流程控制，这种双重接入方式为开发者提供了充分的自由度。
在工具调用方面，我们不仅实现了基础功能，而且通过自定义条件判断和工具节点设计，创建了智能化的触发机制—当用户输入包含“天气”等关键词时，系统能自动生成工具调用请求，这种设计显著提升了交互的自然度和效率。整个过程展示了LangGraph如何将多步骤、多工具的工作流转换为可管理的状态转移过程，其中状态对象的动态更新机制确保了对话上下文的连贯性。
通过实践验证，LangGraph的核心优势在于其将复杂智能体工作流抽象为节点和边的能力。这种范式转换使得开发者能够像设计电路图一样编辑AI行为，将LLM的认知能力与确定性工具无缝结合。工具调用不再只是功能扩展手段，更成为智能体与环境交互的感知延伸。当天气查询工具返回“晴朗”时，这不仅是数据反馈，更是智能体对现实世界认知的更新。
