LangChain 实体记忆实践：从 ConversationEntityMemory 到长期记忆 Store

实体记忆关注的不是“最近聊了什么”，而是“对话里出现了哪些重要对象，以及这些对象有哪些稳定事实”。

比如一段聊天里出现了这些信息：

• 用户叫 Alice。
• Alice 正在做 Flask 项目。
• 项目使用 PostgreSQL。
• Alice 更喜欢生产实践版本的回答。

普通缓冲记忆会把这些内容留在最近几轮消息里；摘要记忆会把它们压进一段摘要；实体记忆则会尝试把 Alice、Flask 项目、PostgreSQL 这些实体抽出来，并给每个实体维护一份事实描述。

它的目标是让系统在后续对话里可以按实体召回信息，而不是每次都扫描完整聊天历史。

实体记忆到底解决什么

实体记忆可以理解成一张不断更新的实体事实表：

实体	类型	事实
Alice	user	正在做 Flask 项目，偏好生产实践回答
Flask 项目	project	使用 PostgreSQL，需要数据库迁移方案
PostgreSQL	database	当前项目数据库

模型下一轮回答时，如果用户问：

那上线前还要检查什么？

系统可以召回 Alice 和 Flask 项目 相关事实，让模型知道“那”指的是 Flask 项目的上线检查，而不是从零理解。

所以实体记忆和聊天历史的区别是：

• 聊天历史保存“原始发生过什么”。
• 摘要记忆保存“这段对话大概在讲什么”。
• 实体记忆保存“某个对象目前有哪些事实”。

旧版 ConversationEntityMemory 的思路

早期 LangChain 提供过 ConversationEntityMemory。它的基本流程是：

读取最近对话
  -> 用 LLM 抽取实体
  -> 查询已有实体描述
  -> 用 LLM 更新每个实体的摘要
  -> 下一轮 Prompt 注入相关实体信息

它内部通常会有一个 entity_store，用于保存实体名到实体描述的映射。

旧式代码大概是这样：

from langchain_classic.chains.conversation.base import ConversationChain
from langchain_classic.memory import ConversationEntityMemory
from langchain_classic.memory.prompt import ENTITY_MEMORY_CONVERSATION_TEMPLATE
from langchain_openai import ChatOpenAI


llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)

chain = ConversationChain(
    llm=llm,
    prompt=ENTITY_MEMORY_CONVERSATION_TEMPLATE,
    memory=ConversationEntityMemory(llm=llm),
)

chain.invoke({"input": "你好，我叫 Alice。我正在掌握 LangChain。"})
chain.invoke({"input": "我最喜欢的编程语言是 Python。"})
chain.invoke({"input": "我住在广州。"})

print(chain.memory.entity_store.store)

运行后，实体存储里可能出现类似结构：

{
    "Alice": "Alice 正在掌握 LangChain，喜欢 Python，住在广州。",
    "LangChain": "LangChain 是 Alice 正在掌握的框架。",
    "Python": "Python 是 Alice 最喜欢的编程语言。",
    "广州": "广州是 Alice 居住的城市。",
}

这个设计很直观：把聊天里的实体变成可查询的键值对。

它的问题在哪里

ConversationEntityMemory 最大的问题不是“不能用”，而是它把太多生产决策藏在一个自动组件里。

第一，实体抽取不稳定。

模型可能把临时词、地点、技术名词、产品名、用户姓名都抽出来。哪些该记，哪些不该记，必须由业务定义，不能完全交给通用 prompt。

第二，实体名很难归一化。

同一个人可能被叫作 Alice、alice、小爱、我。同一个项目可能叫 Flask 项目、后端服务、LLMOps API。如果没有实体 ID 和归一化规则，store 很快会变脏。

第三，事实更新需要冲突处理。

用户先说“我用 MySQL”，后来改成“现在迁到 PostgreSQL”。实体记忆不能简单追加，它要知道新事实覆盖旧事实，还要保留变更时间和来源。

第四，隐私风险更高。

实体记忆天然倾向于保存人名、地址、公司、偏好、账号信息。这些内容比普通聊天摘要更敏感，必须有脱敏、授权、删除和审计机制。

第五，召回可能污染回答。

实体事实一旦错误，后续每次召回都会影响模型。相比短期消息，实体记忆更像长期事实库，错误成本更高。

所以实体记忆在生产里不能只是一个“自动抽取并保存”的组件，它必须是一套可治理的数据管道。

生产是否会这么用

