LangGraph 核心之状态（State）

LangGraph 的核心在于其简洁而强大的图计算模型，这一模型的基石由四个核心原语构成：状态（State）、节点（Node）、边（Edge），以及命令（Command）。
理解这四个原语的概念及其相互作用方式，是掌握 LangGraph 并构建复杂智能体系统的重中之重。可以将这四个原语比作乐高积木最基本的、也是最核心的模块，理解了它们，就如同掌握了乐高搭建的"语言"，后续才能使用更高级的技巧，搭建出各种各样精巧、复杂、功能强大的智能体系统。

本章将深入探讨 LangGraph 核心原语之状态（State）

状态（State）

在 LangGraph 中，状态是贯穿智能体系统运行始终的核心概念。我们可以将其理解为智能体的"短期记忆"、“工作记忆"或者"临时共享数据空间”，它承载着智能体在执行过程中产生的各种信息，例如用户的输入、中间计算结果、工具的输出、对话历史等等。

LangGraph 在状态定义上提供了极大的灵活性，允许开发者根据实际应用的需求，选择最合适的数据结构来表示状态。

示例 1-1：使用 TypedDict 和 Pydantic 定义状态

from typing_extensions import TypedDict
from pydantic import BaseModel, field_validator

# 使用 TypedDict 定义状态
class TypedDictState(TypedDict):
    user_input: str
    agent_response: str
    tool_output: str

# 使用 Pydantic 定义状态，并进行数据验证
class PydanticState(BaseModel):
    user_input: str
    agent_response: str
    tool_output: str
    mood: str = "neutral"  # 默认情绪状态为 neutral

    @field_validator('mood')
    @classmethod
    def validate_mood(cls, value):
        if value not in ["happy", "sad", "neutral"]:
            raise ValueError("情绪状态必须是 'happy', 'sad' 或 'neutral'")
        return value

# 测试状态定义
print("TypedDict 状态示例:")
typed_state = {"user_input": "Hello", "agent_response": "Hi there!", "tool_output": "weather data"}
print(typed_state)

print("\nPydantic 状态示例:")
pydantic_state = PydanticState(
    user_input="Hello", 
    agent_response="Hi there!", 
    tool_output="weather data",
    mood="happy"
)
print(pydantic_state.model_dump())

TypedDict 状态示例:
{'user_input': 'Hello', 'agent_response': 'Hi there!', 'tool_output': 'weather data'}

Pydantic 状态示例:
{'user_input': 'Hello', 'agent_response': 'Hi there!', 'tool_output': 'weather data', 'mood': 'happy'}

💡 核心概念解析：

• TypedDict：可以快速定义简单的状态结构，提供类型提示但不进行运行时验证
• Pydantic：提供更强大的数据建模和验证能力，能够在运行时进行数据验证，确保状态的类型和取值符合预期
• 状态的作用：作为各节点间信息传递的桥梁，也是智能体进行决策和行为调整的重要依据

示例 1-2：使用多结构体实现私有状态

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
import json

# 定义全局的公共状态 Schema
class OverallState(TypedDict):
    user_input: str
    agent_response: str

# 定义节点的私有状态 Schema
class ToolState(TypedDict):
    api_key: str
    tool_config: dict
    user_input: str  # 需要包含从公共状态传递的数据

# 模拟 API 客户端类
class MockAPIClient:
    def __init__(self, api_key: str, config: dict):
        self.api_key = api_key
        self.config = config
        print(f"初始化 API 客户端，API Key: {api_key[:8]}..., 配置: {config}")

    def call_api(self, user_input: str) -> str:
        # 模拟 API 调用
        response = f"基于输入 '{user_input}' 和配置 {self.config}，API 返回处理结果"
        print(f"调用 API，输入: {user_input}")
        return response

def create_api_client(api_key: str, tool_config: dict) -> MockAPIClient:
    """创建 API 客户端的工厂函数"""
    return MockAPIClient(api_key, tool_config)

# 定义一个使用私有状态的节点
def tool_node(state: ToolState) -> OverallState:
    """使用私有状态的工具节点"""
    print(f"工具节点接收到私有状态: {state}")

    # 节点逻辑，例如调用工具 API 并根据 ToolState 中的配置进行操作
    api_client = create_api_client(state['api_key'], state['tool_config'])
    response = api_client.call_api(state['user_input'])

    return {"agent_response": response}  # 返回更新后的公共状态

# 定义一个输入处理节点
def input_processor(state: OverallState) -> OverallState:
    """处理用户输入的节点"""
    print(f"输入处理节点接收到: {state['user_input']}")
    processed_input = f"已处理: {state['user_input']}"
    return {"user_input": processed_input}

