摘要

在 AI 客服里，“记住用户”并不是把全部聊天记录一直塞进 Prompt。更合适的做法是从对话中提取稳定事实，跨会话保存，并在回答前只召回与当前问题有关的部分。

本文完整拆解一条长期记忆链路：用 mem0 管理记忆，适配 DashScope Embedding，将向量记忆与 RAG 知识库分开存入 Milvus，按需使用 Neo4j 保存关系，再用 Rerank 对候选记忆精排。除了配置和代码，文章还会处理并发初始化、非法向量、重复文本映射、后台写入可靠性、失败降级和数据删除等容易遗漏的工程问题。

关键词： mem0、DashScope、Milvus、Neo4j、Rerank、LangGraph、长期记忆、AI 客服

本文对应的主要依赖版本为 mem0ai==1.0.1、dashscope==1.25.6、langchain-core==1.2.6、pymilvus==2.6.6、neo4j==5.28.2。不同版本的配置字段和过滤语法可能变化，升级前应先做回归测试。

环境准备

先安装文中直接使用的依赖：

pip install mem0ai==1.0.1 dashscope==1.25.6 langchain-core==1.2.6 \
  pymilvus==2.6.6 neo4j==5.28.2 aiohttp==3.13.3

运行前还需要可访问的 Milvus；启用图记忆时再准备 Neo4j。LLM、Embedding、Rerank、Milvus 和 Neo4j 的密钥或连接信息都从环境变量读取，不要写进 Python 默认值。

一、先分清三类上下文

一套完整的客服上下文通常包含会话历史、长期记忆和 RAG 知识库。三者都可能进入模型上下文，但职责不同。

数据	解决的问题	生命周期	隔离方式	典型内容
会话历史	保持本轮对话连贯	当前会话	session_id	最近几轮问答
长期记忆	跨会话了解同一客户	跨会话	user_id	单位、职位、产品偏好、采购计划
RAG 知识库	回答产品和业务知识	长期共享	collection / tenant	产品手册、规格参数、FAQ

把三类数据混在一起会带来两个直接问题：一是检索结果互相干扰，二是数据权限很难说明白。因此，mem0 单独使用 mem0_memory collection，产品资料与业务文档则保存在各自的知识库 collection 中。

长期记忆的目标也随之明确：

1. 从对话中提取客户明确表达的事实；
2. 以 user_id 为边界保存和检索；
3. 只把本次回答真正需要的记忆放进 Prompt；
4. 记忆模块不可用时，客服主流程仍能继续。

二、整体调用链

系统入口是 FastAPI，对话由 LangGraph 编排。除 FAQ 快速路径外，一次普通请求会经过下面这条链路：

用户提问
  -> memory_loader：并发检索“当前问题相关记忆”和“用户画像”
  -> router：识别一个或多个意图
  -> product / general / order / after-sales agent
  -> aggregator：生成最终回复
  -> 先向客户端返回结果
  -> memory_saver：后台保存最近 5 轮对话

这里有两个刻意的取舍。

第一，FAQ 已经有稳定答案时，直接走快速路径，不必为每次固定问答增加一次记忆检索。第二，记忆写入排在聚合节点之后。HTTP 和 SSE 接口拿到 aggregator 的结果后立即返回，工作流生成器在后台继续执行 memory_saver，避免 mem0 的抽取与去重延长用户等待时间。

三、DashScope Embedding 适配

1. 问题出在批量调用边界

最初可以把 DashScope 的 OpenAI 兼容地址直接配置给 mem0 的 openai embedder。实际测试中，单条请求可以通过，但 mem0 批量写入记忆时出现过响应解析和向量入库异常。

这类问题需要限定到具体版本和调用方式来看。OpenAI 兼容接口适合减少常规接入成本，但不同 SDK、批量参数和响应解析器组合在一起时，仍然需要集成测试。这里没有继续修改 mem0 内部实现，而是使用它提供的 langchain provider，将 DashScope 原生 SDK 包装成标准的 Embeddings 接口。

