摘要在 AI 客服里“记住用户”并不是把全部聊天记录一直塞进 Prompt。更合适的做法是从对话中提取稳定事实跨会话保存并在回答前只召回与当前问题有关的部分。本文完整拆解一条长期记忆链路用 mem0 管理记忆适配 DashScope Embedding将向量记忆与 RAG 知识库分开存入 Milvus按需使用 Neo4j 保存关系再用 Rerank 对候选记忆精排。除了配置和代码文章还会处理并发初始化、非法向量、重复文本映射、后台写入可靠性、失败降级和数据删除等容易遗漏的工程问题。关键词mem0、DashScope、Milvus、Neo4j、Rerank、LangGraph、长期记忆、AI 客服本文对应的主要依赖版本为mem0ai1.0.1、dashscope1.25.6、langchain-core1.2.6、pymilvus2.6.6、neo4j5.28.2。不同版本的配置字段和过滤语法可能变化升级前应先做回归测试。环境准备先安装文中直接使用的依赖pip install mem0ai1.0.1 dashscope1.25.6 langchain-core1.2.6 \ pymilvus2.6.6 neo4j5.28.2 aiohttp3.13.3运行前还需要可访问的 Milvus启用图记忆时再准备 Neo4j。LLM、Embedding、Rerank、Milvus 和 Neo4j 的密钥或连接信息都从环境变量读取不要写进 Python 默认值。一、先分清三类上下文一套完整的客服上下文通常包含会话历史、长期记忆和 RAG 知识库。三者都可能进入模型上下文但职责不同。数据解决的问题生命周期隔离方式典型内容会话历史保持本轮对话连贯当前会话session_id最近几轮问答长期记忆跨会话了解同一客户跨会话user_id单位、职位、产品偏好、采购计划RAG 知识库回答产品和业务知识长期共享collection / tenant产品手册、规格参数、FAQ把三类数据混在一起会带来两个直接问题一是检索结果互相干扰二是数据权限很难说明白。因此mem0 单独使用mem0_memorycollection产品资料与业务文档则保存在各自的知识库 collection 中。长期记忆的目标也随之明确从对话中提取客户明确表达的事实以user_id为边界保存和检索只把本次回答真正需要的记忆放进 Prompt记忆模块不可用时客服主流程仍能继续。二、整体调用链系统入口是 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 的openaiembedder。实际测试中单条请求可以通过但 mem0 批量写入记忆时出现过响应解析和向量入库异常。这类问题需要限定到具体版本和调用方式来看。OpenAI 兼容接口适合减少常规接入成本但不同 SDK、批量参数和响应解析器组合在一起时仍然需要集成测试。这里没有继续修改 mem0 内部实现而是使用它提供的langchainprovider将 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_keyself.api_key, modelself.model, inputtexts, dimensionself.embedding_dims, output_typedense, ) if response.status_code ! 200: raise RuntimeError( fDashScope embedding failed: fcode{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(keylambda item: item[text_index]) vectors [item[embedding] for item in items] if len(vectors) ! len(texts): raise ValueError( fEmbedding count mismatch: expected {len(texts)}, got {len(vectors)} ) for index, vector in enumerate(vectors): if len(vector) ! self.embedding_dims: raise ValueError( fEmbedding dimension mismatch at {index}: fexpected {self.embedding_dims}, got {len(vector)} ) if not all(isfinite(value) for value in vector): raise ValueError(fEmbedding contains NaN or Inf at {index}) if not any(value ! 0.0 for value in vector): raise ValueError(fEmbedding 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_dictself._config ) return self._memory除了初始化锁还要在应用退出时释放底层连接。FastAPI 服务可以在 lifespan 中统一创建和清理资源比模块导入时创建全局对象更容易测试。还有一个常被AsyncMemory名字掩盖的问题它不代表每个第三方 SDK 调用都一定是非阻塞的。上面的TextEmbedding.call是同步接口要确认当前 mem0/LangChain 版本会把它调度到线程池如果没有就在边界处使用asyncio.to_thread或者直接调用 DashScope 的异步 HTTP 接口。这个结论需要用并发压测验证不能只看函数是否带async。六、两路召回当前问题和用户画像分开查只用用户当前问题搜索往往找不到姓名、单位、职位和沟通偏好等基础画像。可以把召回拆成两个并发任务relevant_task memory.search_memory( querycurrent_message, user_iduser_id, ) profile_task memory.search_memory( query姓名 职位 单位 部门 联系方式 沟通偏好, user_iduser_id, ) relevant, profile await asyncio.gather( relevant_task, profile_task, return_exceptionsTrue, )随后分别整理为“用户画像”和“相关历史记忆”每组最多取 5 条Neo4j 返回的关系按source relationship target去重后再限制数量放进上下文。固定画像查询在中小规模数据上简单有效但长期看更稳的方案是在写入时给记忆标注类型例如identity、preference、project、purchase_plan。这样画像召回可以使用元数据过滤不必依赖中文关键词命中。七、向量召回后为什么还要 RerankEmbedding 适合从大量记忆中快速找候选但向量距离不等同于“这条记忆能否帮助回答当前问题”。例如客户问某型号能否在六月交付“六月前需要完成验收”通常比“客户在上海”更值得进入 Prompt即使两者的向量分数很接近。可以在mem0.search后调用 DashScopegte-rerank-v2mem0 向量召回 top-N - Rerank 对 query-document 对进行相关性打分 - 取 top-K - 保留原记忆的 id、时间和元数据实现时有两点需要特别注意。第一Rerank 结果最好返回原始下标和分数再按下标重排对象。不要用“记忆文本 - 对象”的字典重建结果因为两条记忆可能文本相同字典会覆盖其中一条。ranked_items await rerank( queryquery, documents[item[memory] for item in results], top_nmin(top_k, len(results)), ) reranked_results [results[item[index]] for item in ranked_items]第二Rerank 是增强环节不是客服可用性的前置条件。为请求设置超时和有限重试如果仍然失败就回退到 mem0 的原始顺序并记录降级次数。这样既保留检索结果也能从监控中看出精排服务是否稳定。是否值得增加 Rerank不能只凭几次手工体验判断。建议准备一组“问题—相关记忆”标注集比较 RecallK、MRR 或 NDCG并同时观察 P95 延迟和调用成本。八、Neo4j 记什么不记什么图谱提示词应按客服场景约束重点提取客户、单位、产品、需求、采购计划和服务要求之间的关系。比“提取更多信息”更重要的是限制来源只记录用户明确表达的事实不猜测不从客服回复中的推荐内容反推客户属性“未提供、暂时没有、查不到”不是用户事实不能覆盖已有记忆需求进度、预算周期和交付时间必须带时间信息没有可保存内容时返回空结果。图谱的价值应该落在关系查询上。例如“某单位关注哪些产品”“某项需求关联了哪些产品和交付约束”这些问题单靠零散文本不容易稳定回答。如果实际查询仍然以单条偏好为主向量记忆已经足够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 降级率、非法向量数和后台任务失败数。否则系统表面上一直可用记忆质量却可能已经长期退化。十一、长期记忆还要处理数据治理客服记忆可能包含姓名、手机号、单位、采购计划等信息。技术上能抽取不代表业务上都应该长期保存。上线前至少补齐下面几项明确告知用途并按适用的隐私规则取得必要授权只保存提供服务所需的信息身份证件、支付凭证等高敏感数据默认不进入通用记忆API Key、数据库口令只从密钥管理服务或环境变量读取示例代码统一使用占位符日志不打印完整记忆、联系方式和模型请求正文支持删除单条记忆和按user_id删除全部记忆制定保留期限过期记忆自动删除或重新确认向量库和图数据库的删除结果需要做一致性校验。实现时应同时封装删除单条记忆和删除用户全部记忆的入口。真正上线时还需要把它接到用户注销、隐私请求和后台审计流程而不是只保留一个内部函数。十二、上线前怎么验证我会把测试分成四组而不是只验证“能不能搜到”。1. 正确性跨会话能召回明确表达的单位、职位和产品偏好临时情绪、寒暄和客服推荐不会变成用户事实新事实与旧事实冲突时能更新而不是同时保留两个结论相对时间能转换并保留记录时间文本相同的两条记忆经过 Rerank 后不会丢失元数据。2. 隔离与删除用户 A 的查询永远不能召回用户 B 的记忆多租户场景同时验证tenant_id和user_id删除单条和删除全部后Milvus 与 Neo4j 都无法再查到备份、缓存和异步任务不会把已删除记忆重新写回。3. 异常与恢复DashScope 超时、限流和返回异常结构Rerank 不可用时回退到原始排序Neo4j 不可用时只使用向量记忆服务启动时两路并发检索只初始化一个 mem0 客户端后台写入重复投递时保持幂等。4. 效果与性能建立业务标注集评估 RecallK、MRR/NDCG分别记录 mem0 检索、Rerank、图查询和写入的 P50/P95/P99对比增加长期记忆前后的追问次数、人工转接率和回答采纳率用实际收益决定 top-N、top-K 和 Neo4j 是否启用。十三、总结这套方案最终形成了四层分工mem0 负责事实抽取和记忆生命周期Milvus 负责向量候选召回Rerank 负责相关性精排Neo4j 负责有明确关系价值的结构化记忆LangGraph 则把读取、回答和后台写入串进客服工作流。真正影响稳定性的往往不是配置文件里的 provider 名称而是边界处理批量 Embedding 的格式与维度、并发首请求的初始化竞争、Rerank 结果与原对象的映射、后台写入的可靠性以及用户数据的删除和隔离。如果从头搭建建议按这个顺序推进先区分会话历史、长期记忆和 RAG 知识库用独立 collection 跑通按user_id隔离的向量记忆建立小规模标注集再决定是否增加 Rerank只有出现明确的关系查询需求时再启用 Neo4j最后补齐异步队列、指标、删除和审计。长期记忆不是多接一个数据库而是一条独立的数据链路。抽取什么、何时召回、失败后如何降级、用户怎样删除四个问题都有明确答案后它才适合进入客服主流程。
摘要在 AI 客服里“记住用户”并不是把全部聊天记录一直塞进 Prompt。更合适的做法是从对话中提取稳定事实跨会话保存并在回答前只召回与当前问题有关的部分。本文完整拆解一条长期记忆链路用 mem0 管理记忆适配 DashScope Embedding将向量记忆与 RAG 知识库分开存入 Milvus按需使用 Neo4j 保存关系再用 Rerank 对候选记忆精排。除了配置和代码文章还会处理并发初始化、非法向量、重复文本映射、后台写入可靠性、失败降级和数据删除等容易遗漏的工程问题。关键词mem0、DashScope、Milvus、Neo4j、Rerank、LangGraph、长期记忆、AI 客服本文对应的主要依赖版本为mem0ai1.0.1、dashscope1.25.6、langchain-core1.2.6、pymilvus2.6.6、neo4j5.28.2。不同版本的配置字段和过滤语法可能变化升级前应先做回归测试。环境准备先安装文中直接使用的依赖pip install mem0ai1.0.1 dashscope1.25.6 langchain-core1.2.6 \ pymilvus2.6.6 neo4j5.28.2 aiohttp3.13.3运行前还需要可访问的 Milvus启用图记忆时再准备 Neo4j。LLM、Embedding、Rerank、Milvus 和 Neo4j 的密钥或连接信息都从环境变量读取不要写进 Python 默认值。一、先分清三类上下文一套完整的客服上下文通常包含会话历史、长期记忆和 RAG 知识库。三者都可能进入模型上下文但职责不同。数据解决的问题生命周期隔离方式典型内容会话历史保持本轮对话连贯当前会话session_id最近几轮问答长期记忆跨会话了解同一客户跨会话user_id单位、职位、产品偏好、采购计划RAG 知识库回答产品和业务知识长期共享collection / tenant产品手册、规格参数、FAQ把三类数据混在一起会带来两个直接问题一是检索结果互相干扰二是数据权限很难说明白。因此mem0 单独使用mem0_memorycollection产品资料与业务文档则保存在各自的知识库 collection 中。长期记忆的目标也随之明确从对话中提取客户明确表达的事实以user_id为边界保存和检索只把本次回答真正需要的记忆放进 Prompt记忆模块不可用时客服主流程仍能继续。二、整体调用链系统入口是 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 的openaiembedder。实际测试中单条请求可以通过但 mem0 批量写入记忆时出现过响应解析和向量入库异常。这类问题需要限定到具体版本和调用方式来看。OpenAI 兼容接口适合减少常规接入成本但不同 SDK、批量参数和响应解析器组合在一起时仍然需要集成测试。这里没有继续修改 mem0 内部实现而是使用它提供的langchainprovider将 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_keyself.api_key, modelself.model, inputtexts, dimensionself.embedding_dims, output_typedense, ) if response.status_code ! 200: raise RuntimeError( fDashScope embedding failed: fcode{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(keylambda item: item[text_index]) vectors [item[embedding] for item in items] if len(vectors) ! len(texts): raise ValueError( fEmbedding count mismatch: expected {len(texts)}, got {len(vectors)} ) for index, vector in enumerate(vectors): if len(vector) ! self.embedding_dims: raise ValueError( fEmbedding dimension mismatch at {index}: fexpected {self.embedding_dims}, got {len(vector)} ) if not all(isfinite(value) for value in vector): raise ValueError(fEmbedding contains NaN or Inf at {index}) if not any(value ! 0.0 for value in vector): raise ValueError(fEmbedding 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_dictself._config ) return self._memory除了初始化锁还要在应用退出时释放底层连接。FastAPI 服务可以在 lifespan 中统一创建和清理资源比模块导入时创建全局对象更容易测试。还有一个常被AsyncMemory名字掩盖的问题它不代表每个第三方 SDK 调用都一定是非阻塞的。上面的TextEmbedding.call是同步接口要确认当前 mem0/LangChain 版本会把它调度到线程池如果没有就在边界处使用asyncio.to_thread或者直接调用 DashScope 的异步 HTTP 接口。这个结论需要用并发压测验证不能只看函数是否带async。六、两路召回当前问题和用户画像分开查只用用户当前问题搜索往往找不到姓名、单位、职位和沟通偏好等基础画像。可以把召回拆成两个并发任务relevant_task memory.search_memory( querycurrent_message, user_iduser_id, ) profile_task memory.search_memory( query姓名 职位 单位 部门 联系方式 沟通偏好, user_iduser_id, ) relevant, profile await asyncio.gather( relevant_task, profile_task, return_exceptionsTrue, )随后分别整理为“用户画像”和“相关历史记忆”每组最多取 5 条Neo4j 返回的关系按source relationship target去重后再限制数量放进上下文。固定画像查询在中小规模数据上简单有效但长期看更稳的方案是在写入时给记忆标注类型例如identity、preference、project、purchase_plan。这样画像召回可以使用元数据过滤不必依赖中文关键词命中。七、向量召回后为什么还要 RerankEmbedding 适合从大量记忆中快速找候选但向量距离不等同于“这条记忆能否帮助回答当前问题”。例如客户问某型号能否在六月交付“六月前需要完成验收”通常比“客户在上海”更值得进入 Prompt即使两者的向量分数很接近。可以在mem0.search后调用 DashScopegte-rerank-v2mem0 向量召回 top-N - Rerank 对 query-document 对进行相关性打分 - 取 top-K - 保留原记忆的 id、时间和元数据实现时有两点需要特别注意。第一Rerank 结果最好返回原始下标和分数再按下标重排对象。不要用“记忆文本 - 对象”的字典重建结果因为两条记忆可能文本相同字典会覆盖其中一条。ranked_items await rerank( queryquery, documents[item[memory] for item in results], top_nmin(top_k, len(results)), ) reranked_results [results[item[index]] for item in ranked_items]第二Rerank 是增强环节不是客服可用性的前置条件。为请求设置超时和有限重试如果仍然失败就回退到 mem0 的原始顺序并记录降级次数。这样既保留检索结果也能从监控中看出精排服务是否稳定。是否值得增加 Rerank不能只凭几次手工体验判断。建议准备一组“问题—相关记忆”标注集比较 RecallK、MRR 或 NDCG并同时观察 P95 延迟和调用成本。八、Neo4j 记什么不记什么图谱提示词应按客服场景约束重点提取客户、单位、产品、需求、采购计划和服务要求之间的关系。比“提取更多信息”更重要的是限制来源只记录用户明确表达的事实不猜测不从客服回复中的推荐内容反推客户属性“未提供、暂时没有、查不到”不是用户事实不能覆盖已有记忆需求进度、预算周期和交付时间必须带时间信息没有可保存内容时返回空结果。图谱的价值应该落在关系查询上。例如“某单位关注哪些产品”“某项需求关联了哪些产品和交付约束”这些问题单靠零散文本不容易稳定回答。如果实际查询仍然以单条偏好为主向量记忆已经足够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 降级率、非法向量数和后台任务失败数。否则系统表面上一直可用记忆质量却可能已经长期退化。十一、长期记忆还要处理数据治理客服记忆可能包含姓名、手机号、单位、采购计划等信息。技术上能抽取不代表业务上都应该长期保存。上线前至少补齐下面几项明确告知用途并按适用的隐私规则取得必要授权只保存提供服务所需的信息身份证件、支付凭证等高敏感数据默认不进入通用记忆API Key、数据库口令只从密钥管理服务或环境变量读取示例代码统一使用占位符日志不打印完整记忆、联系方式和模型请求正文支持删除单条记忆和按user_id删除全部记忆制定保留期限过期记忆自动删除或重新确认向量库和图数据库的删除结果需要做一致性校验。实现时应同时封装删除单条记忆和删除用户全部记忆的入口。真正上线时还需要把它接到用户注销、隐私请求和后台审计流程而不是只保留一个内部函数。十二、上线前怎么验证我会把测试分成四组而不是只验证“能不能搜到”。1. 正确性跨会话能召回明确表达的单位、职位和产品偏好临时情绪、寒暄和客服推荐不会变成用户事实新事实与旧事实冲突时能更新而不是同时保留两个结论相对时间能转换并保留记录时间文本相同的两条记忆经过 Rerank 后不会丢失元数据。2. 隔离与删除用户 A 的查询永远不能召回用户 B 的记忆多租户场景同时验证tenant_id和user_id删除单条和删除全部后Milvus 与 Neo4j 都无法再查到备份、缓存和异步任务不会把已删除记忆重新写回。3. 异常与恢复DashScope 超时、限流和返回异常结构Rerank 不可用时回退到原始排序Neo4j 不可用时只使用向量记忆服务启动时两路并发检索只初始化一个 mem0 客户端后台写入重复投递时保持幂等。4. 效果与性能建立业务标注集评估 RecallK、MRR/NDCG分别记录 mem0 检索、Rerank、图查询和写入的 P50/P95/P99对比增加长期记忆前后的追问次数、人工转接率和回答采纳率用实际收益决定 top-N、top-K 和 Neo4j 是否启用。十三、总结这套方案最终形成了四层分工mem0 负责事实抽取和记忆生命周期Milvus 负责向量候选召回Rerank 负责相关性精排Neo4j 负责有明确关系价值的结构化记忆LangGraph 则把读取、回答和后台写入串进客服工作流。真正影响稳定性的往往不是配置文件里的 provider 名称而是边界处理批量 Embedding 的格式与维度、并发首请求的初始化竞争、Rerank 结果与原对象的映射、后台写入的可靠性以及用户数据的删除和隔离。如果从头搭建建议按这个顺序推进先区分会话历史、长期记忆和 RAG 知识库用独立 collection 跑通按user_id隔离的向量记忆建立小规模标注集再决定是否增加 Rerank只有出现明确的关系查询需求时再启用 Neo4j最后补齐异步队列、指标、删除和审计。长期记忆不是多接一个数据库而是一条独立的数据链路。抽取什么、何时召回、失败后如何降级、用户怎样删除四个问题都有明确答案后它才适合进入客服主流程。
双梁正问题 【 torch 实战】（Python代码实现）)
