LangGraph入门知识学习笔记

作者：CSDN博客
LangGraph概述

是什么：基于 LangChain 构建的、面向智能体多轮交互 / 状态持久化 / 分支并行执行的
图结构工作流框架，即LangGraph = LangChain + 图编排 + 状态机
LangGraph 是LangChain 生态的一部分，专门用于构建基于大模型（LLM）的复杂、有状态、多智能体应用的框架。核心思想是将应用的工作流程抽象为一个有向图结构，通过节点和边来定义任务的执行步骤和逻辑流，它是对LangChain核心思想的延伸和升级，它把基础单元从”链”换成了”图（Graph）”。相比传统的线性执行模式，LangGraph 支持条件分支、循环、并行等复杂控制流，能够实现状态持久化、断点续跑、时间旅行、人机协作等高级功能，并提供了多智能体协作、层级架构等多种架构模式。
能用来做什么：对于一步步的线性任务适合langchain，但是对于非一步步线性任务的执行，langgraph比较适合
Chain太流水线，无法优雅地处理循环和条件分支，不适合复杂任务。
Agent太自由，像个黑箱，难以控制、调试和保证稳定性。
LangGraph 提供了强大的状态管理机制，允许 Agent 在不同节点之间传递和维护信息，从而实现长期的记忆和多轮对话能力。 通过定义节点和边，可以精确控制 Agent 的执行逻辑，包括条件分支、循环和并行执行等

LangGraph 能够无缝集成各种外部工具（如搜索引擎、数据库、API 等），让 Agent 能够获取实时信息、执行特定操作，极大地扩展了 LLM 的能力边界。

图结构使得 Agent 的运行路径清晰可见，便于理解 Agent 的决策过程，并在出现问题时进行快速定位和调试。

模块化与可复用性。每个节点都可以是一个独立的、可复用的组件，维护性高且易于扩展。通过子图机制，复杂的工作流可以被分解为多个可独立开发和测试的模块，提高了开发和测试效率

图的构建流程

初始化一个StateGraph实例
加节点
定义边，将所有的节点连接起来
设置特殊节点，入口和出口（可选）
编译图
执行工作流
总结

LangGraph的灵魂：State(状态) + Nodes(节点) + Edges(边) + Graph(图)
入门案例

1. """
2. LangGraph 简单案例HelloWorld：
3. 构建一个最小的有向图，流程是：START → 模型节点 → END
5. LangGraph的灵魂：State(状态) + Nodes(节点) + Edges(边) + Graph(图)
6. """
8. import uuid
9. from typing import TypedDict, Annotated, List
10. from langgraph.graph import StateGraph, START, END
11. from langgraph.graph.message import add_messages
12. import os
13. from langchain.chat_models import init_chat_model
14. from langchain_core.messages import HumanMessage
17. # ========== 1. 定义状态（State） ==========
18. # 存储对话消息
19. class AtguiguState(TypedDict):
20.     # messages 是一个消息列表，Annotated + add_messages 表示支持自动追加消息
21.     messages: Annotated[List, add_messages]
23. # ========== 2. 定义大模型 ==========
24. llm = init_chat_model(
25.     model="qwen-plus",
26.     model_provider="openai",
27.     api_key=os.getenv("aliQwen-api"),
28.     base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
29. )
31. # ========== 3. 定义节点函数 ==========
32. # 节点：调用大模型，并把回复加入到 state["messages"] 里
33. def model_node(state: AtguiguState):
34.     reply = llm.invoke(state["messages"])   # 输入历史消息，调用模型
35.     return {"messages": [reply]}            # 返回新消息，自动加到 state
37. # ========== 4. 构建图结构 ==========
38. graph = StateGraph(AtguiguState)            # 初始化图，指定 State 类型
40. graph.add_node("model", model_node)         # 添加一个节点，名字叫 "model"
42. graph.add_edge(START, "model")      # 从 START 到 "model"
43. graph.add_edge("model", END)        # 从 "model" 到 END
44. # 打印图的边和节点信息
45. #print(graph.edges)
46. print()
47. #print(graph.nodes)
49. # ========== 5. 编译==========
50. app = graph.compile()
52. # ========== 6. 运行 ==========
53. #result = app.invoke({"messages": [HumanMessage(content="请用一句话解释什么是 LangGraph。")]})
54. result = app.invoke({"messages": "请用一句话解释什么是 LangGraph。"})
56. # 打印模型的最后一条回复
57. print("模型回答：", result["messages"][-1].content)
59. print()
60. # =========================
61. #1. 打印图的ascii可视化结构
62. print(app.get_graph().print_ascii())
63. print("="*50)
65. #2. 打印图的Mermaid代码可视化结构并通过https://www.processon.com/mermaid编辑器查看
66. print(app.get_graph().draw_mermaid())
67. print("="*50)
69. #3. 生成 PNG并写入文件
70. # png_bytes = app.get_graph().draw_mermaid_png()
71. # output_path = "langgraph" + str(uuid.uuid4())[:8] + ".png"
72. # with open(output_path, "wb") as f:
73. #     f.write(png_bytes)
74. # print(f"图片已生成：{output_path}")

