用法
要为代理添加长期记忆,请创建一个存储并将其传递给create_agent:
- InMemoryStore
- PostgreSQL
from langchain.agents import create_agent
from langchain_core.runnables import Runnable
from langgraph.store.memory import InMemoryStore
# InMemoryStore 将数据保存到内存字典中。在生产环境中请使用基于数据库的存储。
store = InMemoryStore()
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[],
store=store,
)
pip install langgraph-checkpoint-postgres
from langchain.agents import create_agent
from langchain_core.runnables import Runnable
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
store.setup()
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[],
store=store,
)
runtime.store 参数从存储中读取和写入数据。示例请参阅在工具中读取长期记忆和从工具中写入长期记忆。
要深入了解记忆类型(语义、情景、程序性)以及编写记忆的策略,请参阅记忆概念指南。
记忆存储
LangGraph 将长期记忆作为 JSON 文档存储在存储中。 每条记忆都组织在一个自定义的namespace(类似于文件夹)和一个独特的 key(类似于文件名)下。命名空间通常包含用户或组织 ID 或其他标签,以便于组织信息。
这种结构支持记忆的分层组织。然后可以通过内容过滤器支持跨命名空间搜索。
- InMemoryStore
- PostgreSQL
from collections.abc import Sequence
from langgraph.store.base import IndexConfig
from langgraph.store.memory import InMemoryStore
def embed(texts: Sequence[str]) -> list[list[float]]:
# 替换为实际的嵌入函数或LangChain嵌入对象
return [[1.0, 2.0] for _ in texts]
# InMemoryStore将数据保存到内存字典中。生产环境中请使用基于数据库的存储。
store = InMemoryStore(index=IndexConfig(embed=embed, dims=2))
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)
store.put(
namespace,
"a-memory",
{
"rules": [
"用户喜欢简短、直接的语言",
"用户只说英语和Python",
],
"my-key": "my-value",
},
)
# 通过ID获取"记忆"
item = store.get(namespace, "a-memory")
# 在此命名空间内搜索"记忆",根据内容等价性进行过滤,并按向量相似度排序
items = store.search(
namespace, filter={"my-key": "my-value"}, query="语言偏好"
)
from collections.abc import Sequence
from langgraph.store.base import IndexConfig
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
def embed(texts: Sequence[str]) -> list[list[float]]:
# 替换为实际的嵌入函数或LangChain嵌入对象
return [[1.0, 2.0] for _ in texts]
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresStore.from_conn_string(
DB_URI,
index=IndexConfig(embed=embed, dims=2), # type: ignore[arg-type]
) as store:
store.setup()
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)
store.put(
namespace,
"a-memory",
{
"rules": [
"用户喜欢简短、直接的语言",
"用户只说英语和Python",
],
"my-key": "my-value",
},
)
item = store.get(namespace, "a-memory")
items = store.search(
namespace, filter={"my-key": "my-value"}, query="语言偏好"
)
在工具中读取长期记忆
- InMemoryStore
- PostgreSQL
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
# InMemoryStore 将数据保存到内存字典中。在生产环境中请使用数据库支持的存储。
store = InMemoryStore()
# 使用 put 方法将示例数据写入存储
store.put(
(
"users",
), # 用于将相关数据分组在一起的命名空间(用于用户数据的 users 命名空间)
"user_123", # 命名空间内的键(用户 ID 作为键)
{
"name": "John Smith",
"language": "English",
}, # 为给定用户存储的数据
)
@tool
def get_user_info(runtime: ToolRuntime[Context]) -> str:
"""查找用户信息。"""
# 访问存储 - 与提供给 `create_agent` 的存储相同
assert runtime.store is not None
user_id = runtime.context.user_id
# 从存储中检索数据 - 返回包含值和元数据的 StoreValue 对象
user_info = runtime.store.get(("users",), user_id)
return str(user_info.value) if user_info else "Unknown user"
agent: Runnable = create_agent(
model="google_genai:gemini-3.1-pro-preview",
tools=[get_user_info],
# 将存储传递给代理 - 使代理在运行工具时能够访问存储
store=store,
context_schema=Context,
)
# 运行代理
agent.invoke(
{"messages": [{"role": "user", "content": "look up user information"}]},
context=Context(user_id="user_123"),
)
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
@dataclass
class Context:
user_id: str
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
store.setup()
store.put(("users",), "user_123", {"name": "John Smith", "language": "English"})
@tool
def get_user_info(runtime: ToolRuntime[Context]) -> str:
"""查询用户信息。"""
assert runtime.store is not None
user_info = runtime.store.get(("users",), runtime.context.user_id)
return str(user_info.value) if user_info else "未知用户"
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[get_user_info],
store=store,
context_schema=Context,
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "查询用户信息"}]},
context=Context(user_id="user_123"),
)
从工具中写入长期记忆
- InMemoryStore
- PostgreSQL
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.memory import InMemoryStore
from typing_extensions import TypedDict
# InMemoryStore 将数据保存到内存字典中。在生产环境中,请使用基于数据库的存储。
store = InMemoryStore()
@dataclass
class Context:
user_id: str
# TypedDict 定义了传递给 LLM 的用户信息结构
class UserInfo(TypedDict):
name: str
# 允许代理更新用户信息的工具(对聊天应用很有用)
@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
"""保存用户信息。"""
# 访问存储 - 与提供给 `create_agent` 的存储相同
assert runtime.store is not None
store = runtime.store
user_id = runtime.context.user_id
# 将数据存储在存储中(命名空间,键,数据)
store.put(("users",), user_id, dict(user_info))
return "Successfully saved user info."
agent: Runnable = create_agent(
model="google_genai:gemini-3.1-pro-preview",
tools=[save_user_info],
store=store,
context_schema=Context,
)
# 运行代理
agent.invoke(
{"messages": [{"role": "user", "content": "My name is John Smith"}]},
# 在上下文中传递 user_id 以标识正在更新谁的信息
context=Context(user_id="user_123"),
)
# 你可以直接访问存储来获取值
item = store.get(("users",), "user_123")
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
from typing_extensions import TypedDict
@dataclass
class Context:
user_id: str
class UserInfo(TypedDict):
name: str
@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
"""保存用户信息。"""
assert runtime.store is not None
runtime.store.put(("users",), runtime.context.user_id, dict(user_info))
return "用户信息保存成功。"
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
store.setup()
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[save_user_info],
store=store,
context_schema=Context,
)
agent.invoke(
{"messages": [{"role": "user", "content": "My name is John Smith"}]},
context=Context(user_id="user_123"),
)
将这些文档连接到 Claude、VSCode 等,通过 MCP 获取实时答案。

