LangGraph StateGraph 实战：状态机聊天机器人构建指南

背景

LangGraph 提供了两种构建 Agent 的方式：

1. 高层 API：create_react_agent 快速搭建
2. 底层 API：StateGraph 自定义状态机

本文深入解析 StateGraph 的实现原理，通过构建一个聊天机器人，掌握 LangGraph 的核心概念。

实现原理

StateGraph 核心概念

┌─────────────────────────────────────────────────────────┐
│                    StateGraph                           │
├─────────────────────────────────────────────────────────┤
│  State (状态)  │  应用在某个时刻的数据快照               │
│  Node (节点)   │  执行特定任务的函数                     │
│  Edge (边)     │  定义节点之间的流转关系                 │
└─────────────────────────────────────────────────────────┘

对比：create_react_agent vs StateGraph

代码实现

完整代码

"""
LangGraph 教程 - 构建一个基本聊天机器人

本示例演示如何使用 StateGraph 构建一个基础聊天机器人。
官方教程地址：
"""

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: Annotated[list, add_messages]


# ==================== 2. 创建图构建器 ====================
graph_builder = StateGraph(State)


# ==================== 3. 初始化模型 ====================
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):
    """聊天机器人节点函数。"""
    return {"messages": [llm.invoke(state["messages"])]}


graph_builder.add_node("chatbot", chatbot)


# ==================== 5. 定义边 ====================
graph_builder.add_edge(START, "chatbot")


# ==================== 6. 编译图 ====================
graph = graph_builder.compile()


# ==================== 7. 运行 ====================
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)


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("nn 再见！")
            break
        except Exception as e:
            print(f"发生错误: {e}")
            break


if __name__ == "__main__":
    main()

核心代码解析

状态定义

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 函数，控制状态更新行为

节点函数规范

def chatbot(state: State) -> dict:
    """
    节点函数签名规范：
    - 输入：当前 state
    - 输出：状态更新字典 {状态键: 更新值}
    """
    return {"messages": [llm.invoke(state["messages"])]}

流式执行机制

for event in graph.stream(initial_state):
    for node_name, value in event.items():
        # event = {"节点名": {状态更新}}
        process(value)

stream() vs invoke()：

• stream()：迭代器，实时返回每个节点结果
• invoke()：阻塞式，直接返回最终结果

运行效果

执行

pip install langgraph langchain langchain-openai pydantic python-dotenv typing-extensions
python 02构建一个基本聊天机器人.py

输出

 LangGraph 聊天机器人已启动！
==================================================
提示：输入 'quit'、'exit' 或 'q' 退出对话

用户: 什么是 StateGraph？
助手: StateGraph 是 LangGraph 的核心抽象，用于构建状态机...

用户: quit

 再见！

架构图解

状态流转

┌─────────┐     ┌─────────────┐     ┌─────────┐
│  START  │────▶│  chatbot    │────▶│   END   │
└─────────┘     └─────────────┘     └─────────┘
                      │
                      ▼
              ┌─────────────┐
              │   State     │
              │  messages   │
              └─────────────┘

执行时序

User Input
    │
    ▼
┌─────────────────┐
│  Initial State  │
└─────────────────┘
    │
    ▼
┌─────────────────┐
│  graph.stream() │
└─────────────────┘
    │
    ├──▶ Node: chatbot
    │       │
    │       ▼
    │   LLM.invoke()
    │       │
    │       ▼
    │   Update State
    │       │
    ◀───────┘
    │
    ▼
Extract Response
    │
    ▼
Print Output

总结

本文通过实战代码展示了 StateGraph 的核心机制：

1. 状态定义：使用 TypedDict + Annotated 定义状态结构
2. 节点编写：输入 state，返回状态更新字典
3. 图构建：add_node + add_edge + compile
4. 流式执行：graph.stream() 实时获取节点输出

进阶方向

• 条件边：add_conditional_edges 实现分支逻辑
• 持久化：集成检查点器实现状态持久化
• 多节点：构建复杂的多步骤工作流
• 人机协作：Human-in-the-loop 节点

参考

• LangGraph 官方教程
• StateGraph API 文档