复制代码

Graph API

Graph API之Graph(图)

图是一种由节点和边组成的用于描述节点之间关系的数据结构，分为无向图和有向图，有向图是带有方向的图。LangGraph通过有向图定义AI工作流中的执行步骤和执行顺序，从而实现复杂、有状态、可循环的应用程序逻辑。

Graph API之State(状态)

在LangGraph中，State是一个贯穿整个工作流执行过程中的共享数据的结构，代表当前快照，它存储了从工作流开始到结束的所有必要的信息（历史对话、检索到的文档、工具执行结果等），
在各个节点中共享，且每个节点都可以修改。
状态包含两部分：
一是图的模式（schema），
二是“规约函数”（reducer functions）；后者指明如何把更新应用到状态上。
Schema

是langgraph应用的核心数据结构，定义了应用的所有数据，在所有节点穿通，并且被持久化报错保存
构成三要素

state_schema：图的完整内部状态，包含了所有节点可能读写的字段，必须指定，不能为空
input_schema：定义图接受什么输入，是 state_schema 的子集，可选参数，如果不指定，默认等于 state_schema，限制图的输入接口，只能传入这些字段
output_schema：定义图返回什么输出，是 state_schema 的子集，可选参数，如果不指定，默认等于 state_schema，限制图的输出接口，只返回这些字段
注：State可以是TypedDict类型，也可以是pydantic中的BaseModel类型
Reducers

是什么：规约函数决定了节点产生的更新如何作用到 State。State 中的每个字段都拥有自己的独立规约函数。
如果未显式指定，则默认所有对该字段的更新都会直接覆盖旧值。
规约函数有多种类型，首先是默认类型，简单来说就是状态合并策略（Reducers）
Reducer常用函数

default：未指定Reducer时使用覆盖更新
add_messages：用于消息列表追加
operator.add：将元素追加到现有元素中，支持列表、字符串、数值类型的追加
operator.mul：用于数值相乘(此处因为设计策略原因，会乘以一个默认值0.0导致结果一直为0，采用自定义Reducer可以解决次问题)
自定义Reducer：支持用户自定义合并逻辑
Graph API之Node(节点)

是什么

在LangGraph中，节点(Node)就是是Python函数（可以是同步的，也可以是异步的），它们接受以下参数：
Ø state：图的状态
Ø config：一个RunnableConfig对象，包含诸如thread_id之类的配置信息以及诸如tags之类的跟踪信息
Ø runtime：一个Runtime对象，包含运行时context以及其他信息，如store和stream_writer
定义好node函数后，使用add_node方法将这些节点添加到图中。如果在向图中添加节点时未指定名称，系统会为其分配一个与函数名相同的默认名称。
简单来说：Node是LangGraph中的一个基本处理单元，代表工作流中的一个操作步骤，
可以是一个Agent、调用大模型、工具或一个函数（说白了就是绑定一个python函数，具体逻辑可以干任何事情）
Node的设计原则：

单一职责原则：每个节点应该只负责一项职责，避免功能过于复杂
无状态设计：节点本身不应该保存状态，所有数据都通过输入状态传递
幂等性：相同的输入应该产生相同的输出，确保可重试性
可测试性：节点逻辑应该易于单元测试
节点缓存Node Caching

是什么

