LangGraph 指南篇-基础控制

作者：找了一圈尾巴
前言

       本文聚焦 LangGraph 的基础控制方法，通过详细讲解与代码示例实现，帮助读者快速掌握其基础控制的写法，为后续开发 LangGraph Agent 工作流奠定基础。
状态管理

        在 LangGraph 中，状态管理是构建复杂工作流的核心机制。节点通过返回值更新图状态，状态更新规则由 State 对象的定义决定。
一、状态定义方式

TypedDict（适合简单场景）：轻量型定义，适合结构简单、无需复杂验证的状态。

1. # TypedDict 示例（简单场景）
2. from typing import TypedDict
4. class InputState(TypedDict):
5.     user_input: str

复制代码

BaseModel（适合复杂场景）：基于 Pydantic，支持类型验证、约束定义和嵌套结构，适合复杂状态。

1. # BaseModel 示例（复杂场景）
2. from pydantic import BaseModel
4. class State(BaseModel):
5.     # 1. 工具调用相关消息列表（精确类型+长度约束）
6.     messages: List[BaseMessage] = Field(
7.         ...,  # 必须提供初始值，不允许为空
8.         min_length=1,  # 至少包含1条消息（原300可能为笔误，消息数量通常不需要这么大，调整为合理值）
9.         description="工具调用的完整消息记录，包含用户指令、工具输入/输出、系统提示等，每条消息需为BaseMessage子类（如HumanMessage、ToolMessage）"
10.     )
11.    
12.     # 2. 元数据（结构化键值对，明确包含的信息）
13.     metadata: Dict[str, Union[str, int, datetime, bool]] = Field(
14.         default_factory=lambda: {
15.             "session_id": "",  # 会话唯一标识
16.             "step": 0,  # 当前流程步骤（从0开始）
17.             "last_tool": None,  # 上一次调用的工具名称（如"search"）
18.             "is_finished": False,  # 流程是否已结束
19.             "created_at": datetime.now()  # 状态创建时间
20.         },
21.         description="状态元数据，包含会话标识、流程进度、工具调用记录、时间戳等关键信息，值类型支持字符串、数字、时间、布尔值"
22.     )

复制代码

二、状态更新机制

        节点通过返回值更新状态，LangGraph 使用 Reducer（规约器） 控制更新逻辑。

默认 Reducer（覆盖更新）：未指定 Reducer 时，节点返回的字段直接覆盖原状态对应字段。

1. from typing_extensions import TypedDict
3. class State(TypedDict):
4.     foo: int
5.     bar: list[str]
7. # 节点函数：仅更新foo字段
8. def node_1(state: State) -> dict:
9.     return {"foo": 2}  # 无需返回bar，原bar值会保留
11. # 初始状态：{"foo": 1, "bar": ["hi"]}
12. # 节点执行后状态：{"foo": 2, "bar": ["hi"]}（foo被覆盖，bar不变）

复制代码

内置 Reducer：通过 Annotated 指定内置 Reducer，实现复杂更新（如列表追加）。

1. from typing import Annotated
2. from langgraph.graph import add
4. class State(TypedDict):
5.     logs: Annotated[List[str], add]  # 自动追加新元素
7. def add_log(state: State) -> dict:
8.     return {"logs": ["step 1 completed"]}  # 追加到现有logs
10. # 初始：{"logs": ["start"]} → 执行后：{"logs": ["start", "step 1 completed"]}

复制代码

add_messages：消息列表专用（处理消息去重 / 更新）

1. from typing import Annotated
2. from langgraph.graph.message import add_messages
3. from langchain_core.messages import AnyMessage
5. class State(TypedDict):
6.     messages: Annotated[List[AnyMessage], add_messages]  # 智能合并消息
8. def add_ai_message(state: State) -> dict:
9.     from langchain_core.messages import AIMessage
10.     return {"messages": [AIMessage(content="Hello!")]}  # 自动追加新消息

复制代码

自定义 Reducer：根据业务需求定义更新逻辑（如重要消息置顶）。