2. 适配器不要静默返回零向量

下面是整理后的实现。它同时校验数量、维度和数值有效性；失败时抛出异常，由上层决定重试或降级。

from math import isfinite
from typing import List

from dashscope import TextEmbedding
from langchain_core.embeddings import Embeddings


class DashScopeLangChainEmbeddings(Embeddings):
    def __init__(
        self,
        api_key: str,
        model: str = "text-embedding-v4",
        embedding_dims: int = 2048,
    ) -> None:
        self.api_key = api_key
        self.model = model
        self.embedding_dims = embedding_dims

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        if not texts:
            return []

        response = TextEmbedding.call(
            api_key=self.api_key,
            model=self.model,
            input=texts,
            dimension=self.embedding_dims,
            output_type="dense",
        )
        if response.status_code != 200:
            raise RuntimeError(
                f"DashScope embedding failed: "
                f"code={response.code}, message={response.message}"
            )

        items = list(response.output.get("embeddings", []))
        if items and all("text_index" in item for item in items):
            items.sort(key=lambda item: item["text_index"])

        vectors = [item["embedding"] for item in items]
        if len(vectors) != len(texts):
            raise ValueError(
                f"Embedding count mismatch: expected {len(texts)}, got {len(vectors)}"
            )

        for index, vector in enumerate(vectors):
            if len(vector) != self.embedding_dims:
                raise ValueError(
                    f"Embedding dimension mismatch at {index}: "
                    f"expected {self.embedding_dims}, got {len(vector)}"
                )
            if not all(isfinite(value) for value in vector):
                raise ValueError(f"Embedding contains NaN or Inf at {index}")
            if not any(value != 0.0 for value in vector):
                raise ValueError(f"Embedding is an all-zero vector at {index}")

        return vectors

    def embed_query(self, text: str) -> List[float]:
        return self.embed_documents([text])[0]

早期实现为了保证流程继续，在接口异常时返回了零向量。这个处理不合适：零向量没有可用于余弦比较的方向，有的向量库会拒绝，有的实现会给出没有业务意义的分数。无论哪种结果，都不应该写入 Milvus。

适配器层只负责把错误说清楚，不负责假装成功。重试要有次数和超时上限；最终失败时，写入操作可以跳过，读取操作则返回“本次无记忆上下文”，但不能制造一个有效向量。

3. 模型维度必须贯穿整条链路

text-embedding-v4 的输出维度、适配器的 embedding_dims、mem0 的 embedding_model_dims 和 Milvus collection schema 必须一致。实现时应统一从配置读取 DIMENSION，避免在不同文件里分别写 2048。

如果已经创建 collection，再修改维度不能只改环境变量。更稳妥的方式是新建 collection、重新向量化数据并切换别名，保留可回滚窗口。

四、组装 mem0：配置只是开始

核心配置可以整理为下面这样。示例只使用环境变量，不在源码或演示脚本里保留真实地址、口令和 API Key。

from config import settings  # 所有字段均从环境变量加载


def build_mem0_config(
    embeddings,
    memory_prompt: str,
    enable_graph: bool = False,
) -> dict:
    config = {
        "llm": {
            "provider": "openai",
            "config": {
                "model": settings.MEMORY_MODEL,
                "openai_base_url": settings.MEMORY_OPENAI_BASE_URL,
                "api_key": settings.MEMORY_OPENAI_API_KEY,
                "max_tokens": 2000,
                "temperature": 0.2,
            },
        },
        "embedder": {
            "provider": "langchain",
            "config": {"model": embeddings},
        },
        "vector_store": {
            "provider": "milvus",
            "config": {
                "collection_name": "mem0_memory",
                "embedding_model_dims": settings.DIMENSION,
                "url": settings.MILVUS_URL,
                "token": settings.MILVUS_TOKEN,
            },
        },
    }

    if enable_graph:
        config["graph_store"] = {
            "provider": "neo4j",
            "config": {
                "url": settings.NEO4J_URL,
                "username": settings.NEO4J_USERNAME,
                "password": settings.NEO4J_PASSWORD,
            },
            "custom_prompt": memory_prompt,
        }

    return config