LangGraph支持基于节点输入对任务/节点进行缓存。使用缓存的方法如下：
编译图（或指定入口点）时指定缓存。
为节点指定缓存策略。每个缓存策略支持：
Ø key_func用于根据节点的输入生成缓存键。
Ø ttl，即缓存的生存时间（以秒为单位）。如果未指定，缓存将永不过期。
特性

缓存键与命中：
当一个节点开始执行时，系统会使用其配置的 key_func 根据当前节点的输入数据生成一个唯一的键。LangGraph 会检查缓存中是否存在这个键。

如果存在（缓存命中），则直接返回之前存储的结果，跳过该节点的实际执行。
如果不存在（缓存未命中），则正常执行节点函数，并将结果与缓存键关联后存入缓存。

缓存有效期：
ttl 参数能控制缓存的有效期。
错误处理和重试机制

LangGraph还提供了错误处理和重试机制来指定重试次数、重试间隔、重试异常等，用于保证系统的可靠性

为节点添加重试策略，需要在add_node中设置retry_policy参数。retry_policy参数接受一个RetryPolicy命名元组对象。默认情况下，retry_on参数使用default_retry_on函数，该函数会在遇到任何异常时重试

Graph API之Edge(边)

是什么

Edge定义了节点之间的连接和执行顺序，以及不同节点之间是如何通讯的，
一个节点可以有多个出边（指向多个节点），多个节点也可以指向同一个节点（Map-Reduce）
代码说明

Edge定义了节点之间的连接和执行顺序，以及不同节点之间是如何通讯的，
一个节点可以有多个出边（指向多个节点），多个节点也可以指向同一个节点（Map-Reduce）

如下是添加边的代码:

关键类型

Graph API之Send/Command/Runtime context

概述

Send和Command是两种用于实现高级工作流控制的核心机制，用于支持动态地决定下一步执行哪个节点
Send

是什么：动态创建多个执行分支，实现并行处理，每个Send对象都指定了一个执行目标节点和传递给该节点的参数，
LangGraph会并行执行所有的这些任务,常用在Map-Reduce的场景，并行执行多个子节点并最终汇总到一个总节点。

核心思想：拆分任务 →并行执行 → 统一汇总结果
通俗举例：统计全国所有人的年龄分布，Map 阶段就是「把全国按省份拆分成 34 个分片，
每个省份独立统计本省的年龄→人数（KV 对：20 岁→100 万，21 岁→98 万）」
Send 接受两个参数：第一个是节点的名称，第二个是要传递给该节点的状态
代码案例