1. from typing import Annotated
3. # 自定义Reducer：带"important"标签的消息置顶
4. def priority_adder(current: List[dict], new: List[dict]) -> List[dict]:
5.     for item in new:
6.         if "important" in item.get("tags", []):
7.             current.insert(0, item)  # 置顶
8.         else:
9.             current.append(item)  # 普通追加
10.     return current
12. class State(TypedDict):
13.     notifications: Annotated[List[dict], priority_adder]  # 使用自定义Reducer

复制代码

MessagesState（对话场景专用）：预定义状态类，内置 messages 字段和 add_messages Reducer，可直接扩展。

1. from langgraph.graph import MessagesState
3. class ChatState(MessagesState):
4.     user_id: str  # 扩展字段：用户ID
5.     topic: str = ""  # 扩展字段：对话主题
7. # 节点：添加消息并更新主题
8. def update_chat(state: ChatState) -> dict:
9.     from langchain_core.messages import HumanMessage
10.     return {
11.         "messages": [HumanMessage(content="讨论AI应用")],
12.         "topic": "AI应用"
13.     }

复制代码

创建步骤序列-add_sequence

        Sequence 是 LangGraph 中最基础的工作流模式，指节点按固定顺序依次执行的线性流程。每个节点完成后，状态会传递给下一个节点，直到所有节点执行完毕。
        适用于步骤固定的场景，如：数据预处理→模型推理→结果格式化、多轮工具调用的依次执行等。
        LangGraph 有两种方式定义节点顺序：

方式 1：手动添加节点和边（基础方式）

1. from langgraph.graph import START, StateGraph
3. # 1. 初始化图（绑定状态类型）
4. builder = StateGraph(State)
6. # 2. 添加节点（节点名默认为函数名，如"step_1"）
7. builder.add_node(step_1)  # 等价于 builder.add_node("step_1", step_1)
8. builder.add_node(step_2)
9. builder.add_node(step_3)
11. # 3. 定义执行顺序：START → step_1 → step_2 → step_3
12. builder.add_edge(START, "step_1")       # 起始点连接第一个节点
13. builder.add_edge("step_1", "step_2")    # 节点1执行完→节点2
14. builder.add_edge("step_2", "step_3")    # 节点2执行完→节点3
16. # 4. 编译图（生成可执行对象）
17. graph = builder.compile()

复制代码

方式2：使用add_sequence简化（推荐）

1. # 一行代码添加序列节点，自动按列表顺序连接
2. builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
3. builder.add_edge(START, "step_1")  # 只需指定起始点
5. graph = builder.compile()

复制代码

分支机制（Branches）

        LangGraph 支持通过分支机制实现节点的并行执行、条件路由等复杂工作流，核心包括并行执行、延迟执行和条件分支三种模式。
一、并行执行节点（Parallel Execution）

        通过 “扇出（fan-out）→ 扇入（fan-in）” 机制让多个节点同时执行，提升工作流效率，适用于可并行处理的任务（如多源数据获取、并行计算等）。

实现核心：

扇出：从一个节点同时连接到多个节点（如 A → B 和 A → C）；
扇入：多个节点的输出汇聚到同一个后续节点（如 B → D 和 C → D）；
聚合规则：通过 reducer 定义并行节点的状态合并方式（如列表拼接）。

关键代码示例：

1. # 示例：A → B/C (并行) → D
2. builder = StateGraph(State)
3. builder.add_edge("a", "b")  # 并行分支1
4. builder.add_edge("a", "c")  # 并行分支2
5. builder.add_edge("b", "d")  # 汇聚点
6. builder.add_edge("c", "d")  # 汇聚点

复制代码

二、延迟执行节点（Defer Execution）

        当分支长度不同时（如一条分支多一个节点），让后续节点等待所有前置分支完成后再执行，避免因分支进度不一致导致的状态不完整。

实现核心：对需要等待的节点设置 defer=True，使其延迟执行，直到所有指向它的前置节点（包括长分支）都完成。
关键代码示例：