结论：不会直接照搬 ConversationEntityMemory 这种方式，但会使用实体记忆这个思想。

生产系统确实需要记住用户、项目、组织、订单、设备、代码仓库、工单这些实体，也确实需要围绕实体保存事实。但实现方式通常会更工程化：

• 实体有明确类型，例如 user、project、tenant、ticket。
• 实体有稳定 ID，而不是只靠自然语言名字。
• 事实有 schema，不是一段随意摘要。
• 写入前有置信度、来源、时间、权限和敏感级别。
• 更新时有覆盖、合并、冲突检测。
• 删除时能按用户、会话、实体、字段精确删除。
• 召回时按当前任务选择，不是把所有实体事实塞进 Prompt。

也就是说，生产会用“实体记忆架构”，不会把旧版实体记忆类当成核心数据库。

现在是否已经抛弃

从当前 LangChain 生态看，ConversationEntityMemory 没有完全消失，但已经不是新项目主线。

它现在属于 langchain-classic 体系，官方参考文档仍然能查到 ConversationEntityMemory，定位是 entity extractor and summarizer memory，也就是旧版 Memory 组件的一部分。

这意味着：

• 老项目可以继续维护。
• 掌握记忆模式时可以参考它的设计。
• 新项目不建议围绕它搭生产架构。

当前 LangChain v1 的主线已经转向：

• 短期记忆：create_agent + checkpointer + thread_id。
• 长期记忆：LangGraph Store，按 namespace 和 key 存 JSON 文档。
• 读写记忆：通过 tools、runtime.store、middleware 或业务管道完成。

所以更准确的说法不是“实体记忆被抛弃了”，而是：旧版 ConversationEntityMemory 被迁到了 classic，实体记忆的实现方式升级为 Store + schema + 业务治理。

最新版怎么实现

最新版更推荐把实体记忆作为长期记忆的一种：用 Store 保存实体事实，用工具或后台管道负责读写。

下面是一个最小实现：用 PostgresStore 保存用户实体，Agent 在工具里通过 runtime.store 读取。

安装：

pip install langgraph-checkpoint-postgres

定义上下文和读取工具：

from dataclasses import dataclass
from typing import Any

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


@dataclass
class UserContext:
    tenant_id: str
    user_id: str


def user_namespace(ctx: UserContext) -> tuple[str, str, str]:
    return ("tenants", ctx.tenant_id, "users")


@tool
def get_user_entity(runtime: ToolRuntime[UserContext]) -> dict[str, Any]:
    """读取当前用户的长期实体记忆。"""
    assert runtime.store is not None

    namespace = user_namespace(runtime.context)
    item = runtime.store.get(namespace, runtime.context.user_id)

    if item is None:
        return {}

    return item.value

创建 Agent：

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"


with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()

    agent = create_agent(
        model="gpt-5.5",
        tools=[get_user_entity],
        store=store,
        context_schema=UserContext,
        system_prompt=(
            "你是一个严谨的技术助手。"
            "回答前可以读取用户实体记忆，但不能把记忆中没有的内容当成事实。"
        ),
    )

    result = agent.invoke(
        {
            "messages": [
                {"role": "user", "content": "继续给我数据库迁移上线建议。"}
            ]
        },
        context=UserContext(tenant_id="tenant-1", user_id="user-42"),
    )

这个实现只解决读取。写入不能随意自动化，生产里最好单独做实体抽取和审核管道。

实体抽取和写入

实体写入可以放在聊天完成后的后台任务里：先用结构化输出抽取候选事实，再做校验，最后写入 Store。

定义实体事实 schema：

from typing import Literal

from pydantic import BaseModel, Field


class EntityFact(BaseModel):
    entity_type: Literal["user", "project", "organization", "preference"]
    entity_id: str
    field: str
    value: str
    confidence: float = Field(ge=0, le=1)
    source_message_id: str

抽取候选事实时，要求模型只抽“未来有用且用户明确表达”的内容：

from langchain_openai import ChatOpenAI


extractor = ChatOpenAI(model="gpt-4.1-mini", temperature=0).with_structured_output(
    list[EntityFact]
)


facts = extractor.invoke(
    [
        {
            "role": "system",
            "content": (
                "从对话中抽取可长期保存的实体事实。"
                "只抽取用户明确表达或业务系统确认的信息。"
                "不要保存寒暄、临时问题、模型猜测和敏感凭据。"
            ),
        },
        {
            "role": "user",
            "content": "我叫 Alice，现在负责 Flask 迁移项目，回答尽量给生产实践。",
        },
    ]
)

