长期记忆与运行时上下文

长期记忆与运行时上下文

2026年9月21日·#编程学习/langchain学习笔记LangChain/AI·9607 字 48 分钟
浏览量加载中...
AI 摘要

长期记忆的三类划分、store→namespace→key→value 四层存储架构、put/get/search 三个 API、在工具与中间件里读写记忆，以及何时写入记忆与静态运行时上下文

短期 vs 长期：一句话区分#

短期记忆长期记忆
记录范围会话级（线程 Thread）的数据用户特定或应用级的数据
会话间不共享任何会话都可以随时访问
例子刚才聊了什么“你喜欢简短回答”、“你偏好 Python”、“某个用户是 VIP”、“某个流程过去怎么做效果更好”

图：短期记忆走 Checkpointer、长期记忆走 Store，最后一起喂给模型

这类信息不属于某一条聊天，而属于”用户/组织/应用本身”——所以不能塞在某个 thread 的 messages 里。

长期记忆分三类#

LangChain 参考 CoALA 论文把长期记忆划分为三类：

类型存什么例子
Semantic（语义记忆）事实用户喜欢简洁回答、常用中文、某公司属于哪个行业
Episodic（情景记忆）经验过去某个任务是怎么成功的、某种输入下怎么回答效果最好（在 Agent 里常表现为 few-shot 示例）
Procedural（程序性记忆）规则/做事方法系统提示词、工作流程、工具调用规则

情景记忆的思路值得记住：不直接告诉模型规则，而是给它看几个”输入 → 输出”的例子，让它照着学。

存储架构：四层结构#

长期记忆的存储是 store → namespace → key → value 的四层架构：

层说明数据类型
Store（记忆仓库）langgraph.store.base.BaseStore 的子类实例。常用实现：InMemoryStore（测试）、PostgresStore（生产）对象
Namespace（命名空间）层级路径，作用很像”文件路径/文件夹”，用于分组和隔离记忆字符串元组 tuple[str, ...]
Key（键）该 namespace 下的唯一标识str
Value（值）存储的内容dict[str, Any]
namespace = ("users", "user_123", "preferences")   # 元组类型
key = "profile"                                     # 字符串类型
value = {                                           # 字典类型
    "language": "zh-CN",
    "style": "short_direct",
    "likes": ["python", "rag"],
}
store.put(namespace, key, value)

同一个 AI 应用通常这样组织：每个会话有自己的短期状态 State（按 thread_id 区分），而长期记忆共享同一个 Store 实例，再通过 namespace 区分不同用户、组织、业务域：

AI 应用
├─ thread_id = t1 → state_1
│   ├─ messages = [{"role": "user", "content": "我想去北京旅游"}, ...]
│   ├─ current_intent = "travel_planning"
│   └─ collected_slots = {"destination": "北京"}
├─ thread_id = t2 → state_2
│   ├─ messages = [{"role": "user", "content": "帮我写周报"}]
│   └─ ...
└─ store（跨会话共享）
    └─ namespace ("users", "alice", "memories")
        ├─ key "pref_food" → {...}
        └─ key "pref_lang" → {...}

三个基础 API#

LangChain 1.2.x 的长期记忆基于 store 持久化数据，API 有三个：put() 写入、get() 读取、search() 检索。它们可以在 Agent 执行流程之外直接调用。

put() / get()#

store.put(
    ("users", "Alice", "memories"),      # namespace
    "pref_food",                          # key
    {"category": "food", "text": "Alice likes sushi"},   # value
)


item = store.get(("users", "Alice", "memories"), "pref_food")
print(item.value)      # {'category': 'food', 'text': 'Alice likes sushi'}

put() 的签名（本机核对过源码）：

