2.5 实战：LangGraph 状态系统

🎯 小白理解指南：这一节讲什么？

前面我们学了各种"积木"：

• 列表（存消息历史）
• 字典（存配置信息）
• 集合（去重）
• TypedDict（类型安全的字典）
• Pydantic（数据验证）

这一节我们要把这些积木组装起来，构建一个真正的 LangGraph Agent 状态系统！

LangGraph 工作原理简述：

1. 定义一个状态（State）——告诉系统要记住哪些信息
2. 创建多个节点（Node）——每个节点处理状态、返回更新
3. 用**边（Edge）**连接节点——决定执行顺序
4. 运行图——状态在节点间传递，最终得到结果

完整的 LangGraph Agent 状态实现

"""
生产级 LangGraph 状态管理系统
整合 TypedDict + Pydantic + 数据结构
"""

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
import operator

# 1. 定义状态结构
# 🎯 小白解读：
# AgentState 就是 Agent 的"记忆表"，规定了它要记住哪些信息
# TypedDict 确保每个字段的类型都是固定的，防止出错

class AgentState(TypedDict):
    """Agent 状态"""
    # 消息列表 - 使用 Annotated 定义合并策略
    # 🎯 operator.add 的意思是：新消息会"追加"到旧消息后面，而不是替换
    messages: Annotated[Sequence[BaseMessage], operator.add]

    # 当前步骤（比如 "init" → "processing" → "responding"）
    current_step: str

    # 迭代计数（记录循环了多少次，防止无限循环）
    iteration: int

    # 工具使用历史（记录用过哪些工具）
    tools_used: list[str]

    # 上下文信息（可以存任何额外信息）
    context: dict

# 2. 节点函数
# 🎯 小白解读：
# 节点就像流水线上的"工作站"，每个节点负责一项工作
# 节点函数接收当前状态，处理后返回要更新的部分

def agent_node(state: AgentState) -> AgentState:
    """Agent 节点：处理用户输入"""
    # 从状态中读取信息
    messages = state["messages"]
    iteration = state["iteration"]

    # 模拟 LLM 调用（实际项目中这里会调用 ChatGPT 等）
    last_message = messages[-1]
    response = AIMessage(content=f"处理: {last_message.content}")

    # 返回要更新的状态
    # 🎯 注意：只返回要更新的字段，其他字段会保持不变
    return {
        "messages": [response],      # 新消息会追加（因为用了 operator.add）
        "iteration": iteration + 1,   # 迭代次数 +1
        "current_step": "processing"  # 更新当前步骤
    }

def tool_node(state: AgentState) -> AgentState:
    """工具节点：调用外部工具"""
    # 🎯 工具节点执行具体任务（搜索、计算等）
    return {
        "current_step": "responding",
        "tools_used": state["tools_used"] + ["search"]  # 记录使用了搜索工具
    }

# 3. 路由逻辑
# 🎯 小白解读：
# 路由就像"交通指挥员"，根据当前状态决定下一步去哪个节点
# 返回的字符串对应后面 add_conditional_edges 中定义的路径

def should_continue(state: AgentState) -> str:
    """决定下一步"""
    # 如果迭代次数超过 3 次，强制结束（防止死循环）
    if state["iteration"] >= 3:
        return "end"

    # 如果消息中包含"搜索"，就去调用工具
    last_message = state["messages"][-1]
    if "搜索" in last_message.content:
        return "tools"

    # 否则直接结束
    return "end"

# 4. 构建图
# 🎯 小白解读：
# 这一步是把前面定义的节点和路由"组装"成一个完整的流程图

def create_agent_graph():
    """创建 Agent 图"""
    # 创建一个空的流程图，告诉它用什么状态结构
    workflow = StateGraph(AgentState)

    # 添加节点（给每个节点起个名字）
    workflow.add_node("agent", agent_node)  # 主处理节点
    workflow.add_node("tools", tool_node)   # 工具调用节点

    # 设置入口点（从哪个节点开始）
    workflow.set_entry_point("agent")

    # 添加条件边（根据 should_continue 的返回值决定走哪条路）
    # 🎯 这就像地铁线路图：从 agent 站出发，根据条件去不同站
    workflow.add_conditional_edges(
        "agent",           # 从 agent 节点出发
        should_continue,   # 用这个函数判断走哪条路
        {
            "tools": "tools",  # 返回 "tools" 就去 tools 节点
            "end": END         # 返回 "end" 就结束
        }
    )

    # 工具节点执行完后，回到 agent 节点（形成循环）
    workflow.add_edge("tools", "agent")

    # 编译图（让它可以运行）
    # 🎯 memory 是"记忆存储"，用于保存对话历史（这里用内存临时存储）
    memory = SqliteSaver.from_conn_string(":memory:")
    app = workflow.compile(checkpointer=memory)

    return app

# 5. 使用
# 🎯 小白解读：
# 这部分展示如何启动和运行这个 Agent

def main():
    # 创建 Agent 图
    app = create_agent_graph()

    # 准备初始状态（告诉 Agent 用户说了什么，初始值是多少）
    initial_state = {
        "messages": [HumanMessage(content="帮我搜索Python教程")],  # 用户的消息
        "current_step": "init",   # 初始步骤
        "iteration": 0,           # 迭代次数从 0 开始
        "tools_used": [],         # 还没用过任何工具
        "context": {}             # 没有额外上下文
    }

    # 配置（thread_id 用于区分不同的对话）
    config = {"configurable": {"thread_id": "1"}}

    # 🎯 运行！Agent 会自动执行流程图，直到结束
    result = app.invoke(initial_state, config)

    # 查看最终状态
    print("最终状态:")
    print(f"迭代次数: {result['iteration']}")    # 执行了几轮
    print(f"使用工具: {result['tools_used']}")   # 用了哪些工具
    print(f"消息数: {len(result['messages'])}")  # 产生了几条消息

if __name__ == "__main__":
    main()

---

下一节：2.6 小结和复习