LangGraph Memory 学习笔记

State         = Agent 当前正在使用的数据
Checkpointer  = 保存和恢复某个对话线程的 State
Store         = 跨对话保存用户的长期资料
Summarization = 压缩 messages，避免上下文不断增长

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "Look up user information"}
    ],
    "user_id": "user_123",
})

{
    "messages": [...],
    "user_id": "user_123",
}

@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """Look up information about the current user."""
    user_id = runtime.state["user_id"]
    return f"Current user ID: {user_id}"

State 负责运行
Checkpointer 负责保存

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=None,
)

from langchain.agents import AgentState


class CustomState(AgentState):
    user_id: str
    user_name: str
    request_count: int

agent = create_agent(
    model=model,
    tools=tools,
    state_schema=CustomState,
)

state_schema = 定义状态的结构
checkpointer = 保存状态字段的值

from langchain.tools import tool, ToolRuntime


@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """Get information about the current user."""
    user_id = runtime.state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

runtime.state          # 当前线程的 State
runtime.context        # 本次调用的不可变上下文
runtime.store          # 长期记忆 Store
runtime.tool_call_id   # 当前工具调用 ID
runtime.config         # RunnableConfig
runtime.stream_writer  # 自定义流式事件

@tool
def set_name(name: str) -> dict:
    return {"user_name": name}

from langgraph.types import Command


@tool
def set_name(name: str) -> Command:
    return Command(
        update={"user_name": name}
    )

from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime


@tool
def set_name(name: str, runtime: ToolRuntime) -> Command:
    """Set the current user's name."""
    return Command(
        update={
            "user_name": name,
            "messages": [
                ToolMessage(
                    content=f"Name set to {name}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

runtime.state["user_name"] = name

messages
user_id
user_name
request_count

import operator
from typing import Annotated


class CustomState(AgentState):
    request_count: Annotated[int, operator.add]

return {"request_count": 1}

config = {
    "configurable": {
        "thread_id": "thread-001"
    }
}

agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "My name is Bob"}
        ]
    },
    config,
)

agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "What is my name?"}
        ]
    },
    config,
)

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:password@localhost:5432/langgraph"

config = {
    "configurable": {
        "thread_id": "thread-001"
    }
}

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    agent = create_agent(
        model=model,
        tools=tools,
        state_schema=CustomState,
        checkpointer=checkpointer,
    )

    result = agent.invoke(
        {
            "messages": [
                {"role": "user", "content": "Hello"}
            ],
            "user_id": "user_123",
        },
        config,
    )

checkpoint_migrations
checkpoints
checkpoint_blobs
checkpoint_writes

user_123
├── thread_001
├── thread_002
└── thread_003

from dataclasses import dataclass


@dataclass
class AppContext:
    user_id: str

@tool
def get_user_info(runtime: ToolRuntime[AppContext]) -> str:
    user_id = runtime.context.user_id
    return f"User ID: {user_id}"

SummarizationMiddleware(
    model=summarization_model,
    trigger=("tokens", 4000),
    keep=("messages", 20),
)

消息达到触发阈值
    ↓
将较早消息生成摘要
    ↓
保留最近 20 条消息
    ↓
使用摘要替换更早的消息

message.additional_kwargs.get("lc_source") == "summarization"

snapshot = agent.get_state(config)

print(snapshot.values)
print(snapshot.next)
print(snapshot.tasks)
print(snapshot.metadata)

for key, value in snapshot.values.items():
    print(key, value)

for snapshot in agent.get_state_history(config):
    print(snapshot.config)
    print(snapshot.values)
    print(snapshot.next)

checkpoint_tuple = checkpointer.get_tuple(config)

if checkpoint_tuple:
    for task_id, channel, value in checkpoint_tuple.pending_writes:
        print(task_id, channel, value)