写入 Store 前要做业务校验：

ALLOWED_FIELDS = {
    "user": {"name", "role", "preference"},
    "project": {"name", "tech_stack", "database"},
    "preference": {"answer_style"},
}


def should_save(fact: EntityFact) -> bool:
    if fact.confidence < 0.8:
        return False

    allowed = ALLOWED_FIELDS.get(fact.entity_type, set())
    return fact.field in allowed

持久化时，不要把所有事实揉成一句自然语言，最好按字段保存：

from datetime import datetime, timezone


def upsert_entity_fact(store, namespace, fact: EntityFact) -> None:
    item = store.get(namespace, fact.entity_id)
    current = item.value if item else {"facts": {}}

    current["facts"][fact.field] = {
        "value": fact.value,
        "confidence": fact.confidence,
        "source_message_id": fact.source_message_id,
        "updated_at": datetime.now(timezone.utc).isoformat(),
    }

    store.put(namespace, fact.entity_id, current)

这样做比旧版 entity_store.store["Alice"] = "..." 更适合生产，因为每个事实都有字段、来源、置信度和更新时间。

召回策略

实体记忆不应该每次全部塞进 Prompt。

更合理的召回策略是：

• 当前用户相关事实：默认可召回。
• 当前项目相关事实：只有当前任务涉及项目时召回。
• 敏感事实：需要明确授权或业务场景。
• 低置信度事实：不直接注入 Prompt，可以先向用户确认。
• 冲突事实：优先最新确认版本，必要时让用户确认。

如果 Store 开了向量索引，也可以按当前问题搜索相关记忆。官方长期记忆文档里，Store 支持 namespace/key 组织 JSON，也支持带索引后的 search。

items = store.search(
    ("tenants", "tenant-1", "users"),
    query="answer style and project preferences",
)

但即便用了向量搜索，也建议配合 metadata 过滤。实体记忆是事实系统，不应该只靠相似度决定是否召回。

实体记忆和其他记忆的关系

实体记忆不是替代缓冲记忆、摘要记忆，而是补充它们。

记忆类型	保存什么	生命周期
ChatMessageHistory	原始消息	会话内或审计留存
缓冲记忆	最近几轮上下文	当前会话
摘要记忆	长对话压缩摘要	当前会话
实体记忆	用户、项目、组织等结构化事实	跨会话
向量记忆	可相似度检索的文本片段	跨会话或知识库

一个生产聊天系统通常会同时使用几层：

当前请求
  -> 读取 thread state 中的最近消息
  -> 读取当前会话摘要
  -> 召回相关实体事实
  -> 召回 RAG 文档
  -> 调用模型
  -> 保存消息
  -> 后台抽取实体事实
  -> 写入长期 Store

实体记忆负责稳定事实，缓冲和摘要负责当前会话连续性，RAG 负责外部知识。

生产落地检查清单

实体记忆上线前，至少要回答这些问题：

• 哪些实体类型允许保存？
• 每类实体有哪些字段？
• 哪些字段禁止模型自动写入？
• 用户如何查看和删除自己的记忆？
• 旧事实被纠正时如何覆盖？
• 冲突事实是否需要人工或用户确认？
• 敏感信息如何识别和脱敏？
• 记忆写入失败是否影响主链路？
• 召回了哪些实体事实，是否能在 trace 里看到？
• 多租户、多用户、多会话的 namespace 是否隔离？

这些问题比 API 调用本身更重要。实体记忆一旦上线，就不只是上下文优化，而是用户数据系统。

最后回答三个问题

第一，实际生产是否会这么用？

会用实体记忆这个思想，但不会直接用 ConversationEntityMemory 当生产核心。生产更倾向于 schema 化实体、稳定实体 ID、数据库 Store、后台抽取、人工或规则校验、可删除和可审计。

第二，现在是否抛弃了？

旧版 ConversationEntityMemory 没有消失，但已经进入 langchain-classic，不再是 LangChain v1 新项目的主线。它适合掌握原理和维护老代码，不适合新系统从它开始搭。

第三，最新版怎么实现？

最新版建议用 LangGraph Store 做长期记忆：create_agent(..., store=store)，通过 runtime.store 在工具或业务管道中读写实体事实。生产环境使用 PostgresStore、RedisStore、MongoDBStore 这类持久化 Store，并把实体事实设计成 JSON schema，而不是一段不可治理的自然语言摘要。

参考：

• LangChain Long-term memory (opens new window)
• ConversationEntityMemory API (opens new window)