def put(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None:
参数说明
namespace / key / value层级路径 / 唯一键 / 要保存的 JSON-like 字典
index控制语义检索索引，三档语义见下表
ttl可选过期时间（是否支持取决于具体 store 实现）

index 的三种取值要分清：

取值语义
None（默认）用 store 初始化时配置的索引配置；如果初始化时没有指定索引策略，这个参数会被直接忽略
False不为该 item 建语义索引（数据照样存，只是搜不到）
list[str]只对指定字段路径建索引

get() 的签名：

def get(
    self,
    namespace: tuple[str, ...],
    key: str,
    *,
    refresh_ttl: bool | None = None,
) -> Item | None:
参数说明
namespace / key层级路径 / 该路径下的唯一键
refresh_ttl是否刷新当前 item 的 TTL（time-to-live，存活时间）。默认 None 表示沿用创建 store 时指定的同名配置；没配置 TTL 时该参数被忽略
Tip

实测 get() 返回的不是裸 value，而是一个 Item 对象（字段有 key、namespace、value、created_at、updated_at、dict）。所以取值要写 item.value——别忘了 .value。

created_at / updated_at：InMemoryStore 与 PostgresStore 不一样#

Item 上新增了 created_at（创建时间）和 updated_at（更改时间），但两种 store 的语义完全不同：

InMemoryStorePostgresStore
更新逻辑每次 put 都新建一个 Item 对象（无论 namespace/key 是否相同）真正的 update，不是覆盖
created_at每次 put 都被重置固定为 Item 首次创建的时间
updated_at跟着一起重置每次更新时改变
两者关系永远相等（同一次 datetime.now()）可能不同，符合直觉

看源码就明白了——InMemoryStore 的 _apply_put_ops 里，不管这条 key 之前存不存在，都直接盖一个新 Item 上去：

def _apply_put_ops(self, put_ops):
    for (namespace, key), op in put_ops.items():
        if op.value is None:
            self._data[namespace].pop(key, None)
            self._vectors[namespace].pop(key, None)
        else:
            self._data[namespace][key] = Item(
                value=op.value,
                key=key,
                namespace=namespace,
                created_at=datetime.now(timezone.utc),   # ← 每次都重新取
                updated_at=datetime.now(timezone.utc),   # ← 每次都重新取
            )

PostgresStore 的用法和 PostgresSaver 如出一辙——from_conn_string + setup()：

from langgraph.store.postgres import PostgresStore


DB_URL = "postgresql://langchain_user:abcd1234@<你的IP>:5432/langchain_db?sslmode=disable"


with PostgresStore.from_conn_string(DB_URL) as store:
    store.setup()                                    # ← 第一次跑，建 store 相关表
    store.put(("users",), "user-11", {"name": "小蓝"})
    print(store.get(("users",), "user-11"))
    # Item(..., created_at='2026-06-11T23:24:31.132342+08:00', updated_at='2026-06-11T23:24:31.132342+08:00')


    store.put(("users",), "user-11", {"name": "小红"})   # 更新同一条
    print(store.get(("users",), "user-11"))
    # Item(..., created_at='2026-06-11T23:24:31.132342+08:00',   ← 没变
    #            updated_at='2026-06-11T23:26:11.956257+08:00')   ← 变了
Tip

本机实测（InMemoryStore，同一 namespace/key 先写”小蓝”、隔 1.2 秒再写”小红”）：

首次 put: Item(..., created_at='...12:09:27.014794+00:00', updated_at='...12:09:27.014800+00:00')
再次 put: Item(..., created_at='...12:09:28.215231+00:00', updated_at='...12:09:28.215235+00:00')

两次的 created_at 都跟着 put 一起往后跳了——证明”更新”在 InMemoryStore 里其实是新建。注意：两次时间戳相差仅 6 微秒（014794 vs 014800），是同一行里两次 datetime.now() 的差值，不是”相等”——课程正文写的是”created_at 和 updated_at 始终一致”，本机看是”始终相差几微秒”，但结论一样：它们总是同一次 put 的产物，不能用来区分创建和修改。要区分创建/修改时间，得换 PostgresStore。

store.search(
    ("users",),                       # namespace_prefix：在该前缀下搜索
    query="用户喜欢什么食物",           # 语义检索的自然语言（需要向量支持）
    filter={"category": "food"},       # 结构化过滤：按 value 里的键值对筛
    limit=10,
    offset=0,
)
参数说明
namespace_prefix命名空间前缀，在该前缀下搜索（位置参数）
query语义检索用的自然语言，需要把输入转成向量
filter结构化过滤：用 value 里的键值对筛选
limit / offset最多返回几条 / 跳过前几条（相当于 SQL 的 limit / offset）
refresh_ttl是否刷新过期时间

它支持两种检索方式：

  1. 按 filter 做结构化过滤（精确匹配，不需要向量）
  2. 按 query 做语义相似度检索（需要向量支持）

返回值是 SearchItem 列表，额外携带匹配分数等检索元信息。

Tip

实测（InMemoryStore）：search(("users",)) 能按前缀拿到全部 2 条；search(ns, filter={"category": "code"}) 精确筛出 pref_lang；limit=1 只返回第一条。结构化过滤不需要任何模型/向量，开箱即用。

让 search() 支持语义检索：index 配置#

上面说的”按 query 语义检索”不是自动就有的——得在创建 store 时通过 index 参数配置，否则 query 无从算起。

from langgraph.store.memory import InMemoryStore


index_config = {
    "embed": embed,        # 把文本转成向量的"嵌入函数"
    "dims": 6,             # 输出向量维度
    "fields": ["$", "course"],   # 给 value 里哪些字段建索引
}
store = InMemoryStore(index=index_config)

IndexConfig 的三个字段（源码）：

class IndexConfig(TypedDict, total=False):
    dims: int
    embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str
    fields: list[str] | None
字段说明
embed把输入文本转成向量的嵌入函数。可以是自定义函数、嵌入模型对象，也可以直接写字符串（模型名，如 "openai:text-embedding-3-large"）
dims输出向量维度，必须和 embed 实际产出的维度一致
fields用于计算向量的属性列表（都是 value 里的 key）。四种写法可以同时出现

fields 的四种路径写法（这里讲得比”入门”细，值得单独记）：

写法含义
["$"]把整个 value 当作一个整体嵌入
["field1", "field2"]单独指定某个一级字段
["parent.child"]从嵌套 JSON 对象里取子字段的值
["array[*].field"]从 JSON 数组的每个对象里取子字段的值
Note

fields 列表的每个元素都会生成一个嵌入向量——所以 ["$", "course"] 会为一条 item 生成 2 个向量（一条按整体、一条按 course）。检索时底层做的是 max pooling（同一个 item 的多个向量取最高分），所以你不用操心”命中哪个字段”。

实测：索引到底建了什么#

用自定义嵌入函数（返回全 1 向量）能直接把向量结构打出来：

from pprint import pprint


def embed(texts: list[str]) -> list[list[float]]:
    return [[1.0] * 6 for _ in texts]


index_config = {"embed": embed, "dims": 6, "fields": ["$", "course"]}
store = InMemoryStore(index=index_config)
store.put(("users", "Alice", "memories"), "preferences",
          {"course": "计算机组成原理", "sports": "跑步", "food": "紫光园奶皮子酸奶"})


pprint(store._vectors)      # ← 私有属性，仅用于观察

store._vectors 是个 defaultdict，三层结构 namespace → key → 字段名 → 向量：

{('users', 'Alice', 'memories'): {'preferences': {'$': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
                                                  'course': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0]}}}

也能按 namespace + key + 字段名只取一个向量：

pprint(store._vectors[('users', 'Alice', 'memories')]['preferences']['$'])
# [1.0, 1.0, 1.0, 1.0, 1.0, 1.0]
Tip

本机实测（自定义嵌入函数记录每次被喂进去的文本，fields=["$", "course", "info.city", "tags[*].name"]）：

传给 embed 的文本:
    ['{"course": "数字电路", "info": {"city": "北京", "zip": "100000"}, "tags": [{"name": "ai"}, {"name": "rag"}]}',   ← $
     '数字电路',        ← course（一级字段）
     '北京',            ← info.city（嵌套字段）
     'ai', 'rag']       ← tags[*].name（数组，每个元素一条）

三点结论：

  1. $ 传进去的是整个 value 的 JSON 字符串（json.dumps 后的结果）
  2. 嵌套和数组路径真的被解析了——info.city 拿到 '北京'，tags[*].name 展开成两条文本
  3. 数组元素会各生成一个向量——_vectors 里存成了 tags[*].name.0、tags[*].name.1 两个键

真实语义检索：用嵌入模型#

把 embed 换成真实的嵌入模型就行。课程用的是 CloseAI 平台的 openai:text-embedding-3-large，它的嵌入维度是 3072，所以 dims 必须写 3072：

from langchain.embeddings import init_embeddings


embedding_model = init_embeddings(
    model="openai:text-embedding-3-large",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)


index_config = {"embed": embedding_model, "dims": 3072, "fields": ["$"]}
store = InMemoryStore(index=index_config)

配上之后就能按自然语言搜了：

for item in store.search(("users",), query="数电模电"):
    print(item)
# Item(..., value={'course': '数字电路与模拟电路', ...}, score=0.22494289397943232)
# Item(..., value={'course': '数字电路与模拟电路', ...}, score=0.21367212817763243)
# Item(..., value={'course': '计算机组成原理', ...},     score=0.1253981117634029)
Warning

只给 query 时，返回的是”前缀下全部 item”，不是”最相关的几条”——底层对每个候选算 score，然后按 score 降序排列，默认 limit=10 只是把结果截到 10 条。所以语义检索一定要配合 limit（或 filter）用，否则等于”全表返回、只是排了个序”。

本机用假嵌入函数实测（limit 默认 10、库里只有 3 条）：3 条全被返回，只是顺序按 score 从高到低；改成 limit=2 才只剩前 2 条。

Tip

本机实测索引开关（同一份数据，只改 put 的 index 参数）：

