从零理解LangGraph状态图：构建你的第一个AI聊天机器人
## 前言
在上一篇文章中，我们使用 `create_react_agent` 快速搭建了第一个 LangGraph Agent。但在实际开发中，我们常常需要更灵活的控制流程。本文将带你深入理解 LangGraph 的核心概念——**StateGraph（状态图）**，通过从零构建一个聊天机器人，掌握状态管理、节点编排和流式输出等关键技术。
---
## 一、核心概念
### 1.1 什么是 StateGraph
StateGraph 是 LangGraph 的核心抽象，它将应用建模为**状态机**：
- **状态（State）**：应用在某个时刻的数据快照
- **节点（Node）**：执行特定任务的函数
- **边（Edge）**：定义节点之间的流转关系
类比理解：
- StateGraph 就像一张**流程图**
- 节点是流程图中的**处理步骤**
- 边是连接步骤的**箭头**
### 1.2 为什么需要 StateGraph
| 场景 | `create_react_agent` | `StateGraph` |
|------|---------------------|--------------|
| 快速原型 | ✅ 适合 | ⚠️ 需要更多代码 |
| 自定义流程 | ❌ 固定 ReAct 模式 | ✅ 完全可控 |
| 多节点协作 | ❌ 单节点 | ✅ 支持多节点 |
| 条件分支 | ❌ 不支持 | ✅ 支持条件边 |
StateGraph 提供了**细粒度的控制能力**，适合构建复杂的 AI 工作流。
---
## 二、环境准备
### 2.1 安装依赖
```bash
pip install langgraph langchain langchain-openai pydantic python-dotenv typing-extensions
```
### 2.2 配置 API 密钥
创建 `.env` 文件：
```env
SILICONFLOW_API_KEY=your_api_key_here
```
---
## 三、代码实现
### 3.1 完整代码
```python
"""
LangGraph 教程 - 构建一个基本聊天机器人
本示例演示如何使用 StateGraph 构建一个基础聊天机器人。
这是 LangGraph 系列教程的基础，后续将逐步添加更复杂的功能。
官方教程地址：https://langchain-ai.github.io/langgraph/tutorials/introduction/
"""
# 过滤警告信息
import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore", category=UserWarning)
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
# ==================== 1. 定义状态 ====================
class State(TypedDict):
"""
定义图的状态结构。
messages: 消息列表，使用 add_messages reducer 函数
确保新消息追加到列表，而不是覆盖
"""
messages: Annotated[list, add_messages]
# ==================== 2. 创建图构建器 ====================
graph_builder = StateGraph(State)
# ==================== 3. 初始化模型 ====================
# 使用硅基流动平台的 Qwen 模型
llm = ChatOpenAI(
model="Qwen/Qwen3-Next-80B-A3B-Instruct",
openai_api_key=os.getenv("SILICONFLOW_API_KEY"),
openai_api_base="https://api.siliconflow.cn/v1",
temperature=0.7
)
# ==================== 4. 定义聊天机器人节点 ====================
def chatbot(state: State):
"""
聊天机器人节点函数。
Args:
state: 当前状态，包含消息历史
Returns:
包含 LLM 响应消息的字典
"""
return {"messages": [llm.invoke(state["messages"])]}
# 添加节点到图
# 第一个参数是节点名称，第二个参数是节点函数
graph_builder.add_node("chatbot", chatbot)
# ==================== 5. 添加入口点 ====================
# 告诉图每次运行时从哪里开始
graph_builder.add_edge(START, "chatbot")
# ==================== 6. 编译图 ====================
# 编译图，创建可执行的 CompiledGraph
graph = graph_builder.compile()
# ==================== 7. 可视化图（可选）====================
def visualize_graph():
"""可视化图结构（需要安装额外依赖）。"""
try:
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception as e:
print(f"可视化失败（可选功能）: {e}")
print("提示：可以通过安装依赖来使用可视化功能")
# ==================== 8. 运行聊天机器人 ====================
def stream_graph_updates(user_input: str):
"""
流式处理图更新。
Args:
user_input: 用户输入的消息
"""
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
for value in event.values():
print("助手:", value["messages"][-1].content)
def main():
"""主函数 - 运行交互式聊天机器人。"""
print("🤖 LangGraph 聊天机器人已启动！")
print("=" * 50)
print("提示：输入 'quit'、'exit' 或 'q' 退出对话\n")
while True:
try:
# 获取用户输入
user_input = input("用户: ")
# 检查退出命令
if user_input.lower() in ["quit", "exit", "q"]:
print("\n👋 再见！")
break
# 处理用户输入并获取响应
stream_graph_updates(user_input)
print() # 空行分隔对话
except KeyboardInterrupt:
print("\n\n👋 再见！")
break
except Exception as e:
print(f"发生错误: {e}")
# 备用方案：使用默认输入
user_input = "What do you know about LangGraph?"
print(f"\n用户: {user_input}")
stream_graph_updates(user_input)
break
if __name__ == "__main__":
main()
```
### 3.2 代码解析
#### 3.2.1 定义状态
```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
```
**关键点**：
- `TypedDict`：定义状态的结构类型
- `Annotated`：为类型添加元数据
- `add_messages`：**reducer 函数**，确保消息追加而非覆盖
#### 3.2.2 创建图构建器
```python
from langgraph.graph import StateGraph, START
graph_builder = StateGraph(State)
```
`StateGraph` 是图的构建器，接收状态类型作为参数。
#### 3.2.3 定义节点
```python
def chatbot(state: State):
"""聊天机器人节点函数。"""
return {"messages": [llm.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
```
节点函数的特点：
- **输入**：当前 `state`
- **输出**：状态更新字典
- **命名**：通过 `add_node` 注册到图中
#### 3.2.4 定义边
```python
graph_builder.add_edge(START, "chatbot")
```
- `START`：图的入口点
- `"chatbot"`：目标节点名称
#### 3.2.5 编译图
```python
graph = graph_builder.compile()
```
编译后的 `graph` 可以被执行，支持 `stream()` 和 `invoke()` 方法。
#### 3.2.6 流式执行
```python
def stream_graph_updates(user_input: str):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
for value in event.values():
print("助手:", value["messages"][-1].content)
```
`graph.stream()` 的特点：
- **流式输出**：实时返回每个节点的执行结果
- **事件结构**：`{节点名: {状态更新}}`
- **适用场景**：长任务、需要实时反馈的场景
---
## 四、运行效果
### 4.1 执行步骤
```bash
python 02构建一个基本聊天机器人.py
```
### 4.2 输出结果
---
## 五、核心概念图解
### 5.1 状态流转图
```
┌─────────┐ ┌─────────────┐ ┌─────────┐
│ START │────▶│ chatbot │────▶│ END │
│ (入口) │ │ (LLM 节点) │ │ (结束) │
└─────────┘ └─────────────┘ └─────────┘
│
▼
┌─────────────┐
│ State │
│ messages │
└─────────────┘
```
### 5.2 执行流程
```
用户输入 ──▶ 构建初始 State ──▶ graph.stream() ──▶ 节点执行
│
▼
用户 ◀── 提取响应 ◀── 更新 State ◀── LLM 调用 ◀── chatbot 节点
```
---
## 六、踩坑记录
| 问题 | 原因 | 解决方案 |
|------|------|----------|
| `TypedDict` 导入错误 | Python 版本问题 | 使用 `typing_extensions` 替代 |
| 消息被覆盖而非追加 | 缺少 reducer | 使用 `Annotated[list, add_messages]` |
| 节点函数返回值格式错误 | 不了解规范 | 必须返回 `{状态键: 更新值}` 字典 |
| 图无法执行 | 忘记编译 | 调用 `graph_builder.compile()` |
---
## 七、总结
通过本文，你学会了：
1. ✅ **StateGraph 核心概念**：状态、节点、边
2. ✅ **状态定义**：使用 `TypedDict` 和 `Annotated`
3. ✅ **节点编写**：输入状态、返回更新
4. ✅ **图构建**：添加节点、定义边、编译执行
5. ✅ **流式输出**：使用 `graph.stream()` 实时获取结果
### 进阶方向
- **条件边**：使用 `add_conditional_edges` 实现分支逻辑
- **多节点协作**：构建复杂的多步骤工作流
- **持久化存储**：集成检查点器实现对话记忆
- **人机协作**：添加 Human-in-the-loop 节点
---
## 参考资料
- [LangGraph 官方教程](https://langchain-ai.github.io/langgraph/tutorials/introduction/)
- [StateGraph API 文档](https://langchain-ai.github.io/langgraph/reference/graphs/)
- [Python TypedDict 文档](https://docs.python.org/3/library/typing.html#typing.TypedDict)
---
> 📌 本文首发于 CSDN，作者：码上AI_123
> 🔗 转载请注明出处
>
> 💡 如果觉得有帮助，欢迎点赞、收藏、关注！
> 🎯 你的支持是我持续创作的动力！
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
20
0- 0
已为社区贡献3条内容
所有评论(0)