跳到主要内容

上下文与状态管理

什么是上下文（Context）​

定义：上下文是消息列表之外的任何数据，能够影响 Agent 行为或工具执行。

上下文的作用​

  • 调整系统提示 — 动态生成个性化提示
  • 为工具提供必要输入 — 如用户 ID、API 凭据
  • 跟踪对话中的事实 — 维护状态一致性

LangGraph 中三种上下文方式​

类型可变性生命周期说明
Configurable❌ 不可变每次运行运行开始时传入的不可变数据
AgentState✅ 可变每次运行/对话执行期间可更改的动态数据
存储（Store）✅ 可变跨对话跨对话共享的持久数据

Configurable（不可变配置）​

定义与特点​

  • 不可变：运行时一次性传入，整个执行过程保持不变
  • 适用场景：用户元数据、API 密钥、用户 ID 等静态数据

使用方式​

from langgraph.prebuilt import create_react_agent

graph = create_react_agent(llm, tools=[...])

# 方式一：在 config 中传入
config = {"configurable": {"user_id": "user_123"}}
graph.invoke({"messages": [{"role": "user", "content": "hi!"}]}, config)

动态生成系统提示词​

from langchain.agents import AgentState
from langchain_core.messages import AnyMessage
from langchain_core.runnables import RunnableConfig

def dynamic_prompt(state: AgentState, config: RunnableConfig) -> list[AnyMessage]:
    """生成动态系统提示词"""
    user_name = config['configurable'].get('user_name', 'zhangsan')
    system_message = f'你是一个智能助手，当前用户的名字是 {user_name}'
    return [{'role': 'system', 'content': system_message}] + state['messages']

graph = create_react_agent(
    llm,
    tools=[...],
    prompt=dynamic_prompt
)

AgentState（可变状态）​

定义与特点​

  • 可变性：✅ 可变，可在执行中多次更新
  • 生命周期：单次运行或对话
  • 角色：Agent 的"短期记忆"，保存执行过程中产生的数据

自定义状态类​

from langgraph.prebuilt.chat_agent_executor import AgentState
from typing import Annotated

# AgentState 继承了 TypedDict，本质是字典
# 包含 messages（消息列表）、is_last_step、remaining_steps 等属性
class CustomState(AgentState):
    username: str  # 自定义字段，存放用户名

在工具中更新状态​

使用 InjectedToolCallId 和 InjectedState 注入状态：

from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.types import Command
from typing import Annotated

@tool
def get_user_name(
    tool_call_id: Annotated[str, InjectedToolCallId],
    config: Annotated[RunnableConfig, '']
) -> Command:
    """获取当前用户的 username"""
    user_name = config['configurable'].get('user_name', 'zhangsan')
    return Command(update={
        'username': user_name,
        'messages': [
            ToolMessage(
                content='成功获取 username',
                tool_call_id=tool_call_id
            )
        ]
    })

@tool
def greet_user(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """根据 username 生成祝福语"""
    user_name = state['username']
    return f'祝贺你：{user_name}'

在 Agent 中使用自定义状态​

graph = create_react_agent(
    llm,
    tools=[get_user_name, greet_user],
    state_schema=CustomState  # 指定自定义状态
)

记忆存储​

短期存储（线程级）​

定义​

通过 checkpointer 和 thread_id 实现，用于跟踪多轮对话。同一 thread_id 下的多次调用共享状态。

内存存储（开发环境）​

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

# 创建内存检查点
checkpointer = InMemorySaver()

graph = create_react_agent(
    llm,
    tools=[...],
    checkpointer=checkpointer
)

# 同一线程，多轮对话
config = {"configurable": {"thread_id": "1"}}

# 第一轮
graph.invoke(
    {"messages": [{"role": "user", "content": "北京天气怎么样？"}]},
    config
)

# 第二轮（自动继承上一轮上下文）
graph.invoke(
    {"messages": [{"role": "user", "content": "那长沙呢？"}]},
    config
)

PostgreSQL 存储（生产环境）​

pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/langgraph_db"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 初次创建数据库
    graph = create_react_agent(llm, tools=[...], checkpointer=checkpointer)

Redis 存储​

pip install -U langgraph langgraph-checkpoint-redis
from langgraph.checkpoint.redis import RedisSaver

DB_URI = "redis://:6379"
with RedisSaver.from_conn_string(DB_URI) as checkpointer:
    graph = create_react_agent(llm, tools=[...], checkpointer=checkpointer)

长期存储（跨线程）​

定义​

用于跨对话（不同 thread_id）持久化用户特定的数据，如用户偏好、记忆信息。

PostgreSQL 实现​

pip install langgraph-store
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://user:password@localhost:5432/langgraph_db"

with (
    PostgresStore.from_conn_string(DB_URI) as store,
    PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
    store.setup()  # 初次创建数据库
    checkpointer.setup()

    graph = create_react_agent(
        llm,
        tools=[...],
        checkpointer=checkpointer,
        store=store
    )

使用 API​

config = {"configurable": {"thread_id": "1"}}

# 获取当前状态
state = graph.get_state(config=config)

# 获取历史状态
history = graph.get_state_history(config=config)

速记要点​

概念关键点记忆
Configurable不可变，一次传入配置文件
AgentState可变，运行时更新内存变量
checkpointer短期记忆，thread_id会话级
store长期记忆，跨对话持久化

核心对比​

  • Configurable vs AgentState：

    • Configurable = 不可变的"静态配置"
    • AgentState = 可变的"动态状态"

  • 短期 vs 长期：

    • 短期 = 同一 thread_id 的多轮对话
    • 长期 = 不同 thread_id 共享的数据

短期记忆 vs 长期记忆​

维度短期记忆（checkpointer）长期记忆（store）
数据内容对话历史（消息列表）用户画像、偏好、知识
生命周期同一 thread_id跨所有 thread_id
谁写入自动（Agent 自动记录）手动（工具中写入）
谁使用自动手动（提示词中读取）
"短期"含义作用域小，仅限当前会话永久保存

关键区别：

  • 短期记忆：同一 thread_id 的新对话能"继承"历史，不同 thread_id 则无法访问
  • 长期记忆：任意 thread_id 都能访问的共享数据

长期记忆使用方式：

from langgraph.store import BaseStore

# 写入（在工具中）
@tool
def remember_preference(store: BaseStore, preference: str) -> str:
    """记住用户偏好"""
    store.put(("preference", "drink"), {"value": preference})
    return "已记住"

# 读取（在提示词中）
def prompt_with_memory(state, config, store: BaseStore):
    pref = store.get(("preference", "drink"))
    system_msg = f"用户偏好：{pref.value}" if pref else "新用户"
    return [{"role": "system", "content": system_msg}] + state["messages"]

记忆口诀​

  • Configurable：开头传什么，后面用什么
  • AgentState：工具返回 Command，状态自动更新
  • 短期记忆：checkpointer + thread_id
  • 长期记忆：store 跨对话

一句话总结​

LangGraph 状态管理 = Configurable（静态） + AgentState（动态） + checkpointer（会话） + store（持久）