场景_vectors 里有什么
InMemoryStore()（没有 index 配置）+ put(index=["$"])空——没有配置时 index 参数被忽略
InMemoryStore(index=cfg) + put()（默认 None）{'$': [...], 'course': [...]}——按初始化配置建索引
InMemoryStore(index=cfg) + put(index=False)空，但 store.get() 照样取得到数据（只是搜不到）
InMemoryStore(index=cfg) + put(index=["course"])只有 {'course': [...]}——只给指定字段建索引

在 Agent 里访问长期记忆#

我们可以在工具或中间件中访问长期记忆。

在工具中访问#

工具函数多接收一个 runtime: ToolRuntime 参数（框架自动注入），通过它拿到 store 和 state：

from typing import NotRequired


from langchain.agents import AgentState, create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore


store = InMemoryStore()


class CustomState(AgentState):
    user_id: NotRequired[str]                 # ← 自定义状态字段


@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
    """将用户信息保存在长期记忆中


    Args:
        name: 用户名


    Returns:
        str: 保存状态
    """
    runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
    return "saved"


@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
    """从长期记忆中读取用户信息


    Returns:
        str: 用户信息
    """
    item = runtime.store.get(("users",), runtime.state["user_id"])
    return str(item.value) if item else "unknown"


agent = create_agent(
    model=model,
    tools=[save_user_info, get_user_info],
    store=store,                              # ← 把 store 交给 Agent
    state_schema=CustomState,                 # ← 用自定义状态把 user_id 带进来
    system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
)


agent.invoke({"messages": ["我叫韩立"], "user_id": "user_1"})

三个要点：

要点说明
工具加 runtime: ToolRuntime 参数框架自动注入，不用你传；runtime.store / runtime.state 都在里面
创建 Agent 时传 store=store不传的话工具里拿不到 store
用 state_schema 扩展状态自定义 user_id 这类字段，invoke 时一起传进来（这里传的是 {"user_id": "user_1"}）
Tip

实测：模型调用 save_user_info 后，store.get(("users",), "user_1") 里就有 {'name': '韩立'} 了；换个会话（甚至换一次进程，只要 store 是持久化的）再读，数据依然在——这就是”跨会话记忆”。

换成 PostgresStore：只改两行#

把 InMemoryStore() 换成 PostgresStore.from_conn_string(...)，工具代码一个字都不用动：

from typing import NotRequired


from langchain.agents import AgentState, create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.store.postgres import PostgresStore


DB_URI = "postgresql://langchain_user:abcd1234@<你的IP>:5432/langchain_db?sslmode=disable"


class CustomState(AgentState):
    user_id: NotRequired[str]


@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
    """将用户信息保存在长期记忆中


    Args:
        name: 用户名


    Returns:
        str: 保存状态
    """
    runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
    return "saved"


@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
    """从长期记忆中读取用户信息


    Returns:
        str: 用户信息
    """
    item = runtime.store.get(("users",), runtime.state["user_id"])
    return str(item.value) if item else "unknown"


with PostgresStore.from_conn_string(DB_URI) as store:      # ← 1. 换 store
    store.setup()                                          # ← 2. 建表（第一次跑）
    agent = create_agent(
        model=model,
        tools=[save_user_info, get_user_info],
        store=store,
        system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
        state_schema=CustomState,
    )


    agent.invoke({"messages": ["你好，很高兴认识你，我是小花"], "user_id": "user-1"})
    agent.invoke({"messages": ["我是谁"], "user_id": "user-1"})     # ← 新会话也能答上来
Note

PostgresStore 和 PostgresSaver 的用法完全对称：from_conn_string(DB_URL) 上下文管理器 + setup() 初始化表结构。区别只在于——前者管长期记忆（store 表），后者管短期记忆/检查点（checkpoint_* 表）。

查看 PostgreSQL 里到底建了哪些表#

在服务器上连进数据库（psql "postgresql://..."），然后 \dt 看当前 schema 下的所有表：

langgraph_db=> \dt
                     List of relations
 Schema |            Name       | Type  |    Owner
--------+-----------------------+-------+----------------
 public | checkpoint_blobs      | table | langgraph_user
 public | checkpoint_migrations | table | langgraph_user
 public | checkpoint_writes     | table | langgraph_user
 public | checkpoints           | table | langgraph_user
 public | store                 | table | langgraph_user
 public | store_migrations      | table | langgraph_user
(6 rows)

六张表，正好对应两个持久化器：

表谁建的存什么
checkpointsPostgresSaver.setup()主表：每个 thread 在某个时刻的 checkpoint 快照
checkpoint_blobs同上不适合直接内联进 checkpoints.checkpoint 的复杂 channel 值
checkpoint_writes同上中间写入 / pending writes，不是最终完整 checkpoint
checkpoint_migrations同上迁移版本表（不是业务数据）
storePostgresStore.setup()长期记忆的 key-value 本体
store_migrations同上长期记忆的迁移版本表

也就是说：只用了 PostgresSaver 就只多前 4 张表；再加 PostgresStore 才多出 store 和 store_migrations 两张。课程在”2.2.3 查看持久化数据”里只看到 4 张表，就是因为那一节只用了 checkpointer。

顺带记几个 psql 常用命令（课程演示过）：

命令作用
\l列出所有数据库
\dn列出所有 schema
select current_schema();查看当前所处的 schema
\dt列出当前 schema 下的所有表

PostgreSQL 的存储结构是 Database → Schema → Table（数据库 → 分区 → 表）三层。

在中间件中访问#

中间件里通过 runtime.store 读写长期记忆。典型用法是在 before_model 里把用户偏好注入系统提示：

from langchain.agents.middleware import before_model


@before_model
def load_user_preference(state, runtime):
    item = runtime.store.get(("users",), state.get("user_id", ""))
    if item:
        print("已知用户偏好:", item.value)
    return None

三条访问路径：runtime / request.runtime#

不同风格的钩子拿到 runtime 的方式不一样，三条路径要分清：

钩子风格具体钩子访问方式
Node-stylebefore_agent / before_model / after_model / after_agentruntime.store（runtime 是第二个参数）
Wrap-stylewrap_model_callrequest.runtime.store
Wrap-stylewrap_tool_callrequest.runtime.store

Node-style 钩子的签名，runtime 直接作为参数传进来：

def before_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:

Runtime 的定义（源码，本机核对字段为 context / store / stream_writer / previous）：

@dataclass(**_DC_KWARGS)
class Runtime(Generic[ContextT]):
    context: ContextT = field(default=None)     # type: ignore[assignment]
    """Static context for the graph run, like `user_id`, `db_conn`, etc.


    Can also be thought of as 'run dependencies'."""


    store: BaseStore | None = field(default=None)
    """Store for the graph run, enabling persistence and memory."""


    stream_writer: StreamWriter = field(default=_no_op_stream_writer)
    """Function that writes to the custom stream."""


    previous: Any = field(default=None)
    """The previous return value for the given thread.
    Only available with the functional API when a checkpointer is provided.
    """
    ...
字段用途
context静态运行时上下文（启动时传入、运行中不变），如 user_id、db_conn
store长期记忆（跨会话持久化）
stream_writer写自定义流（第 17 篇的 custom 流式模式）
previous同一 thread 上一次的返回值（只在 functional API + 有 checkpointer 时可用）