# 定义状态适配节点，将公共状态转换为私有状态
def state_adapter(state: OverallState) -> dict:
    """适配器节点：将公共状态转换为私有状态"""
    from langgraph.constants import Send

    # 创建私有状态数据
    private_state = {
        "api_key": "secret_api_key_12345",
        "tool_config": {
            "timeout": 30,
            "retry_count": 3,
            "endpoint": "https://api.example.com"
        },
        "user_input": state["user_input"]
    }

    # 使用 Send 将私有状态发送给工具节点
    return Send("tool_node", private_state)

# 构建图
builder = StateGraph(OverallState)

# 添加节点
builder.add_node("input_processor", input_processor)
builder.add_node("state_adapter", state_adapter)
builder.add_node("tool_node", tool_node)

# 定义边
builder.add_edge(START, "input_processor")
builder.add_conditional_edges("input_processor", state_adapter, ["tool_node"])
builder.add_edge("tool_node", END)

# 编译图
graph = builder.compile()

# 测试运行
print("=== 多结构体状态管理示例 ===\n")

# 初始状态
initial_state = {
    "user_input": "查询天气信息",
    "agent_response": ""
}

print(f"初始状态: {initial_state}\n")

# 运行图
try:
    final_state = graph.invoke(initial_state)
    print(f"\n最终状态: {final_state}")

    print(f"\n=== 执行结果 ===")
    print(f"用户输入: {final_state['user_input']}")
    print(f"智能体响应: {final_state['agent_response']}")

except Exception as e:
    print(f"执行出错: {e}")

=== 多结构体状态管理示例 ===

初始状态: {'user_input': '查询天气信息', 'agent_response': ''}

输入处理节点接收到: 查询天气信息
工具节点接收到私有状态: {'api_key': 'secret_api_key_12345', 'tool_config': {'timeout': 30, 'retry_count': 3, 'endpoint': 'https://api.example.com'}, 'user_input': '已处理: 查询天气信息'}
初始化 API 客户端，API Key: secret_a..., 配置: {'timeout': 30, 'retry_count': 3, 'endpoint': 'https://api.example.com'}
调用 API，输入: 已处理: 查询天气信息

最终状态: {'user_input': '已处理: 查询天气信息', 'agent_response': "基于输入 '已处理: 查询天气信息' 和配置 {'timeout': 30, 'retry_count': 3, 'endpoint': 'https://api.example.com'}，API 返回处理结果"}

=== 执行结果 ===
用户输入: 已处理: 查询天气信息
智能体响应: 基于输入 '已处理: 查询天气信息' 和配置 {'timeout': 30, 'retry_count': 3, 'endpoint': 'https://api.example.com'}，API 返回处理结果

这个完整的示例展示了：

1. 多结构体定义：
   • OverallState：全局公共状态，包含用户输入和智能体响应
   • ToolState：私有状态，包含 API 密钥、工具配置和用户输入
2. 状态转换机制：
   • state_adapter 节点负责将公共状态转换为私有状态
   • 使用 Send API 将私有状态发送给特定节点
3. 私有状态的使用：
   • tool_node 接收私有状态，包含敏感信息（API 密钥、配置）
   • 节点处理完成后返回公共状态格式的更新
4. 完整的工作流：
   • 输入处理 → 状态适配 → 工具调用 → 结果返回

示例 1-3：定义输入/输出结构体

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# 定义内部的、全面的状态结构体
class InternalState(TypedDict):
    user_query: str
    search_results: list[str]
    llm_response: str
    debug_info: str  # 内部调试信息，不需要对外暴露

# 定义输入结构体 (只包含 user_query)
class InputSchema(TypedDict):
    user_query: str

# 定义输出结构体 (只包含 llm_response)
class OutputSchema(TypedDict):
    llm_response: str

# 模拟搜索节点
def search_node(state: InternalState) -> InternalState:
    query = state["user_query"]
    print(f"🔍 搜索节点: 处理查询 '{query}'")

    # 模拟搜索结果
    mock_results = [
        f"搜索结果1: 关于 '{query}' 的信息",
        f"搜索结果2: {query} 相关数据",
        f"搜索结果3: {query} 详细说明"
    ]

    debug = f"搜索完成，找到 {len(mock_results)} 条结果"
    print(f"📊 {debug}")

    return {
        "search_results": mock_results,
        "debug_info": debug
    }