这里有三个容易被忽略的细节。

1. collection 独立只是第一步

记忆 collection 与知识库 collection 分开后，还要保证所有增删查都带上 user_id。如果系统存在多个企业租户，仅有 user_id 还不够，应该把 tenant_id 放进隔离模型，并在进入记忆层前完成鉴权，不能依赖 Prompt 约束数据访问。

agent_id、run_id 这类附加过滤维度也不要只看接口签名。mem0 与不同向量库版本组合时，过滤表达式的支持范围可能不同。用得到就为当前版本补一组 add/search/delete 集成测试；用不到就不要传，避免让无效维度增加排障成本。

2. 相对时间不能永久冻结在启动时刻

记忆提示词可以注入当前日期，用于解释“下周交付”“月底前采购”一类表达。这个思路有效，但常驻进程如果只在启动时构造一次配置，日期会逐渐过期。

生产环境有两种处理方式：在写入前把相对时间正规化，连同消息时间一起交给抽取模型；或者按日期刷新提取上下文。无论采用哪一种，最终事实都应保留“事件时间”和“记录时间”，不能只存一句脱离时间背景的文本。

3. 图谱应该可以按配置关闭

Neo4j 适合保存“客户—单位—需求—产品—采购计划”这类关系，但不是所有记忆都需要图查询。建议用功能开关控制 graph_store：先以向量记忆跑通闭环，再根据多跳关系查询的实际命中率决定是否启用图谱。这样更容易衡量收益，也便于故障隔离。

五、异步单例要防住首请求并发

复用 AsyncMemory 可以避免每次检索都重新创建 Milvus 和 Neo4j 客户端。不过“全局单例 + 懒加载”还不完整。

如果 memory_loader 用 asyncio.gather 同时发起两次检索，服务刚启动时，两次调用都可能看到 self._memory is None，从而重复执行 AsyncMemory.from_config。初始化路径需要加锁，并在锁内再次检查。

import asyncio

from mem0 import AsyncMemory


class MemoryOperator:
    def __init__(self, config: dict) -> None:
        self._config = config
        self._memory: AsyncMemory | None = None
        self._init_lock = asyncio.Lock()

    async def _get_client(self) -> AsyncMemory:
        if self._memory is not None:
            return self._memory

        async with self._init_lock:
            if self._memory is None:
                self._memory = await AsyncMemory.from_config(
                    config_dict=self._config
                )

        return self._memory

除了初始化锁，还要在应用退出时释放底层连接。FastAPI 服务可以在 lifespan 中统一创建和清理资源，比模块导入时创建全局对象更容易测试。

还有一个常被 AsyncMemory 名字掩盖的问题：它不代表每个第三方 SDK 调用都一定是非阻塞的。上面的 TextEmbedding.call 是同步接口，要确认当前 mem0/LangChain 版本会把它调度到线程池；如果没有，就在边界处使用 asyncio.to_thread，或者直接调用 DashScope 的异步 HTTP 接口。这个结论需要用并发压测验证，不能只看函数是否带 async。

六、两路召回：当前问题和用户画像分开查

只用用户当前问题搜索，往往找不到姓名、单位、职位和沟通偏好等基础画像。可以把召回拆成两个并发任务：

relevant_task = memory.search_memory(
    query=current_message,
    user_id=user_id,
)
profile_task = memory.search_memory(
    query="姓名 职位 单位 部门 联系方式 沟通偏好",
    user_id=user_id,
)

relevant, profile = await asyncio.gather(
    relevant_task,
    profile_task,
    return_exceptions=True,
)

随后分别整理为“用户画像”和“相关历史记忆”，每组最多取 5 条；Neo4j 返回的关系按 source + relationship + target 去重后，再限制数量放进上下文。