Wrap-style 的 wrap_model_call，runtime 挂在 request 上：

def wrap_model_call(
    self,
    request: ModelRequest[ContextT],
    handler: Callable[[ModelRequest[ContextT]], ModelResponse[ResponseT]],
) -> ModelResponse[ResponseT] | AIMessage | ExtendedModelResponse[ResponseT]:

ModelRequest 的定义：

@dataclass(init=False)
class ModelRequest(Generic[ContextT]):
    """Model request information for the agent."""


    model: BaseChatModel
    messages: list[AnyMessage]                  # excluding system message
    system_message: SystemMessage | None
    tool_choice: Any | None
    tools: list[BaseTool | dict[str, Any]]
    response_format: ResponseFormat[Any] | None
    state: AgentState[Any]
    runtime: Runtime[ContextT]                  # ← store 在这里面
    model_settings: dict[str, Any] = field(default_factory=dict)
    ...

所以 request.runtime.store 就是长期记忆入口。

wrap_tool_call 同理，但 request 换成了 ToolCallRequest：

def wrap_tool_call(
    self,
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
) -> ToolMessage | Command[Any]:
@dataclass
class ToolCallRequest:
    """Tool execution request passed to tool call interceptors."""


    tool_call: ToolCall                       # 模型产出的 tool call：name / args / id
    tool: BaseTool | None                     # 要调用的工具实例
    state: Any                                # Agent 状态
    runtime: ToolRuntime                      # ← 注意：这里是 ToolRuntime，不是 Runtime
    ...
Important

两个 Runtime 不要搞混：中间件的 Node-style 钩子和 wrap_model_call 拿到的是 Runtime（langgraph.runtime）；而 wrap_tool_call 的 request.runtime 是 ToolRuntime——它是专门给工具用的、字段更多：

@dataclass
class ToolRuntime(_DirectlyInjectedToolArg, Generic[ContextT, StateT]):
    state: StateT
    context: ContextT
    config: RunnableConfig
    stream_writer: StreamWriter
    tool_call_id: str | None
    store: BaseStore | None
RuntimeToolRuntime
出现位置中间件的 Node-style 钩子、ModelRequest.runtime工具函数参数、ToolCallRequest.runtime
独有字段previous（上次返回值）config、tool_call_id
共有字段context、store、stream_writer同左
Tip

本机实测（ToolCallRequest / ToolRuntime 的字段用 __dataclass_fields__ 打印核对）：ToolCallRequest 是 tool_call / tool / state / runtime；ToolRuntime 是 state / context / config / stream_writer / tool_call_id / store；Runtime 是 context / store / stream_writer / previous——和课程源码一致。ToolRuntime 的泛型默认值是 ContextT = TypeVar("ContextT", default=None)、StateT = TypeVar("StateT", default=dict)。

何时写入记忆#

官方介绍两种时机：

方式做法优点缺点
主流程写（hot path）用户发消息，AI 一边回答一边决定要不要记立即生效、下一轮就能用、用户可感知（透明）增加延迟、逻辑变复杂
后台写（background）先回答用户，记忆整理放到后台异步做主流程更快、记忆逻辑独立、适合批量整理不能立刻生效、要决定整理频率、触发时机不好选

工程上通常这么选：

  • 用户偏好、账号资料 → 适合热路径写
  • 对话摘要、经验沉淀、行为分析 → 更适合后台写

附：静态运行时上下文（课后阅读）#

静态运行时上下文（Static runtime Context） 表示不可变的数据，比如用户元数据、工具、数据库连接对象。通常在运行开始时通过 invoke / stream 的 context 参数传递，运行期间不会更改。

用法要点：

位置访问方式
Node-style 钩子中通过 runtime.context 访问
定义方式用户自定义 ContextSchema 用 @dataclass 修饰
传入agent.invoke(..., context=UserContext(user_id="user_1"))

一个典型场景：创建 Agent 时声明 context_schema=UserContext，然后在钩子里用 runtime.context.user_id 去长期记忆里查用户额度，额度用尽就中断流程。

记住区别就行：context 是”启动时传入、运行中不变”的参数，适合放数据库连接、用户身份这类东西；state 是运行中不断变化的状态；store 是跨会话的持久记忆。

定义 ContextSchema：@dataclass + context_schema#

三步走：@dataclass 定义 → 创建 Agent 时用 context_schema= 声明 → invoke 时用 context= 传入。

from dataclasses import dataclass


from langchain.agents import create_agent


@dataclass
class UserContext:
    username: str


agent = create_agent(
    model=model,
    store=store,
    context_schema=UserContext,     # ← 声明用哪个 schema
)


agent.invoke(
    {"messages": ["你好啊，你知道 Ada Lovelace 的贡献吗？"]},
    context=UserContext(username="Ada Lovelace"),   # ← 每次调用传入具体值
)
Warning

课程 PDF 正文 P60 把参数名误写成了 state_schema（原文：“用户自定义 ContextSchema 用 @dataclass 修饰，在Agent创建时通过 state_schema 参数传递”）。正确写法是 context_schema——PDF 里同一节的示例代码用的就是 context_schema=UserContext，state_schema 是另一回事（它用于扩展 state，比如上一篇的 CustomState(user_id)）。两个参数在 create_agent 里同时存在、互不替代：

create_agent(..., state_schema=CustomState, context_schema=UserContext)   # 可以一起用

4.1.1 Node-style hooks 中访问 context#

Node-style 的四个钩子都通过 runtime.context 访问上下文对象。课程的例子是额度校验：从长期记忆里查用户额度，额度用尽就 jump_to="end" 中断流程。

from dataclasses import dataclass
from typing import Any


from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langchain_core.messages import AIMessage
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore


store = InMemoryStore()
namespace = ("users", "credits")
store.put(namespace, "Ada Lovelace", {"tokens_credit_left": 5000, "user_level": 5})
store.put(namespace, "Blackwell",    {"tokens_credit_left": 2999, "user_level": 5})
store.put(namespace, "Ampere",       {"tokens_credit_left": 1000, "user_level": 5})


@dataclass
class UserContext:
    username: str


class CheckCredit(AgentMiddleware):
    @hook_config(can_jump_to=["end"])          # ← 类写法要跳转，必须加这个装饰器
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        store = runtime.store
        context = runtime.context               # ← 静态上下文


        username = context.username
        credit = store.get(namespace, username)


        if not credit:                          # 查无此人
            return {"jump_to": "end", "messages": AIMessage("您尚未注册~")}


        if credit.value["tokens_credit_left"] < 3000:     # 额度不足
            return {"jump_to": "end", "messages": AIMessage(f"{username}额度不足，请充值")}


        return None


    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        store = runtime.store
        context = runtime.context


        username = context.username
        credit = store.get(namespace, username)


        # 用上一条 AIMessage 的 usage_metadata 算本次消耗
        usage_metadata = state["messages"][-1].usage_metadata
        token_usage = usage_metadata["input_tokens"] + usage_metadata["output_tokens"] * 6
        credit.value["tokens_credit_left"] -= token_usage


        # 更新长期记忆
        store.put(namespace, username, credit.value)
        return None


    def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        store = runtime.store
        context = runtime.context


        username = context.username
        item = store.get(namespace, username)
        print(f"{username} 当前剩余额度：{item.value['tokens_credit_left']}")
        return None