# LLM 处理节点
def llm_node(state: InternalState) -> InternalState:
    query = state["user_query"]
    results = state["search_results"]

    print(f"🤖 LLM 节点: 基于 {len(results)} 条搜索结果生成回复")

    # 模拟 LLM 生成响应
    combined_info = " | ".join(results[:2])  # 使用前两条结果
    response = f"基于您的查询 '{query}'，我找到了以下信息：{combined_info}。希望这能帮到您！"

    debug = f"LLM 处理完成，生成 {len(response)} 字符的回复"
    print(f"💭 {debug}")
    print(f"📝 生成回复: {response[:50]}...")

    return {
        "llm_response": response,
        "debug_info": state.get("debug_info", "") + f" | {debug}"
    }

# 创建图并指定输入输出结构体
print("🏗️ 构建 LangGraph（带输入输出结构体约束）...")
builder = StateGraph(InternalState, input=InputSchema, output=OutputSchema)

# 添加节点
builder.add_node("search", search_node)
builder.add_node("llm", llm_node)

# 定义边
builder.add_edge(START, "search")
builder.add_edge("search", "llm")
builder.add_edge("llm", END)

# 编译图
graph = builder.compile()
print("✅ 图构建完成！")

# 测试运行
print("\n=== 🚀 输入输出结构体约束示例 ===")

# 创建符合 InputSchema 的输入
input_data = {"user_query": "什么是人工智能"}
print(f"📥 输入数据 (InputSchema): {input_data}")

# 运行图
result = graph.invoke(input_data)

# 结果自动符合 OutputSchema 格式
print(f"\n📤 输出数据 (OutputSchema): {result}")
print(f"📊 输出类型: {type(result)}")
print(f"📋 输出键: {list(result.keys())}")

🏗️ 构建 LangGraph（带输入输出结构体约束）...
✅ 图构建完成！

=== 🚀 输入输出结构体约束示例 ===
📥 输入数据 (InputSchema): {'user_query': '什么是人工智能'}
🔍 搜索节点: 处理查询 '什么是人工智能'
📊 搜索完成，找到 3 条结果
🤖 LLM 节点: 基于 3 条搜索结果生成回复
💭 LLM 处理完成，生成 80 字符的回复
📝 生成回复: 基于您的查询 '什么是人工智能'，我找到了以下信息：搜索结果1: 关于 '什么是人工智能' 的信息 ...

📤 输出数据 (OutputSchema): {'llm_response': "基于您的查询 '什么是人工智能'，我找到了以下信息：搜索结果1: 关于 '什么是人工智能' 的信息 | 搜索结果2: 什么是人工智能 相关数据。希望这能帮到您！"}
📊 输出类型: <class 'dict'>
📋 输出键: ['llm_response']

1. 三层状态结构体：
   • InternalState：内部完整状态，包含所有中间数据
   • InputSchema：外部输入接口，只需要用户查询
   • OutputSchema：外部输出接口，只返回最终回复
2. 状态封装：
   • 内部节点可以访问和修改完整的内部状态
   • 外部只能看到定义的输入输出格式
   • 调试信息、搜索结果等中间数据被隐藏
3. 实际工作流程：
   • 搜索节点模拟信息检索
   • LLM 节点基于搜索结果生成回复
   • 完整的状态管理和数据流
4. 输入输出约束验证：
   • 输入只需要符合 InputSchema
   • 输出自动符合 OutputSchema
   • 内部复杂性被完全封装

1.1 状态 Reducer

状态 Reducer 是 LangGraph 提供的用于自定义状态更新逻辑的核心机制。它允许我们精细地控制状态在节点执行过程中的演变方式，尤其是在处理并发状态更新、复杂数据结构以及需要特定合并策略的场景下。

示例 1：使用状态 Reducer

from typing_extensions import TypedDict, Annotated
from operator import add
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

# 定义状态 Schema，并为 'message_history' 键指定 add_messages Reducer
class ChatState(TypedDict):
    message_history: Annotated[list[BaseMessage], add_messages]
    user_intent: str
    tool_output: str

# 演示 add_messages Reducer 的工作方式
print("演示 add_messages Reducer：")

# 初始状态
initial_state = {
    "message_history": [HumanMessage(content="Hello")],
    "user_intent": "greeting",
    "tool_output": ""
}

print("初始状态:")
for msg in initial_state["message_history"]:
    print(f"  {type(msg).__name__}: {msg.content}")

# 新的消息更新
new_messages = [AIMessage(content="Hi there! How can I help you?")]
print("\n添加新消息:")
for msg in new_messages:
    print(f"  {type(msg).__name__}: {msg.content}")

# add_messages Reducer 会自动合并消息
updated_messages = add_messages(initial_state["message_history"], new_messages)
print("\n合并后的消息历史:")
for msg in updated_messages:
   print(f"  {type(msg).__name__}: {msg.content}")

演示 add_messages Reducer：
初始状态:
  HumanMessage: Hello

添加新消息:
  AIMessage: Hi there! How can I help you?