固定画像查询在中小规模数据上简单有效，但长期看更稳的方案是在写入时给记忆标注类型，例如 identity、preference、project、purchase_plan。这样画像召回可以使用元数据过滤，不必依赖中文关键词命中。

七、向量召回后为什么还要 Rerank

Embedding 适合从大量记忆中快速找候选，但向量距离不等同于“这条记忆能否帮助回答当前问题”。例如客户问某型号能否在六月交付，“六月前需要完成验收”通常比“客户在上海”更值得进入 Prompt，即使两者的向量分数很接近。

可以在 mem0.search 后调用 DashScope gte-rerank-v2：

mem0 向量召回 top-N
  -> Rerank 对 query-document 对进行相关性打分
  -> 取 top-K
  -> 保留原记忆的 id、时间和元数据

实现时有两点需要特别注意。

第一，Rerank 结果最好返回原始下标和分数，再按下标重排对象。不要用“记忆文本 -> 对象”的字典重建结果，因为两条记忆可能文本相同，字典会覆盖其中一条。

ranked_items = await rerank(
    query=query,
    documents=[item["memory"] for item in results],
    top_n=min(top_k, len(results)),
)
reranked_results = [results[item["index"]] for item in ranked_items]

第二，Rerank 是增强环节，不是客服可用性的前置条件。为请求设置超时和有限重试；如果仍然失败，就回退到 mem0 的原始顺序，并记录降级次数。这样既保留检索结果，也能从监控中看出精排服务是否稳定。

是否值得增加 Rerank，不能只凭几次手工体验判断。建议准备一组“问题—相关记忆”标注集，比较 Recall@K、MRR 或 NDCG，并同时观察 P95 延迟和调用成本。

八、Neo4j 记什么，不记什么

图谱提示词应按客服场景约束，重点提取客户、单位、产品、需求、采购计划和服务要求之间的关系。比“提取更多信息”更重要的是限制来源：

1. 只记录用户明确表达的事实，不猜测；
2. 不从客服回复中的推荐内容反推客户属性；
3. “未提供、暂时没有、查不到”不是用户事实，不能覆盖已有记忆；
4. 需求进度、预算周期和交付时间必须带时间信息；
5. 没有可保存内容时返回空结果。

图谱的价值应该落在关系查询上。例如“某单位关注哪些产品”“某项需求关联了哪些产品和交付约束”，这些问题单靠零散文本不容易稳定回答。如果实际查询仍然以单条偏好为主，向量记忆已经足够，Neo4j 可以暂不启用。

图谱抽取还需要单独评估实体合并、关系准确率和删除一致性。不能因为图已经生成，就默认其中每条边都可信。

九、记忆写入为什么只取最近 5 轮

memory_saver 会把 LangGraph state 中的消息转成 mem0 所需的 user/assistant 格式，再截取最近 10 条消息，也就是最多 5 轮对话。

这样做主要是为了控制三个成本：

• 避免每次重复处理整段历史；
• 减少记忆抽取模型的输入长度；
• 降低新事实与已有事实比对、更新和去重的耗时。

需要注意，“最近 5 轮”只是当前业务下的工程参数，不是通用答案。参数应该根据单轮消息长度、事实跨轮距离和写入延迟来调整。

HTTP/SSE 链路可以使用 asyncio.create_task 继续消费剩余工作流，适合“偶尔少写一条记忆也不影响本次答复”的场景。如果业务要求记忆必达，进程退出或发布重启时不能丢任务，就应把写入事件投递到持久化队列，由 worker 幂等消费。建议使用 user_id + conversation_id + turn_id 作为幂等键，避免重试产生重复记忆。

十、失败降级要分层设计

记忆模块位于客服增强链路，任何单点异常都不应直接变成用户侧 500。降级关系可以概括为：