agent = create_agent(
    model=model,
    middleware=[CheckCredit()],
    store=store,
    context_schema=UserContext,
)

同一个 Agent，invoke 时换 context 就换身份——额度够的正常回答，不够的直接被拦下（课程输出）：

Ada Lovelace 当前剩余额度：2721
Blackwell 当前剩余额度：2999
============================== -> Ada Lovelace messages <-
你好啊，你知道 Ada Lovelace 的贡献吗？      ← 正常回答（很长的介绍）
============================== -> Blackwell messages <-
你好啊，你知道 Blackwell 的贡献吗？
Blackwell额度不足，请充值                    ← 模型根本没被调用
钩子干了什么
before_model查额度：没注册 / 额度 < 3000 → jump_to="end" + 一条兜底 AIMessage，模型不会被调用
after_model从 usage_metadata 算出 token 消耗（输入 + 输出 × 6，输出按 6 倍计价），扣减后写回 store
after_agent整个 Agent 跑完后，从 store 读最新额度打印出来
Tip

本机假服务端实测（额度 5000 的用户跑一轮，假服务端返回 usage: {prompt_tokens:1, completion_tokens:1}）：

[after_agent] Ada Lovelace 当前剩余额度：4993        ← 5000 - (1 + 1×6) = 4993
store 里现在是: {'tokens_credit_left': 4993, 'user_level': 5}
[after_agent] Ampere 当前剩余额度：1000
Ampere -> 最后一条: Ampere额度不足，请充值           ← 被 before_model 拦下，额度一分没扣

三点确认：① 扣费公式真的按 输入 + 输出 × 6 算（1 + 6 = 7，5000 → 4993）；② 被拦截的用户额度不变（before_model 直接跳 end，after_model 没机会跑）；③ 写回的是整个 credit.value，所以 user_level 等其他字段一起被保留下来。

4.1.2 Wrap-style hooks 中访问 context#

Node-style 钩子用 runtime.context，Wrap-style 钩子用 request.runtime.context。

wrap_model_call：按用户身份裁剪工具集#

本例通过 context 里的用户身份，动态决定这次调用暴露给模型哪些工具——额度/权限存在长期记忆里，用 request.runtime.store 读出来：

from dataclasses import dataclass
from typing import Callable


from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain.tools import tool
from langgraph.store.memory import InMemoryStore


store = InMemoryStore()
namespace = ("users", "credits")
store.put(namespace, "Ada Lovelace", {"tokens_left": 3000, "get_weather": "yes", "get_news": "no"})
store.put(namespace, "Blackwell",    {"tokens_left": 2000, "get_weather": "no",  "get_news": "yes"})


@tool
def get_weather(city: str):
    """查询指定城市当日天气


    Args:
        city: 城市名称
    """
    return f"{city} 今天天气不错"


@tool()
def get_news():
    """查询当日新闻"""
    return "美伊尚未达成停战协议"


@dataclass
class UserContext:
    username: str


class CheckCredit(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        context = request.runtime.context            # ← 静态上下文
        store = request.runtime.store                # ← 长期记忆


        value = store.get(namespace, context.username).value
        tools = []
        for tool in request.tools:
            tool_name = tool.name
            if value[tool_name] == "yes":
                tools.append(tool)
            else:
                print(f"{context.username} 无权调用 {tool_name}")


        # 更改调用请求携带的工具集，仅本次生效（transient request update）
        request = request.override(tools=tools)
        return handler(request)


agent = create_agent(
    model=model,
    middleware=[CheckCredit()],
    tools=[get_weather, get_news],
    store=store,
    context_schema=UserContext,
)

三个用户拿到的是三套不同的工具集（课程输出）：Ada Lovelace 被拒 get_news、Blackwell 被拒 get_weather、Ampere 两个都有——同一份代码，靠 context 区分。

Tip

本机假服务端实测（CheckCredit + 按工具名自动回 tool_call 的假服务端）：

===== Ada Lovelace =====
   [wrap_model_call] Ada Lovelace 无权调用 get_news
   [wrap_tool_call] Ada Lovelace 调用 get_weather 花费 10，剩余额度 2990
   store: {'tokens_left': 2990, 'get_weather': 'yes', 'get_news': 'no'}
===== Blackwell =====
   [wrap_model_call] Blackwell 无权调用 get_weather
   最终回答: 我没有可用的工具

坐实两点：① request.override(tools=...) 真的改掉了发往模型的工具列表——假服务端收到的 tools 里只剩被放行的那个；② 改动是”仅本次”的，下一轮 request.tools 又是完整的全量工具，不会累积。

wrap_tool_call：额度校验 + 写回长期记忆#

在 wrap_model_call 的基础上再加一道工具级闸门：额度不够就直接拒绝（连 handler 都不调），够就先扣费、写回 store，再放行。

from typing import Any, Callable


from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse, ToolCallRequest
from langchain_core.messages import ToolMessage
from langgraph.types import Command


# 每个工具的"话费"
CREDIT_MAP = {"get_weather": 10, "get_news": 20}


class ToolGuard(AgentMiddleware):
    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
    ) -> ToolMessage | Command[Any]:
        store = request.runtime.store
        context = request.runtime.context


        username = context.username
        value = store.get(namespace, username).value
        tool_name = request.tool.name


        credits_cost = CREDIT_MAP.get(tool_name)     # 本次调用花费
        credits_left = value["tokens_left"]          # 剩余额度


        if credits_left < credits_cost:
            print(f"{username} 额度不足，{tool_name} 调用失败！")
            # 直接返回一条 ToolMessage，不调用 handler —— 工具根本没执行
            return Command(
                update={
                    "messages": [
                        ToolMessage(
                            content="额度不足，无法调用工具，请充值~",
                            tool_call_id=request.runtime.tool_call_id,   # ← 必须回填，配对 tool_call
                        )
                    ]
                }
            )


        # 更新剩余额度并写回长期记忆
        credits_left -= credits_cost
        value["tokens_left"] = credits_left
        store.put(namespace, username, value)


        return handler(request)

两个细节值得记：

细节说明
拒绝时返回 Command(update={"messages": [ToolMessage(...)]})不调 handler(request) 工具就不会执行；但必须给一条 ToolMessage，否则模型那边”工具调用没有回音”，流程会卡住
tool_call_id=request.runtime.tool_call_id工具结果要和模型发出的 tool_call 配对，这个 id 不能少
Tip

本机假服务端实测（CheckCredit + ToolGuard 串联，Ada Lovelace 额度 3000、get_weather 花费 10）：