合并后的消息历史:
  HumanMessage: Hello
  AIMessage: Hi there! How can I help you?

示例 2 & 3：自定义 Reducer 函数

def reducer_extend_unique(left: list[str] | None, right: list[str] | None) -> list[str]:
    """
    自定义 Reducer 函数，用于合并两个字符串列表，并进行去重
    """
    existing_items = left if left else []  # 如果 left 为 None，则初始化为空列表
    new_items = right if right else []     # 如果 right 为 None，则初始化为空列表
    combined_items = existing_items + new_items
    return list(set(combined_items))       # 使用 set 去重并转换为 list 返回

# 测试自定义 Reducer
print("测试自定义 Reducer：")

existing_list = ["apple", "banana", "orange"]
new_list = ["banana", "grape", "apple", "mango"]

print(f"已有列表: {existing_list}")
print(f"新增列表: {new_list}")

result = reducer_extend_unique(existing_list, new_list)
print(f"合并去重结果: {result}")

# 在状态结构体中应用自定义 Reducer
class ChatStateWithCustomReducer(TypedDict):
    message_history: Annotated[list[BaseMessage], add_messages]
    user_intent: str
    tool_output: str
    item_list: Annotated[list[str], reducer_extend_unique]  # 应用自定义 Reducer

测试自定义 Reducer：
已有列表: ['apple', 'banana', 'orange']
新增列表: ['banana', 'grape', 'apple', 'mango']
合并去重结果: ['banana', 'grape', 'orange', 'apple', 'mango']

1.2 Message 与 MessagesState

在构建对话型 AI 智能体时，对话历史至关重要。LangGraph 引入了消息（Message）和 MessagesState 的概念，专门优化对话场景的状态管理。

LangChain 定义了多种消息类型：

• HumanMessage：代表人类用户的消息
• AIMessage：代表 AI 模型生成的消息
• ToolMessage：代表工具执行后的输出结果消息
• SystemMessage：代表系统发出的消息

示例 1：使用 MessagesState 定义状态结构体

from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage

class MyChatState(MessagesState):
    """
    自定义的 ChatState, 继承自 MessagesState, 
    自动包含 messages 状态键和 add_messages Reducer
    """
    user_intent: str
    tool_output: str
    # ... 可以添加其他自定义的状态键 ...

# 演示不同类型的消息
print("演示不同类型的消息：")

# 创建不同类型的消息
human_msg = HumanMessage(content="你好，我想查询今天的天气")
system_msg = SystemMessage(content="你是一个专业的天气助手，请友好地回复用户")
ai_msg = AIMessage(content="好的，我来帮你查询天气信息")
tool_msg = ToolMessage(content="北京今天晴转多云，温度 20-28°C", tool_call_id="weather_001")

messages = [system_msg, human_msg, ai_msg, tool_msg]

for i, msg in enumerate(messages, 1):
    print(f"{i}. {type(msg).__name__}: {msg.content}")

# 演示 MessagesState 的使用
print("\n演示 MessagesState 的状态管理：")
chat_state = {
    "messages": messages,
    "user_intent": "weather_query",
    "tool_output": "weather_data_retrieved"
}

print(f"对话轮次: {len(chat_state['messages'])}")
print(f"用户意图: {chat_state['user_intent']}")
print(f"工具输出: {chat_state['tool_output']}")

演示不同类型的消息：
1. SystemMessage: 你是一个专业的天气助手，请友好地回复用户
2. HumanMessage: 你好，我想查询今天的天气
3. AIMessage: 好的，我来帮你查询天气信息
4. ToolMessage: 北京今天晴转多云，温度 20-28°C

演示 MessagesState 的状态管理：
对话轮次: 4
用户意图: weather_query
工具输出: weather_data_retrieved

💡 MessagesState 核心特性：

• 内置 messages 状态键：自动提供消息列表管理
• 默认 add_messages Reducer：自动处理消息追加、更新和去重
• 消息序列化与反序列化：支持 JSON 兼容的字典格式传递消息数据
• 可扩展性：可以自由添加其他自定义状态键

总结

在 LangGraph 框架中，State 是一个核心概念，用于在对话系统或复杂任务中保存和共享上下文信息，帮助跟踪用户输入、系统生成的中间结果，以及动态的对话状态。它通常以键值对的形式存在，类似于 Python 的字典，帮助跟踪用户输入和系统状态。开发者可以使用 TypedDict、Pydantic 模型或数据类来定义状态，并通过 Reducer 机制控制状态更新。掌握这些知识可以帮助开发者灵活构建各种 LangGraph 应用。

参考文献和资料
• LangGraph文档：https://docs.langchain.com/oss/python/langgraph/overview
• 《LangGraph实战》