1. """
2. SendDemo.py
3. LangGraph Map-Reduce 模式演示
4. 通过使用 Send 对象，LangGraph 提供了一种优雅的方式来实现这种动态图结构，
5. 使得我们可以根据运行时状态来决定执行路径。
7. 解释：
8. （1）首先执行 generate_subjects主题列表节点，生成主题列表：['猫', '狗', '程序员']
9. （2）然后通过条件边函数 map_subjects_to_jokes 为每个主题创建一个 Send 对象
10. （3）make_joke 节点被并行执行3次，每次处理一个主题
11. （4）最终将所有生成的笑话合并到一个列表中
12. 这种模式非常适合处理动态数量的任务
13. """
15. from typing import Annotated, List, Sequence
16. from typing_extensions import TypedDict
17. from langgraph.graph import StateGraph, START, END
18. from langgraph.types import Send
21. # 定义状态
22. class OverallState(TypedDict):
23.     subjects: List[str]
24.     jokes: Annotated[List[str], lambda x, y: x + y]  # 使用列表合并的方式
27. # 第一个节点：生成需要处理的主题列表
28. def generate_subjects(state: OverallState) -> dict:
29.     """生成需要处理的主题列表"""
30.     print("执行节点: generate_subjects")
31.     subjects = ["猫", "狗", "程序员"]
32.     print(f"生成主题列表: {subjects}")
33.     return {"subjects": subjects}
36. # Map节点：为每个主题生成笑话
37. def make_joke(state: OverallState) -> dict:
38.     """为单个主题生成笑话"""
39.     subject = state.get("subject", "未知")
40.     print(f"执行节点: make_joke，处理主题: {subject}")
42.     # 根据主题生成相应笑话
43.     jokes_map = {
44.         "猫": "为什么猫不喜欢在线购物？因为它们更喜欢实体店！",
45.         "狗": "为什么狗不喜欢计算机？因为它们害怕被鼠标咬！",
46.         "程序员": "为什么程序员喜欢洗衣服？因为他们在寻找bugs！",
47.         "未知": "这是一个关于未知主题的神秘笑话。"
48.     }
50.     joke = jokes_map.get(subject, f"这是一个关于{subject}的即兴笑话。")
51.     print(f"生成笑话: {joke}")
52.     return {"jokes": [joke]}
55. # 条件边函数：根据主题列表生成Send对象列表
56. def map_subjects_to_jokes(state: OverallState) -> Sequence[Send]:
57.     """将主题列表映射到joke生成任务"""
58.     print("执行条件边函数: map_subjects_to_jokes")
59.     subjects = state["subjects"]
60.     print(f"映射主题到joke任务: {subjects}")
62.     # 为每个主题创建一个Send对象，指向make_joke节点
63.     # 每个Send对象包含节点名称和传递给该节点的状态
64.     send_list = [Send("make_joke", {"subject": subject}) for subject in subjects]
65.     print(f"生成Send对象列表: {send_list}")
66.     return send_list
69. def main():
70.     """演示Map-Reduce模式"""
71.     print("=== Map-Reduce 模式演示 ===\n")
73.     # 创建图
74.     builder = StateGraph(OverallState)
76.     # 添加节点
77.     builder.add_node("generate_subjects", generate_subjects)
78.     builder.add_node("make_joke", make_joke)
80.     # 添加边
81.     builder.add_edge(START, "generate_subjects")
83.     # 添加条件边，使用Send对象实现map-reduce
84.     builder.add_conditional_edges(
85.         "generate_subjects",  # 源节点
86.         map_subjects_to_jokes  # 路由函数，返回Send对象列表
87.     )
89.     # 从make_joke到结束
90.     builder.add_edge("make_joke", END)
92.     # 编译图
93.     graph = builder.compile()
94.     print(graph.get_graph().print_ascii())
96.     # 执行图
97.     initial_state = {"subjects": [], "jokes": []}
98.     print("初始状态:", initial_state)
99.     print("\n开始执行图...")
101.     result = graph.invoke(initial_state)
102.     print(f"\n最终结果: {result}")
104.     print("\n=== 演示完成 ===")
107. if __name__ == "__main__":
108.     main()

复制代码

Command

Command：不仅可以指定下一个节点，还支持更新状态、处理中断恢复，以及在嵌套图之间导航。
常用于复杂的人机交互（Human-in-the-loop）和多智能体协同工作中智能体与智能体之间交接执行权（handoffs）

Command与条件边的区别是：

条件边只会路由下一个node节点，

而Command不仅路由下一个node节点，还支持状态更新，如果需要同时更新状态和路由到不同的节点时，则使用 Command。
代码案例：