[wrap_model_call] Ada Lovelace 无权调用 get_news
[wrap_tool_call]  Ada Lovelace 调用 get_weather 花费 10，剩余额度 2990
store: {'tokens_left': 2990, 'get_weather': 'yes', 'get_news': 'no'}

额度真的被扣掉并写回了 store（3000 → 2990），而且这是在没有真实模型的情况下跑通的——说明这一整套”鉴权 + 计费”逻辑完全跑在中间件里，和模型无关。

4.1.3 便捷中间件 @dynamic_prompt：动态系统提示词#

@dynamic_prompt 的底层实现和通用钩子函数是统一的，只是给”动态提示词”这个高频需求开了个更便捷的入口。

本例：按 context 里的用户名 + store 里的 chat_preferences，动态生成系统提示词。

from dataclasses import dataclass


from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langgraph.store.memory import InMemoryStore


store = InMemoryStore()
namespace = ("users", "preferences")
store.put(namespace, "Ada Lovelace", {"chat_preferences": ["不喜欢啰嗦", "尽可能用最精简的文字解释清楚"]})
store.put(namespace, "Blackwell",    {"chat_preferences": ["喜欢长篇大论、引经据典"]})


@dataclass
class UserContext:
    username: str


@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
    username = request.runtime.context.username
    preferences = request.runtime.store.get(namespace, username).value["chat_preferences"]
    custom_prompt = "# 用户偏好\n" + "\n".join(preferences)
    print(f"{username} 自定义系统提示词\n{custom_prompt}")
    return custom_prompt            # ← 返回的就是这次的 system prompt


agent = create_agent(
    model=model,
    middleware=[personalized_prompt],
    store=store,
    context_schema=UserContext,
)

效果对比（同一个问题”为什么花儿这样红？”）：

用户动态生成出的系统提示词模型回答风格
Ada Lovelace# 用户偏好\n不喜欢啰嗦\n尽可能用最精简的文字解释清楚一句话：“花儿红的主要原因是花瓣细胞中的花青素在酸性环境下显红色……”
Blackwell# 用户偏好\n喜欢长篇大论、引经据典长篇大论：分”科学视角 / 文学艺术 / 文化差异 / 现代发现”四大段
Important

@dynamic_prompt 只改”本次调用前的系统提示词”，不写进消息列表。所以模型输出能看出提示词生效了，但翻 response["messages"] 是看不到这条提示词的。

本机假服务端实测（抓发往模型的原始请求 + 返回的 state）：

=== 发给模型的 system 消息 ===
  role=system: # 用户偏好
不喜欢啰嗦
尽可能用最精简的文字解释清楚
  role=user: 为什么花儿这样红？


=== 返回 state 里的消息 ===
  HumanMessage: 为什么花儿这样红？
  AIMessage: 好的。

请求里有 system 消息，state["messages"] 里没有——这正是它和”往消息列表里插 SystemMessage”的本质区别。顺带一提，这也解释了为什么它能每次调用重新算：如果写进消息列表，第二次调用就会把上次的提示词一起带上，越滚越多。

4.2 工具中访问 context#

工具里拿静态上下文靠 runtime.context。上面用的是 runtime.state["user_id"]（把用户 id 放 state 里），这里换成放 context 里——顺便演示工具参数用 pydantic 模型的写法。

先说一个前提：参数类型注解是 pydantic 模型时，LangChain 会自动把它解析进工具的 schema。用 convert_to_openai_tool 一看便知：

from typing import List


from pydantic import BaseModel, Field
from langchain.tools import tool
from langgraph.prebuilt.tool_node import ToolRuntime
from langchain_core.utils.function_calling import convert_to_openai_tool


class UserInfo(BaseModel):
    username: str = Field(description="用户名")
    age: int = Field(description="年龄")
    hobbies: List[str] = Field(description="兴趣爱好")


@tool(parse_docstring=True)
def save_user_info(user_info: UserInfo, runtime: ToolRuntime) -> str:
    """保存用户信息


    Args:
        user_info: 用户信息对象
    """
    return "ok"


print(convert_to_openai_tool(save_user_info))

输出的 JSON Schema 里，user_info 被完整展开成了对象（properties 里是 username / age / hobbies，required 三项全要），而 runtime: ToolRuntime 压根没出现：

{
  "type": "function",
  "function": {
    "name": "save_user_info",
    "description": "保存用户信息",
    "parameters": {
      "properties": {
        "user_info": {
          "properties": {
            "username": { "description": "用户名", "type": "string" },
            "age":      { "description": "年龄",   "type": "integer" },
            "hobbies":  { "description": "兴趣爱好", "items": {"type": "string"}, "type": "array" }
          },
          "required": ["username", "age", "hobbies"],
          "type": "object",
          "description": "用户信息对象"
        }
      },
      "required": ["user_info"],
      "type": "object"
    }
  }
}
Tip

本机实测：convert_to_openai_tool(save_user_info) 的输出与课程逐字一致（user_info 被展开、runtime 被隐藏）。

完整案例：读写长期记忆里的用户信息，用户 id 从 context 拿，工具参数用 pydantic 模型：

from dataclasses import dataclass
from typing import Any, List


from pydantic import BaseModel, Field


from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt.tool_node import ToolRuntime
from langgraph.store.memory import InMemoryStore


@dataclass
class UserContext:
    user_id: str


store = InMemoryStore()
namespace = ("users", "user_info")


class UserInfo(BaseModel):
    username: str = Field(description="用户名", default="unknown")
    age: int = Field(description="年龄", default=0)
    # 这里相当于告诉模型，每次创建新实例时，调用 list() 创建一个新的 list 传进去
    hobbies: List[str] = Field(description="兴趣爱好", default_factory=list)
    # 这种写法是用可变对象做默认值，会有多个对象共享一个列表对象的风险
    # hobbies: List[str] = Field(description="兴趣爱好", default=[])


@tool(parse_docstring=True)
def read_user_info(runtime: ToolRuntime[UserContext, Any]) -> UserInfo | str:
    """读取用户信息


    Returns:
        UserInfo | str: 如果找到用户信息，返回 UserInfo 对象；否则返回空字符串。
    """
    user_id = runtime.context.user_id            # ← 从 context 拿用户 id
    item = runtime.store.get(namespace, user_id)
    return UserInfo(**item.value) if item else ''


@tool(parse_docstring=True)
def write_user_info(user_info: UserInfo, runtime: ToolRuntime[UserContext, Any]) -> bool:
    """将用户信息写入长期记忆


    Args:
        user_info：用户信息对象


    Returns:
        bool: 如果成功写入则返回 True，否则返回 False
    """
    user_id = runtime.context.user_id
    runtime.store.put(namespace, user_id, user_info.model_dump())
    return True


agent = create_agent(
    model=model,
    tools=[read_user_info, write_user_info],
    store=store,
    checkpointer=InMemorySaver(),
    context_schema=UserContext,
)

跑起来是这样（课程输出）：