1. # 在并行执行示例基础上，给B添加一个子节点B_2
2. def b_2(state: State): return {"aggregate": ["B_2"]}
4. builder = StateGraph(State)
5. builder.add_node(a)
6. builder.add_node(b)
7. builder.add_node(b_2)  # B的后续节点
8. builder.add_node(c)
9. builder.add_node(d, defer=True)  # D延迟执行，等待所有前置完成
11. # 分支结构：A→B→B_2→D，A→C→D
12. builder.add_edge(START, "a")
13. builder.add_edge("a", "b")
14. builder.add_edge("a", "c")
15. builder.add_edge("b", "b_2")
16. builder.add_edge("b_2", "d")
17. builder.add_edge("c", "d")
18. builder.add_edge("d", END)
20. graph = builder.compile()
21. graph.invoke({"aggregate": []})
22. # 输出：['A', 'B', 'C', 'B_2', 'D']（D等待B_2完成后执行）

复制代码

三、条件分支（Conditional Branching）

        根据运行时的状态动态选择执行路径，适用于需要根据输入或中间结果决策的场景（如根据用户问题类型路由到不同处理节点）。

实现核心：使用 add_conditional_edges 定义条件路由函数，根据状态返回下一步节点名称（可返回单个或多个节点）。
动态路由：

1. class State(TypedDict):
2.     which: str  # 决策字段
4. def conditional_edge(state: State) -> Literal["b", "c"]:
5.     return state["which"]  # 动态路由
7. builder.add_conditional_edges("a", conditional_edge)

复制代码

单/多分支路由：

1. def route_nodes(state: State) -> Sequence[str]:
2.     if state["type"] == "complex":
3.         return ["c", "d"]  # 多分支
4.     return ["b"]  # 单分支

复制代码

MapReduce 模式

        在 LangGraph 中，Map-Reduce 是一种通过并行处理子任务（Map 阶段） 再聚合结果（Reduce 阶段） 的工作流模式，适用于需要批量处理多个相似任务并汇总结果的场景（如多文档摘要、多源数据整合等）。其核心通过 Send API 实现动态扇出（Fan-out）并行任务，再通过汇聚节点完成结果合并。

代码示例

1. import operator
2. from typing import Annotated
3. from langchain_openai import ChatOpenAI
4. from typing_extensions import TypedDict
5. from langgraph.types import Send
6. from langgraph.graph import END, StateGraph, START
7. from pydantic import BaseModel, Field
9. """
10. 生成与{topic}相关的1到3个示例的逗号分隔列表。
11. 生成一个关于{subject}的笑话
12. 下面是一些关于{topic}的笑话。选择最好的一个！返回最佳的ID。
13. {jokes}
14. """
16. subjects_prompt = """Generate a comma separated list of between 1 and 3 examples related to: {topic}."""
17. joke_prompt = """Generate a joke about {subject}"""
18. best_joke_prompt = """Below are a bunch of jokes about {topic}. Select the best one! Return the ID of the best one.
19. {jokes}"""
22. class Subjects(BaseModel):
23.     subjects: list[str]
26. class Joke(BaseModel):
27.     joke: str
30. class BestJoke(BaseModel):
31.     id: int = Field(description="Index of the best joke, starting with 0")
34. from dotenv import load_dotenv
36. load_dotenv()
37. model = ChatOpenAI(model="gpt-4o-mini")
40. class OverallState(TypedDict):
41.     topic: str
42.     subjects: list
43.     jokes: Annotated[list, operator.add]
44.     best_selected_joke: str
47. class JokeState(TypedDict):
48.     subject: str
51. def generate_topics(state: OverallState):
52.     prompt = subjects_prompt.format(topic=state["topic"])
53.     response = model.with_structured_output(Subjects).invoke(prompt)
54.     return {"subjects": response.subjects}
57. def generate_joke(state: JokeState):
58.     prompt = joke_prompt.format(subject=state["subject"])
59.     response = model.with_structured_output(Joke).invoke(prompt)
60.     return {"jokes": [response.joke]}
63. def continue_to_jokes(state: OverallState):
64.     return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]
66. """for循环与列表推导式 以时间换空间的方式   python没有真正意义上的多线程"""
68. def best_joke(state: OverallState):
69.     jokes = "\n\n".join(state["jokes"])
70.     prompt = best_joke_prompt.format(topic=state["topic"], jokes=jokes)
71.     response = model.with_structured_output(BestJoke).invoke(prompt)
72.     return {"best_selected_joke": state["jokes"][response.id]}
75. graph = StateGraph(OverallState)
76. graph.add_node("generate_topics", generate_topics)
77. graph.add_node("generate_joke", generate_joke)
78. graph.add_node("best_joke", best_joke)
80. graph.add_edge(START, "generate_topics")
81. graph.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])
82. graph.add_edge("generate_joke", "best_joke")
83. graph.add_edge("best_joke", END)
84. app = graph.compile()
86. from IPython.display import Image
88. Image(app.get_graph().draw_mermaid_png(output_file_path='./img/示例4.png'))
90. # Call the graph: here we call it to generate a list of jokes
91. for s in app.stream({"topic": "animals"}):
92.     print(s)