1. """
2. CommandDemo.py | LangGraph 1.0.6 正式版
3. Command 基础演示：状态更新+流程控制+动态路由
4. """
5. from typing import Annotated
6. from typing_extensions import TypedDict
7. from langgraph.graph import StateGraph, START, END
8. from langgraph.types import Command
10. # 全局常量：统一递归限制，便于维护
11. RECURSION_LIMIT = 50
13. # 定义状态
14. class AgentState(TypedDict):
15.     messages: Annotated[list, lambda x, y: x + y]  # 自动合并消息
16.     current_agent: str
17.     task_completed: bool
19. # 决策代理（核心路由节点）
20. def decision_agent(state: AgentState) -> Command[AgentState]:
21.     """根据消息内容路由代理，任务完成则直接终止"""
22.     print("执行节点: decision_agent")
23.     # 优先终止流程（核心防循环逻辑）
24.     if state["task_completed"]:
25.         print("✅ 检测到任务已完成，直接终止流程")
26.         return Command(
27.             update={"messages": [("system", "所有任务处理完成，流程正常结束")]},
28.             goto=END
29.         )
30.     # 提取消息文本（兼容空消息）
31.     last_message = state["messages"][-1] if state["messages"] else ("", "")
32.     last_msg_content = last_message[1]
33.     print(f"最新消息文本: {last_msg_content}")
35.     # 动态路由
36.     if "数学" in last_msg_content:
37.         print("✅ 检测到数学任务，路由到数学代理")
38.         return Command(
39.             update={"messages": [("system", "路由到数学代理")], "current_agent": "math_agent"},
40.             goto="math_agent"
41.         )
42.     elif "翻译" in last_msg_content:
43.         print("✅ 检测到翻译任务，路由到翻译代理")
44.         return Command(
45.             update={"messages": [("system", "路由到翻译代理")], "current_agent": "translation_agent"},
46.             goto="translation_agent"
47.         )
48.     else:
49.         print("❌ 未识别任务类型，标记任务完成并终止")
50.         return Command(
51.             update={"messages": [("system", "任务完成")], "task_completed": True},
52.             goto=END
53.         )
56. # 数学代理（业务节点）
57. def math_agent(state: AgentState) -> Command[AgentState]:
58.     """处理数学计算任务，完成后返回决策代理"""
59.     print("执行节点: math_agent")
60.     result = "2 + 2 = 4"
61.     print(f"计算结果: {result}")
62.     return Command(
63.         update={
64.             "messages": [("assistant", f"数学计算结果: {result}")],
65.             "current_agent": "decision_agent",
66.             "task_completed": True
67.         },
68.         goto="decision_agent"
69.     )
72. # 翻译代理（业务节点）
73. def translation_agent(state: AgentState) -> Command[AgentState]:
74.     """处理中英翻译任务，完成后返回决策代理"""
75.     print("执行节点: translation_agent")
76.     translation = "Hello -> 你好"
77.     print(f"翻译结果: {translation}")
78.     return Command(
79.         update={
80.             "messages": [("assistant", f"翻译结果: {translation}")],
81.             "current_agent": "decision_agent",
82.             "task_completed": True
83.         },
84.         goto="decision_agent"
85.     )
88. def main():
89.     """演示Command基础用法：状态更新+动态路由+流程终止"""
90.     print("=== Command 基础演示（LangGraph 1.0.6）===\n")
92.     # 1. 构建状态图
93.     builder = StateGraph(AgentState)
94.     builder.add_node("decision_agent", decision_agent)
95.     builder.add_node("math_agent", math_agent)
96.     builder.add_node("translation_agent", translation_agent)
98.     # 2. 定义边（完整节点关系）
99.     builder.add_edge(START, "decision_agent")
100.     builder.add_edge("math_agent", "decision_agent")
101.     builder.add_edge("translation_agent", "decision_agent")
102.     builder.add_edge("decision_agent", END)
104.     # 3. 编译图
105.     graph = builder.compile()
107.     # 测试1：数学任务
108.     print("【测试1: 数学任务】")
109.     initial_state = {"messages": [("user", "我需要计算数学题")], "current_agent": "user", "task_completed": False}
110.     print("初始状态:", initial_state)
111.     result = graph.invoke(initial_state, recursion_limit=RECURSION_LIMIT)
112.     print("最终状态(简化):", {k: v for k, v in result.items() if k != "messages"})  # 简化输出
113.     print("\n" + "-" * 50 + "\n")
115.     # 测试2：翻译任务
116.     print("【测试2: 翻译任务】")
117.     initial_state = {"messages": [("user", "我需要翻译文本")], "current_agent": "user", "task_completed": False}
118.     print("初始状态:", initial_state)
119.     result = graph.invoke(initial_state, recursion_limit=RECURSION_LIMIT)
120.     print("最终状态(简化):", {k: v for k, v in result.items() if k != "messages"})
121.     print("\n" + "-" * 50 + "\n")
123.     # 测试3：未识别任务
124.     print("【测试3: 未识别任务类型】")
125.     initial_state = {"messages": [("user", "你好")], "current_agent": "user", "task_completed": False}
126.     print("初始状态:", initial_state)
127.     result = graph.invoke(initial_state, recursion_limit=RECURSION_LIMIT)
128.     print("最终状态(简化):", {k: v for k, v in result.items() if k != "messages"})
130.     # 新增：可视化图结构（教学演示必备）
131.     print("\n=== 图结构可视化 ===")
132.     print(graph.get_graph().draw_mermaid())
135. if __name__ == "__main__":
136.     main()

复制代码

Runtime context运行时上下文

高级特性

A2A：AI架构的协议革命


原文地址：https://blog.csdn.net/qq_39004307/article/details/158455688