写入记忆, namespace: ('users', 'user_info'), user_id: user_1, user_info: {'username': '韩立', 'age': 0, 'hobbies': ['修仙']}
写入记忆, namespace: ('users', 'user_info'), user_id: user_1, user_info: {'username': '韩立', 'age': 200, 'hobbies': ['修仙', '跑步']}

注意第二次写入的 hobbies 是 ['修仙', '跑步']——模型先 read_user_info 读回旧值，合并后再 write_user_info，所以旧爱好没丢。这就是 system prompt 里那句”在记录之前先查找历史信息，和新增信息合并后记录”的作用。

最后换 thread_2（全新的 thread_id）问”你还记得我吗？“，模型照样能答上来——因为用户信息在共享的 store 里，不在某个 thread 的 messages 里。

课程在这段总结了五条：

#结论
1前两次 invoke 通过相同的 thread_id 串联为一个会话，共享短期记忆
2context 是运行时静态配置，每次 invoke 互相独立
3长期记忆可以在任意位置访问，全局共享同一份信息
4工具参数里的 ToolRuntime 是 LangChain / LangGraph 运行时注入的，在工具 schema 里没有体现
5ToolRuntime 的默认泛型是 [None, Any]（第一个是 Context 的泛型，第二个是 State 的泛型）
Warning

ToolRuntime 不写泛型会踩坑。它的泛型默认值是 ContextT = TypeVar("ContextT", default=None)——不显式指定上下文类型，框架就认为”这个工具的 context 是 None”。这时 invoke(..., context=UserContext(...)) 会触发警告：

UserWarning | Pydantic serializer warnings:
  PydanticSerializationUnexpectedValue(Expected `none` - serialized value may not be as expected
    [field_name='context', input_value=UserContext(user_id='user_1'), input_type=UserContext])

所以课程把 UserContext 作为 ToolRuntime 的第一个泛型写出来：runtime: ToolRuntime[UserContext, Any]。

本机实测：写成 ToolRuntime[UserContext, Any] 后警告消失；写成裸 ToolRuntime 则复现上面这条 UserWarning（值其实照样注入进去了——runtime.context 拿得到 UserContext(user_id='user_1')——只是类型上不匹配，会报序列化警告）。

相关#

练习题#

一、回忆填空（写完再展开对答案）#

  1. 短期记忆记录的是____级数据、会话间不共享；长期记忆记录的是____或应用级数据，任何会话都能访问
  2. 长期记忆三类：____记忆（事实）、____记忆（经验，常表现为 few-shot）、____记忆（规则/做事方法）
  3. 四层存储架构：____ → ____ → ____ → ____
  4. namespace 的数据类型是____，key 是____，value 是____
  5. 三个基础 API：____ 写入、____ 读取、____ 检索；它们可以在 Agent 流程____直接调用
  6. get() 返回的是 ____ 对象，取值要写 item.____（不是直接拿字典）
  7. search() 支持两种检索：按 ____ 做结构化过滤（不需要向量）、按 ____ 做语义检索（需要向量）
  8. 在工具里访问长期记忆，工具函数要多接收一个 ____ 参数，通过 runtime.____ 拿到 store、runtime.state 拿状态
  9. 创建 Agent 时要把 store 通过 ____ 参数传进去；要用自定义字段就得通过 ____ 扩展状态
  10. 写入记忆两种时机：（立即生效但增加延迟）和（主流程更快但不能立刻生效）
填空答案（做完再点开）
  1. 会话（线程） / 用户特定　2. 语义（Semantic） / 情景（Episodic） / 程序性（Procedural）　3. store / namespace / key / value　4. 字符串元组（tuple[str, …]） / 字符串 / 字典（dict[str, Any]）　5. put() / get() / search() / 之外　6. Item / value　7. filter / query　8. ToolRuntime / store　9. store / state_schema　10. 主流程写（hot path） / 后台写（background）

二、裸写题#

  • 2-1 直接操作 store 用 InMemoryStore() 存两条记忆（不同 key，同一个 ("users", "alice", "memories") 命名空间），然后： ① 用 get() 取其中一条，打印 item.value； ② 用 search() 按前缀把两条都查出来； ③ 用 search(..., filter={...}) 只筛出其中一条。

    提示（先自己想，实在想不出再点开）

    一级 · 思路：store 就是一个”带路径的字典” 二级 · 方法：store.put(namespace, key, value) / store.get(namespace, key) / store.search(namespace_prefix, filter=...) 三级 · 骨架：注意 get 返回 Item 对象，要 .value；search 返回的是列表

  • 2-2 让 Agent 自己记住用户信息 用 save_user_info / get_user_info 两个工具（都带 runtime: ToolRuntime）+ CustomState(user_id) 创建 Agent，先让它记住”我叫韩立”，再用新的 thread_id问”我叫什么？“，观察长期记忆是否跨会话生效。

    提示

    一级 · 思路：跨会话 = 记忆不在 thread 的 messages 里，而在共享的 store 里 二级 · 方法：创建 Agent 时传 store=store、state_schema=CustomState 三级 · 骨架：invoke 时要带上 "user_id": "user_1"，否则工具里取不到

  • 2-3 把用户偏好写进系统提示（中间件） 写一个 @before_model 钩子：从 store 里读某个用户的偏好（比如”喜欢简短回答”），打印出来。先手动用 store.put 塞一条偏好，再跑一次 Agent 观察钩子的输出。

    提示

    一级 · 思路：这是”读取记忆并影响行为”的最小示例 二级 · 方法：runtime.store.get(namespace, key) 在钩子里也能用 三级 · 骨架：更进阶的做法是把读到的偏好拼进 system_prompt——但那需要动态提示词（第 15 篇提到的”动态设置”）

三、综合题#

  • 3-1 给客服助手装上”跨会话记忆” 在上一篇的客服助手基础上加长期记忆：

    1. 自定义 CustomState 加一个 user_id 字段
    2. 写 save_order_no(order_no, runtime) 工具：把订单号写进 store 的 ("users", user_id, "orders")
    3. 写 get_order_no(runtime) 工具：从 store 读回来
    4. **新建一个会话（新 thread_id）**问”我的订单号是多少？“，确认它依然能答上来
    提示（先自己想，实在想不出再点开）

    一级 · 思路：短期记忆会因裁剪/换会话而丢失，长期记忆才是”永久档案” 二级 · 方法：runtime.store.put(("users",), user_id, {...}) + store= 参数 三级 · 骨架：系统提示词里明确”用户告知订单号时务必调用 save_order_no 记录”

参考答案（做完再点开）
import os
from typing import NotRequired


from dotenv import load_dotenv
from langchain.agents import AgentState, create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import ToolRuntime, tool
from langchain_core.messages import HumanMessage
from langgraph.store.memory import InMemoryStore


load_dotenv(override=True)


model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="openai",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
)


# ---------- 2-1 直接操作 store ----------
store = InMemoryStore()
ns = ("users", "alice", "memories")
store.put(ns, "pref_food", {"category": "food", "text": "Alice likes sushi"})
store.put(ns, "pref_lang", {"category": "code", "text": "Alice prefers Python"})