复制代码

两个关键阶段：

1. def generate_joke(state: JokeState):
2.     prompt = joke_prompt.format(subject=state["subject"])
3.     response = model.with_structured_output(Joke).invoke(prompt)
4.     return {"jokes": [response.joke]}
7. def continue_to_jokes(state: OverallState):
8.     return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

复制代码

Map 阶段：将一个总任务拆分为多个子任务，并行执行（如为每个子项生成结果）；

generate_joke：将主任务分解为多个子任务
continue_to_jokes：使用 Send API 动态创建并行任务

1. class OverallState(TypedDict):
2.     topic: str
3.     subjects: list
4.     jokes: Annotated[list, operator.add]
5.     best_selected_joke: str
8. def best_joke(state: OverallState):
9.     jokes = "\n\n".join(state["jokes"])
10.     prompt = best_joke_prompt.format(topic=state["topic"], jokes=jokes)
11.     response = model.with_structured_output(BestJoke).invoke(prompt)
12.     return {"best_selected_joke": state["jokes"][response.id]}

复制代码

Reduce 阶段：收集所有子任务的结果，进行汇总、筛选或加工（如从多个结果中选最优）。

jokes 字段：通过 operator.add 规约器实现并行结果的自动拼接，确保多个子任务的输出被合并。
best_joke:从聚合的jokes中筛选最优结果。

循环控制机制

循环结构的三要素

1. builder.add_edge(START, "a")           # 入口节点
2. builder.add_conditional_edges("a", route)  # 条件分支
3. builder.add_edge("b", "a")             # 循环回边

复制代码

入口节点：工作流起点
条件分支：决定继续循环或退出
循环回边：实现节点间循环

终止条件实现

1. def route(state: State) -> Literal["b", END]:
2.     if len(state["aggregate"]) < 7:  # 终止条件判断
3.         return "b"   # 继续循环
4.     else:
5.         return END   # 退出循环

复制代码

递归深度控制

        若终止条件可能无法满足（如逻辑漏洞），需设置递归限制（recursion limit），规定最大执行步数（超级步骤），超过则抛出异常。

1. from langgraph.errors import GraphRecursionError
3. try:
4.     # 设置递归限制为4（最多执行4个超级步骤）
5.     graph.invoke({"aggregate": []}, {"recursion_limit": 4})
6. except GraphRecursionError:
7.     print("Recursion Error: 超过最大执行步数")

复制代码

可视化图

        LangGraph 提供了多种方式可视化工作流图，帮助开发者调试、展示节点关系和执行流程。核心支持 Mermaid 语法生成 和 PNG 图像渲染，适用于不同场景（如文档嵌入、调试分析）。