失败位置	当前请求的处理	必须记录的内容
记忆搜索失败	使用空记忆继续路由	错误类型、耗时、用户维度的匿名标识
Rerank 失败	保留向量召回顺序	重试次数、状态码、降级率
图谱查询失败	保留向量记忆	Neo4j 连接和查询耗时
记忆写入失败	不影响已生成回复	待重试事件或丢弃原因
Embedding 返回非法向量	拒绝入库	请求 ID、数量和维度，不记录原始敏感文本

降级不等于静默吞掉异常。至少要监控搜索耗时、写入耗时、空召回率、Rerank 降级率、非法向量数和后台任务失败数。否则系统表面上一直可用，记忆质量却可能已经长期退化。

十一、长期记忆还要处理数据治理

客服记忆可能包含姓名、手机号、单位、采购计划等信息。技术上能抽取，不代表业务上都应该长期保存。上线前至少补齐下面几项：

1. 明确告知用途，并按适用的隐私规则取得必要授权；
2. 只保存提供服务所需的信息，身份证件、支付凭证等高敏感数据默认不进入通用记忆；
3. API Key、数据库口令只从密钥管理服务或环境变量读取，示例代码统一使用占位符；
4. 日志不打印完整记忆、联系方式和模型请求正文；
5. 支持删除单条记忆和按 user_id 删除全部记忆；
6. 制定保留期限，过期记忆自动删除或重新确认；
7. 向量库和图数据库的删除结果需要做一致性校验。

实现时应同时封装删除单条记忆和删除用户全部记忆的入口。真正上线时，还需要把它接到用户注销、隐私请求和后台审计流程，而不是只保留一个内部函数。

十二、上线前怎么验证

我会把测试分成四组，而不是只验证“能不能搜到”。

1. 正确性

• 跨会话能召回明确表达的单位、职位和产品偏好；
• 临时情绪、寒暄和客服推荐不会变成用户事实；
• 新事实与旧事实冲突时能更新，而不是同时保留两个结论；
• 相对时间能转换，并保留记录时间；
• 文本相同的两条记忆经过 Rerank 后不会丢失元数据。

2. 隔离与删除

• 用户 A 的查询永远不能召回用户 B 的记忆；
• 多租户场景同时验证 tenant_id 和 user_id；
• 删除单条和删除全部后，Milvus 与 Neo4j 都无法再查到；
• 备份、缓存和异步任务不会把已删除记忆重新写回。

3. 异常与恢复

• DashScope 超时、限流和返回异常结构；
• Rerank 不可用时回退到原始排序；
• Neo4j 不可用时只使用向量记忆；
• 服务启动时两路并发检索只初始化一个 mem0 客户端；
• 后台写入重复投递时保持幂等。

4. 效果与性能

• 建立业务标注集，评估 Recall@K、MRR/NDCG；
• 分别记录 mem0 检索、Rerank、图查询和写入的 P50/P95/P99；
• 对比增加长期记忆前后的追问次数、人工转接率和回答采纳率；
• 用实际收益决定 top-N、top-K 和 Neo4j 是否启用。

十三、总结

这套方案最终形成了四层分工：mem0 负责事实抽取和记忆生命周期，Milvus 负责向量候选召回，Rerank 负责相关性精排，Neo4j 负责有明确关系价值的结构化记忆；LangGraph 则把读取、回答和后台写入串进客服工作流。

真正影响稳定性的往往不是配置文件里的 provider 名称，而是边界处理：批量 Embedding 的格式与维度、并发首请求的初始化竞争、Rerank 结果与原对象的映射、后台写入的可靠性，以及用户数据的删除和隔离。

如果从头搭建，建议按这个顺序推进：

1. 先区分会话历史、长期记忆和 RAG 知识库；
2. 用独立 collection 跑通按 user_id 隔离的向量记忆；
3. 建立小规模标注集，再决定是否增加 Rerank；
4. 只有出现明确的关系查询需求时再启用 Neo4j；
5. 最后补齐异步队列、指标、删除和审计。

长期记忆不是多接一个数据库，而是一条独立的数据链路。抽取什么、何时召回、失败后如何降级、用户怎样删除，四个问题都有明确答案后，它才适合进入客服主流程。