item = store.get(ns, "pref_food")
print("① get:", item.value)
print("② search 前缀:", [i.key for i in store.search(("users",))])
print("③ search + filter:", [i.key for i in store.search(ns, filter={"category": "code"})])


# ---------- 2-2 让 Agent 记住用户信息 ----------
class CustomState(AgentState):
    user_id: NotRequired[str]


@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
    """将用户信息保存在长期记忆中


    Args:
        name: 用户名


    Returns:
        str: 保存状态
    """
    runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
    return "saved"


@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
    """从长期记忆中读取用户信息


    Returns:
        str: 用户信息
    """
    item = runtime.store.get(("users",), runtime.state["user_id"])
    return str(item.value) if item else "unknown"


agent = create_agent(
    model=model,
    tools=[save_user_info, get_user_info],
    store=store,
    state_schema=CustomState,
    system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
)


agent.invoke({"messages": ["我叫韩立"], "user_id": "user_1"})
print("store 内容:", store.get(("users",), "user_1").value)


# 换一个新会话（新 thread_id）再问
config2 = {"configurable": {"thread_id": "brand-new"}}
r = agent.invoke({"messages": ["我叫什么？"], "user_id": "user_1"}, config=config2)
print("新会话回答:", r["messages"][-1].content)


# ---------- 2-3 中间件里读偏好 ----------
from langchain.agents.middleware import before_model


store.put(("users", "alice", "prefs"), "style", {"text": "喜欢简短回答"})


@before_model
def load_preference(state, runtime):
    item = runtime.store.get(("users",), "alice")     # 也可以按你的 namespace 约定读
    print("   [中间件] 读到偏好:", item.value if item else "无")
    return None


agent_pref = create_agent(
    model=model,
    store=store,
    state_schema=CustomState,
    middleware=[load_preference],
)
agent_pref.invoke({"messages": ["你好"], "user_id": "alice"})


# ---------- 3-1 跨会话记住订单号 ----------
@tool(parse_docstring=True)
def save_order_no(order_no: str, runtime: ToolRuntime) -> str:
    """把用户订单号写入长期记忆


    Args:
        order_no: 订单号
    """
    runtime.store.put(("orders",), runtime.state["user_id"], {"order_no": order_no})
    return f"已记录订单号 {order_no}"


@tool(parse_docstring=True)
def get_order_no(runtime: ToolRuntime) -> str:
    """读取用户订单号


    Returns:
        str: 订单号
    """
    item = runtime.store.get(("orders",), runtime.state["user_id"])
    return str(item.value) if item else "没有记录"


agent_cs = create_agent(
    model=model,
    tools=[save_order_no, get_order_no],
    store=store,
    state_schema=CustomState,
    system_prompt=(
        "你是客服助手。用户告诉你的订单号、个人信息等重要事实，必须调用工具记录下来；"
        "被问到时先用工具查询，再据实回答。"
    ),
)


s1 = {"configurable": {"thread_id": "cs-1"}}
agent_cs.invoke({"messages": [HumanMessage("我的订单号是 A12345，帮我记一下")], "user_id": "user_1"}, config=s1)


# 关键：换一个全新的会话，看它还记不记得
s2 = {"configurable": {"thread_id": "cs-2"}}
r = agent_cs.invoke({"messages": [HumanMessage("我的订单号是多少？")], "user_id": "user_1"}, config=s2)
print("新会话回答:", r["messages"][-1].content)
# 跨会话生效（因为订单号存在共享的 store 里，而不是某个 thread 的消息里）
RAG入门与文档加载切分
短期记忆与上下文管理

相关阅读

1嵌入模型与向量存储实战编程学习/langchain学习笔记2RAG入门与文档加载切分编程学习/langchain学习笔记3短期记忆与上下文管理编程学习/langchain学习笔记4自定义中间件与hook函数编程学习/langchain学习笔记5其它内置中间件与执行顺序编程学习/langchain学习笔记

评论区

[ 标签 ]
# AI37# AI 编程2# AI工具1# Ajax2# Apifox1# AstrBot3# Astro2# CC Switch1# CDN2# Claude Code1# claudecode2# ClaudeCode1# Cloudflare2# CloudFlare2# CloudFlare-ImgBed3# coc3# CSS6# DeepSeek6# deepseek2# DELETE1# Docker1# EdgeOne3# Gist1# git1# GitHub1# hexo-circle-of-friends1# HTML6# HTTP5# ImageManager1# Java23# java13# JavaScript5# JDBC3# JSON2# JUnit1# LangChain25# Logback1# Maven6# Muse Spark1# Mybatis1# MyBatis4# MySQL28# MySql1# NapCat1# Node.js1# obsidian2# Obsidian5# OpenCode4# ORM1# PathVariable1# PicGo1# PyCharm1# Python65# RequestBody1# RequestMapping1# RESTful风格1# skills1# Slf4j1# SpringBoot11# SQL2# Streamlit5# Svelte2# TailwindCSS1# Telegram3# Tlias2# Vercel1# vscode2# Vue7# Waline3# WebDAV1# Web基础6# Web开发6# WinSCP1# YAML1# 三层架构1# 中二宣言1# 书籍1# 使用文档10# 写作1# 函数2# 刷步数1# 前端32# 动态1# 动漫1# 包1# 单词2# 博客7# 博客工作流1# 博客开发2# 参数接收1# 友链1# 反思2# 图床6# 地图1# 备份2# 大模型1# 奇思妙想1# 存储1# 学习方法6# 学校1# 宝塔面板3# 宝宝10# 对象1# 导航栏1# 工具2# 开发1# 开发工具1# 开发规范1# 开心1# 异常处理1# 影视2# 微信1# 性能优化2# 总结1# 想法15# 感受1# 感悟11# 指南1# 提示词工程1# 插件5# 故障排除1# 效率工具2# 教程10# 数据分析9# 数据库27# 数据结构1# 文件操作2# 斩神1# 日常92# 日志框架1# 朋友圈1# 朱元璋1# 模块1# 模板1# 正则表达式2# 测试1# 游戏2# 爬虫7# 生活迁移1# 电影2# 电脑1# 碎碎念1# 视觉识别1# 类1# 类型注解1# 网络基础2# 网络教室1# 羊毛2# 脚本2# 脚本工具1# 自动化2# 蓝奏云1# 订阅推荐2# 记录2# 评论系统1# 词根1# 词缀1# 说说1# 足迹1# 跑步2# 路径参数1# 转载2# 运动1# 部落冲突1# 配置1# 随机图1# 面向对象5# 音乐3# 音标1# 饮食1# 驼峰命名1# 高德地图1
[ 公告 ]

如果你喜欢，那么欢迎来到我的世界！

了解更多
[ 音乐 ]
封面

音乐

暂未播放

0:000:00
暂无歌词
找不到相关结果。
[ contents ]
[ 全部文章 ]