Mermaid 语法生成

        Mermaid 是一种文本驱动的图表描述语言，可生成流程图、时序图等。LangGraph 可直接将图转换为 Mermaid 语法，便于嵌入文档或通过 Mermaid 工具渲染。

代码示例：

1. # 生成 Mermaid 流程图语法
2. mermaid_code = graph.get_graph().draw_mermaid()
3. print(mermaid_code)

复制代码

生成示例：

1. %%{init: {'flowchart': {'curve': 'linear'}}}%%
2. graph TD;
3.     __start__([<p>__start__</p>]):::first
4.     entry_node(entry_node)
5.     node_entry_node_A(node_entry_node_A)
6.     __end__([<p>__end__</p>]):::last
7.     __start__ --> entry_node;
8.     entry_node --> __end__;
9.     entry_node --> node_entry_node_A;
10.     node_entry_node_A -.-> entry_node;  # 虚线表示条件边
11.     classDef default fill:#f2f0ff,line-height:1.2

复制代码

PNG 图像渲染

        若需要图片格式（如用于报告、演示），LangGraph 支持通过多种工具生成 PNG 图像，以下是三种常用方式：
1. 基于 Mermaid.ink API

        Mermaid.ink 是一个免费的在线 Mermaid 渲染服务，LangGraph 可直接调用该 API 生成 PNG，无需安装本地工具。

代码示例：

1. from IPython.display import Image, display
3. # 生成 PNG 并显示（默认使用 Mermaid.ink）
4. png_data = graph.get_graph().draw_mermaid_png()
5. display(Image(png_data))

复制代码

生成示例：

2. 基于 Mermaid + Pyppeteer

        通过 Pyppeteer（Headless Chrome 工具）在本地渲染 Mermaid 语法为 PNG，支持更多自定义选项（如节点颜色、背景色）。

安装依赖

1. pip install pyppeteer nest-asyncio  # nest-asyncio 用于 Jupyter 环境

复制代码

代码示例：

1. import nest_asyncio
2. from langchain_core.runnables.graph import CurveStyle, NodeStyles, MermaidDrawMethod
3. from IPython.display import Image, display
5. nest_asyncio.apply()  # Jupyter 环境需启用异步支持
7. # 自定义样式：线条类型、节点颜色、背景等
8. png_data = graph.get_graph().draw_mermaid_png(
9.     curve_style=CurveStyle.LINEAR,  # 线条样式：线性
10.     node_colors=NodeStyles(
11.         first="#ffdfba",  # 起始节点颜色
12.         last="#baffc9",   # 终止节点颜色
13.         default="#fad7de" # 默认节点颜色
14.     ),
15.     background_color="white",  # 背景色
16.     padding=10,  # 边距
17.     draw_method=MermaidDrawMethod.PYPPETEER  # 使用 Pyppeteer 本地渲染
18. )
19. display(Image(png_data))

复制代码

生成示例：

3.基于 Graphviz

        Graphviz 是一款强大的开源绘图工具，LangGraph 可通过 pygraphviz 库调用其生成 PNG，适合复杂图的精细化展示。

安装依赖

安装 Graphviz 软件（官网）；
安装 Python 依赖：

1. pip install pygraphviz

复制代码

代码示例：

1. from IPython.display import Image, display
3. try:
4.     png_data = graph.get_graph().draw_png()  # 直接生成 PNG
5.     display(Image(png_data))
6. except ImportError:
7.     print("请安装 pygraphviz 和 Graphviz 工具")

复制代码

适用场景总结

可视化方式	优点	缺点	适用场景
Mermaid 语法	轻量、跨平台、易存储	需工具渲染才能可视化	文档嵌入、版本控制、快速分享
Mermaid.ink API	零依赖、快速生成	依赖网络、样式有限	临时调试、简单展示
Mermaid + Pyppeteer	本地渲染、样式丰富	需安装依赖	自定义需求高的本地展示
Graphviz	复杂图布局优、格式多样	安装复杂	专业报告、复杂工作流展示

参考文献

Use the Graph API

原文地址：https://blog.csdn.net/weixin_41645817/article/details/149915750