LangGraph：构建有状态大模型工作流的利器

作者：AI小模型
一. LangGraph简介

LangGraph 是基于 LangChain 的扩展框架，专为构建有状态（Stateful）的大模型工作流而设计。它通过图结构（Graph）定义多个执行节点（Node）及其依赖关系，支持复杂任务编排，尤其适合多智能体协作、长对话管理等场景。
1.1 核心优势

状态持久化：自动维护任务执行过程中的上下文状态
灵活编排：支持条件分支、循环、并行等控制流
容错机制：内置错误重试、回滚策略
可视化调试：自动生成执行流程图

二. LangGraph最佳实践

2.1 基础代码结构

Python

1. from langgraph.graph import StateGraph, END  
2. from typing import TypedDict, Annotated  
3. import operator  
4. # 定义状态结构  
5. class AgentState(TypedDict):  
6.     input: str  
7.     result: Annotated[list, operator.add]  # 自动累积结果  
8. # 初始化图  
9. graph = StateGraph(AgentState)  
10. # 添加节点与边（后续章节详解）  
11. ...  
12. # 编译并运行  
13. app = graph.compile()  
14. result = app.invoke({"input": "任务描述"})

复制代码

2.2 开发原则

模块化设计：每个节点只完成单一职责

状态最小化：仅保留必要数据，避免内存膨胀

幂等性保证：节点可安全重试

三. 状态设计（State Design）

3.1 状态定义规范

使用 Pydantic模型 或 TypedDict 明确状态结构：
Python

1. from pydantic import BaseModel  
2. class ProjectState(BaseModel):  
3.     requirements: str  
4.     draft_versions: list[str]  
5.     current_step: int  
6. # 初始化状态  
7. initial_state = ProjectState(  
8.     requirements="开发一个聊天机器人",  
9.     draft_versions=[],  
10.     current_step=0  
11. )

复制代码

3.2 状态自动管理

LangGraph通过注解（Annotation）实现状态字段的自动更新：
Python

1. from langgraph.graph import add_messages  
2. class DialogState(TypedDict):  
3.     history: Annotated[list, add_messages]  # 自动追加消息  
4. def user_node(state: DialogState):  
5.     return {"history": ["用户: 你好"]}  
6. def bot_node(state: DialogState):  
7.     return {"history": ["AI: 您好，有什么可以帮您？"]}

复制代码

四. 节点函数（Node Functions）

4.1 节点定义标准

节点是工作流的基本单元，接收状态并返回更新：
Python

1. from langchain_core.runnables import RunnableLambda  
2. # 简单节点  
3. def data_loader(state: dict):  
4.     return {"data": load_dataset(state["input"])}  
5. # 包含LLM调用的节点  
6. llm_node = RunnableLambda(  
7.     lambda state: {"answer": chat_model.invoke(state["question"])}  
8. )  
9. # 注册节点  
10. graph.add_node("loader", data_loader)  
11. graph.add_node("llm", llm_node)

复制代码

4.2 多智能体协作

Python

1. def designer_agent(state):  
2.     return {"design": "界面草图"}  
3. def developer_agent(state):  
4.     return {"code": "实现代码"}  
5. # 并行执行  
6. graph.add_node("designer", designer_agent)  
7. graph.add_node("developer", developer_agent)  
8. graph.add_edge("designer", "reviewer")  
9. graph.add_edge("developer", "reviewer")

复制代码

五. 边的设计（Edge Design）

5.1 条件分支（Conditional Edges）

根据状态值动态路由：
Python

1. from langgraph.graph import conditional_edge  
2. def should_continue(state):  
3.     return "continue" if state["step"] < 5 else "end"  
4. graph.add_conditional_edges(  
5.     source="decision_node",  
6.     path_map={"continue": "next_node", "end": END},  
7.     condition=should_continue  
8. )

复制代码

5.2 循环结构

Python

1. graph.add_edge("start", "process")  
2. graph.add_conditional_edges(  
3.     "process",  
4.     lambda s: "loop" if s["count"] < 3 else "end",  
5.     {"loop": "process", "end": END}  
6. )

复制代码

六. 错误处理（Error Handling）

6.1 重试机制

Python

1. from langgraph.retry import RetryPolicy  
2. policy = RetryPolicy(  
3.     max_retries=3,  
4.     backoff_factor=1.5,  
5.     retry_on=(Exception,)  
6. )  
7. graph.add_node(  
8.     "api_call",   
9.     api_wrapper.with_retry(policy)  
10. )

复制代码

6.2 回滚策略

Python

1. def compensation_action(state):  
2.     # 执行补偿操作  
3.     rollback_transaction(state["tx_id"])  
4.     return {"status": "rolled_back"}  
5. graph.add_edge("failed_node", "compensation")  
6. graph.add_edge("compensation", END)

复制代码

学习书籍文档

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

学习视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

项目实战源码

学以致用 ，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

大模型面试题

面试不仅是技术的较量，更需要充分的准备。
在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

注：本文代码基于LangGraph 0.1+版本实现，需预先安装依赖：
Bash

1. pip install langgraph langchain pydantic

复制代码

原文地址：https://blog.csdn.net/machinegunjoe/article/details/147834311