Repository navigation
feat(memory): 集成 Hy-Memory 长期记忆服务 - #362
raychen911 wants to merge 1 commit into
Conversation
AI Code Review审查结论通过 审查范围：594bfb0..b22a1c8（feat(memory): 集成 Hy-Memory 长期记忆服务），共 12 个文件、1675 行新增。核心新增 计划符合性： 主要发现：1) 门禁结论：无 SEVERE 级缺陷，整体设计严谨、兼容 SDK 2.0.6 与框架契约；建议修复 MODERATE 并发问题后合入。 发现的问题中等
问题： 触发条件： 使用 实际影响： 每次 修正方向： 将锁与事件循环解耦：改用 较低
问题： 触发条件： 用户按文档描述不设置 实际影响： 期待自动场景路由的用户在默认配置下得到的都是 chat 记忆，working/coding 记忆永远不会生成；若调用方在请求级 options 中但未在 config 中设置 修正方向： 将 较低
问题： 新增测试仅覆盖单事件循环内的顺序调用（ 触发条件： 按 实际影响： 缺测导致本文所述 MODERATE 问题的回归无任何测试防线；测试套件全绿会误导合入决策，使跨循环 RuntimeError 的静默内存丢失在真实部署中才暴露。 修正方向： 新增测试：在不同事件循环（如 |
| session_key = (session.save_key, session.id) | ||
| lock = self._store_locks.setdefault(session_key, asyncio.Lock()) | ||
|
|
||
| async with lock: | ||
| stored_ids = self._stored_event_ids.setdefault(session_key, set()) |
There was a problem hiding this comment.
问题： _store_locks 缓存的 asyncio.Lock 对象绑定创建它的第一个事件循环（Python 3.10+ 的 Lock._get_loop 强制校验），但 store_session 的实际执行循环并不固定。Runner 的 defer_post_turn_processing=True 模式在专用线程 _PostTurnWorkerThread 中新建事件循环执行 post-turn 处理；runner close（_shutdown_post_turn_worker 置 _post_turn_thread=None）后再次 run 会启动新线程/新事件循环，此时对同一 session 复用旧的 asyncio.Lock 会抛 RuntimeError: <lock> is bound to a different event loop。该异常发生在 async with lock（第 167 行，位于内层 try 之外），向上传播到 Runner._run_post_turn_processing 的 except Exception 后仅记录日志。
触发条件： 使用 defer_post_turn_processing=True（或 enable_post_turn_processing 非阻塞模式）时：① 同一 Runner 在 close() 之后复用继续 run_async；② 文档明确支持的『多个短生命周期 Runner 共享同一 memory service』（close_memory_service_on_close=False 场景，不同 Runner 的 post-turn 线程各有独立事件循环）处理同一 (save_key, session.id)。两种情况下第二次 store_session 都会跨循环获取已绑定旧循环的 lock。
实际影响： 每次 store_session 都在获取锁时抛 RuntimeError，该 session 的后续所有轮次内存写入被 Runner 静默吞掉（仅 error 日志），且每次重试均复现同一错误——长期记忆永久静默丢失，直到进程重启。
修正方向： 将锁与事件循环解耦：改用 threading.Lock（临界区仅做内存 set 操作与发起提交，不跨 await 持有），或按 (session_key, loop_id) 分桶并在 loop 关闭时清理旧锁（例如在 close() 中清空 _store_locks/_stored_event_ids），或改用 asyncio.Lock 每次在获取前检查 lock._loop is asyncio.get_running_loop() 失效则重建。
| scene: Optional[Literal["chat", "working", "coding"]] = Field( | ||
| default="chat", | ||
| description="Explicit memory scene, or None for Hy-Memory routing.", | ||
| ) |
There was a problem hiding this comment.
问题： HyMemoryServiceConfig.scene 字段默认值为 "chat"，但字段 docstring 与 zh/memory.md 文档均宣称 None 时由 Hy-Memory 自动判定场景（『None：写入包含工具轨迹时由 Hy-Memory 判断 Scene；普通文本直接走 Chat』）。_normalize_scene 虽然实现了 None → 自动路由 的路径，但由于默认值不是 None，用户不显式配置 scene=None 时自动路由分支实际不可达——所有未配置的写入都被强制走 chat 链路。
触发条件： 用户按文档描述不设置 scene（预期『包含工具轨迹时自动判定 working/coding』），实际写入全部强制为 chat 场景，工具轨迹被 SDK 的 strip_tool_messages 剥离。
实际影响： 期待自动场景路由的用户在默认配置下得到的都是 chat 记忆，working/coding 记忆永远不会生成；若调用方在请求级 options 中但未在 config 中设置 scene 同理。功能可用性低于文档承诺，且 _normalize_scene 的 None 分支成为死代码路径（仅显式传 None 才可达）。
修正方向： 将 scene 默认值改为 None（与文档一致，自动路由为默认），保留 chat 为显式选择；或同步修改文档说明默认强制 chat、需显式 scene=None 才启用自动路由。
| self._stored_event_ids: dict[tuple[str, str], set[str]] = {} | ||
| self._store_locks: dict[tuple[str, str], asyncio.Lock] = {} |
There was a problem hiding this comment.
问题： 新增测试仅覆盖单事件循环内的顺序调用（test_store_preserves_tool_trace_and_deduplicates、test_failed_write_is_retried），未覆盖 _store_locks/_stored_event_ids 在框架真实部署形态下的并发行为：① defer_post_turn_processing=True 时 post-turn 在专用线程/事件循环执行；② 多个短生命周期 Runner 共享同一 HyMemoryService 实例（close_memory_service_on_close=False 是框架文档化的共享模式）处理同一 session。这两类场景正是 asyncio.Lock 跨事件循环绑定的隐患所在（见 store_session 的并发锁实现）。
触发条件： 按 _PostTurnWorkerThread 的机制构造测试：两个独立事件循环（各自线程）中先后对同一 (save_key, session.id) 调用 store_session，或在 runner.close() 后复用同一 runner 再次 run。
实际影响： 缺测导致本文所述 MODERATE 问题的回归无任何测试防线；测试套件全绿会误导合入决策，使跨循环 RuntimeError 的静默内存丢失在真实部署中才暴露。
修正方向： 新增测试：在不同事件循环（如 asyncio.new_event_loop + run_coroutine_threadsafe）中连续调用同一 service 实例的 store_session，断言不抛 RuntimeError 且事件被正确提交；并在修复锁实现后立即接入该测试。
- 背景： 当前框架缺少 Hy-Memory 适配，无法直接使用其 Chat、Working 和 Coding记忆能力。 - 方案： 新增HyMemoryService，支持增量写入、工具轨迹转换、跨会话召回及请求级隔离；补充可选依赖、测试和中英文文档。 - 示例： 新增 Hy-Memory 示例，演示对接远程 BGE-M3 Embedding，并验证用户偏好可在不同 Session 间正确召回。
b22a1c8 to
45bdfb4
Compare
AI Code Review审查结论不通过 审查范围：base 594bfb0..head 45bdfb4（feat(memory): 集成 Hy-Memory 长期记忆服务），共 12 个文件、约 1788 行新增，核心为新增的 trpc_agent_sdk/memory/hy_memory_service.py（HyMemoryService 适配器，551 行）及其配置导出、pyproject 依赖（hy-memory>=2.0.6 额外组）、2 个测试文件和 1 个示例目录，另含中英文 docs。全部结论均对照 Hy-Memory SDK v2.0.6 源码（/tmp/hy-memory-repo）逐一验证。 计划符合性：整体方向符合"集成 Hy-Memory 长期记忆服务"的意图——store_session/search_memory/close 均委托 async_add/async_search，scene 归一化、agent 上下文选项注入、会话锁序列化等设计合理。但适配器在失败重试、异常隔离、并发与锁、以及 tool 消息映射上存在若干由本次变更引入的高置信正确性缺陷，其中失败重试路径经 SDK 幂等回放语义验证后确认为永久性数据丢失，属阻止合入级别。 主要问题（均已对照 SDK 源码验证）：
测试充分性：测试丰富（443 行）且覆盖并发序列化、scene 校验、自定义 embedder，但 test_failed_write_is_retried 断言了与 SDK 幂等回放语义相矛盾的行为（真实 SDK 下重试永远失败），且未覆盖异常路径下的重试行为、无效 scene 崩溃、锁取消竞态，测试与实现的失败语义未对齐。 门禁结论：FAILED。建议修复 request_id 唯一性/失败重试语义（如失败后放弃该批事件或轮换 request_id 并持久化已尝试集合）、将 _normalize_scene 移入 try、修复锁泄漏与取消竞态后再合入。 发现的问题严重
问题： 确定性 触发条件： 实际影响： 每次重试都返回与首次相同的失败结果，且失败路径（第 214-228 行）不更新 修正方向： 采用兼具幂等与可重试的请求标识：如将失败批次从 严重
问题： 触发条件： 用户通过 实际影响： runner 的 post-turn 处理在每次触发该 session 的存储时抛出未处理异常，中断消息循环/agent 执行流程；search 路径的异常则被下游 修正方向： 将 中等
问题： 触发条件： 任意异常或 SDK 拒绝写入后，同一 session 继续产生新的事件并再次进入 post-turn 存储（或直接重跑 实际影响： 对同一批损坏/失败事件无限重试，占用 post-turn 处理时长、重复消耗 SDK 调用与 token；若根因持续（如消息格式不被 SDK 接受）则每次对话后续 turn 都被该批次阻塞，且与第 199-230 行的确定性 request_id 缺陷叠加后重试也永远不会成功。 修正方向： 失败路径明确处置该批次：剔除/标记导致失败的事件（如超出可写 Agent 协议的事件）不再重试，或对失败批次做有界重试（数量与退避上限）后放弃；成功路径保持现状即可。 中等
问题： 触发条件： 持有 实际影响： 锁被泄漏后该 session 的所有后续存储写入在自旋中永久等待（ 修正方向： 在 中等
问题： 触发条件： 任一事件含 实际影响： Devin SDK 的记忆生成管线会把 tool 消息与其调用分离甚至丢弃（ 修正方向： 为同一事件内的调用与响应使用一致的合成 id 规则（如同一 较低
问题： 触发条件： 实际影响： 召回记忆注入 prompt 时角色失真，下游按"用户说过"的语义消费 assistant 内容，影响记忆回放的真实性；与仓库中 in_memory/mem0 适配器按事件作者保留 role 的实现不一致。 修正方向： 按 较低
问题： 触发条件： SDK 连接异常、scene 归一化被抛出（第 248 行在 try 外）或远端返回错误时，任意一次用户查询触发检索。 实际影响： 检索失败被当作"没有记忆"吞掉，runner 不会告警或降级，用户得到看似正常的空记忆回复，掩盖了后端故障，排查困难；与 修正方向： 失败时在 |
| result = await self._client.async_add( | ||
| messages, | ||
| user_id=session.user_id, | ||
| agent_id=agent_id, | ||
| session_id=session.id, | ||
| source=options.get("source", self._config.source), | ||
| scene=scene, | ||
| workspace_id=workspace_id, | ||
| branch=branch, | ||
| request_id=self._request_id(session, event_ids), | ||
| meta=meta, | ||
| ) | ||
| if isinstance(result, dict) and result.get("success") is False: | ||
| logger.warning( | ||
| "Hy-Memory rejected session write. save_key=%s, " | ||
| "session_id=%s, result=%s", | ||
| session.save_key, | ||
| session.id, | ||
| result, | ||
| ) | ||
| return | ||
| except Exception as exc: # pylint: disable=broad-except | ||
| logger.warning( | ||
| "Failed to store session in Hy-Memory. save_key=%s, " | ||
| "session_id=%s, err=%s", | ||
| session.save_key, | ||
| session.id, | ||
| exc, | ||
| ) | ||
| return | ||
|
|
||
| stored_ids.update(event_ids) |
There was a problem hiding this comment.
问题： 确定性 _request_id（sha256(app\0user\0session\0event_ids)，第 352-356 行）与 Hy-Memory SDK 的失败回放语义组合导致重试永远失败：SDK 的 async_add 以 write_outcomes 表（request_id 为主键，INSERT OR IGNORE）实现幂等，失败结果会以 "failed" 状态持久化，之后对相同 request_id 的调用直接回放该失败结果（cache_sqlite.py _claim_write_outcome 返回 {**result, "replayed": True}）。
触发条件： async_add 任意一次失败（网络异常、SDK 返回 {"success": False}，或另一线程对同一批事件仍在写入途中的 in-flight 回放），随后同一事件集合再次调用 store_session（含测试 test_failed_write_is_retried 的第二次调用）。
实际影响： 每次重试都返回与首次相同的失败结果，且失败路径（第 214-228 行）不更新 stored_ids，事件永久静默丢失、无法写入长期记忆；用户与 README 宣称的"失败可重试"保证不成立。
修正方向： 采用兼具幂等与可重试的请求标识：如将失败批次从 stored_ids 中移除并轮换/废弃该 request_id，或先查询 SDK 的 outcome 状态区分"已写入"与"已失败"，失败时剔除损坏事件后以新 request_id 重试；同时将 test_failed_write_is_retried 的断言改为匹配 SDK 真实语义。
| scene = self._normalize_scene(options.get("scene", self._config.scene)) | ||
| workspace_id = options.get("workspace_id", self._config.workspace_id) | ||
| branch = options.get("branch", self._config.branch) | ||
| if branch and not workspace_id: | ||
| logger.warning("Ignoring Hy-Memory branch because workspace_id is not set.", ) | ||
| branch = None | ||
|
|
||
| event_ids = [event.id for event in pending] | ||
| meta = { | ||
| "app_name": session.app_name, | ||
| "event_ids": event_ids, | ||
| } | ||
| custom_meta = options.get("meta") | ||
| if isinstance(custom_meta, dict): | ||
| meta.update(custom_meta) | ||
|
|
||
| try: | ||
| result = await self._client.async_add( |
There was a problem hiding this comment.
问题： _normalize_scene（第 336-343 行，非法值抛 ValueError）在 store_session 中位于 try 块之外（第 182 行，try 从第 198 行开始），search_memory 同样（第 248 行，try 从第 251 行开始），无效 scene 异常不受任何捕获。
触发条件： 用户通过 meta={"scene": ...}、agent_context 选项或 HyMemoryServiceConfig.scene 传入 _VALID_SCENES 之外的任意值（如 "gaming"、拼写错误），或传入不可字符串化的对象。
实际影响： runner 的 post-turn 处理在每次触发该 session 的存储时抛出未处理异常，中断消息循环/agent 执行流程；search 路径的异常则被下游 except Exception（第 292-297 行）连带吞掉，无任何错误信号，检索结果静默为空——同一根因两种后果均不可控。
修正方向： 将 _normalize_scene 调用移入 try 块内（与 async_add/async_search 同域），或改为返回默认 scene 并记录 warning 的容错语义，保证无效配置只影响单次调用而不中断 runner。
| "session_id=%s, result=%s", | ||
| session.save_key, | ||
| session.id, | ||
| result, | ||
| ) | ||
| return | ||
| except Exception as exc: # pylint: disable=broad-except | ||
| logger.warning( | ||
| "Failed to store session in Hy-Memory. save_key=%s, " | ||
| "session_id=%s, err=%s", | ||
| session.save_key, | ||
| session.id, | ||
| exc, | ||
| ) | ||
| return | ||
|
|
||
| stored_ids.update(event_ids) |
There was a problem hiding this comment.
问题： async_add 抛异常或返回 success=False 时，两个 return 分支（第 228、226 行）都不执行 stored_ids.update(event_ids)，事件保持 pending 状态，并被 session.events 永久保留（事件在 rehydration 后 id 不变）。
触发条件： 任意异常或 SDK 拒绝写入后，同一 session 继续产生新的事件并再次进入 post-turn 存储（或直接重跑 store_session），pending 每次都会包含同一批旧事件并整体重发。
实际影响： 对同一批损坏/失败事件无限重试，占用 post-turn 处理时长、重复消耗 SDK 调用与 token；若根因持续（如消息格式不被 SDK 接受）则每次对话后续 turn 都被该批次阻塞，且与第 199-230 行的确定性 request_id 缺陷叠加后重试也永远不会成功。
修正方向： 失败路径明确处置该批次：剔除/标记导致失败的事件（如超出可写 Agent 协议的事件）不再重试，或对失败批次做有界重试（数量与退避上限）后放弃；成功路径保持现状即可。
| async def _store_lock( | ||
| self, | ||
| session_key: tuple[str, str], | ||
| ) -> AsyncIterator[None]: | ||
| """Serialize a session across threads without blocking event loops.""" | ||
| with self._store_state_lock: | ||
| lock = self._store_locks.setdefault(session_key, threading.Lock()) | ||
| while not lock.acquire(blocking=False): | ||
| await asyncio.sleep(_STORE_LOCK_RETRY_INTERVAL_SECONDS) | ||
| try: | ||
| yield | ||
| finally: | ||
| lock.release() |
There was a problem hiding this comment.
问题： _store_lock 的 while not lock.acquire(blocking=False): await asyncio.sleep(...) 自旋循环中，若任务在睡眠期间被取消（CancelledError 在 await 点抛出），lock 已被本协程 close 但 acquire 的结果未被记录，finally: lock.release() 只会在 yield 之后执行——取消发生在 yield 之前，因此 release 不会运行、或对未持有的锁调用 release 抛 RuntimeError。
触发条件： 持有 _store_lock 的任务在自旋等待期间被取消（runner 关闭、超时、任务取消、进程退出）；此外 _store_locks 与 _stored_event_ids 两个字典随 session 数持续增长且从不清理。
实际影响： 锁被泄漏后该 session 的所有后续存储写入在自旋中永久等待（asyncio.sleep 死循环），长时间运行的服务中 session 一多内存无界增长；同进程内多线程共享锁的场景（tests/memory/test_hy_memory_service.py 的 test_shared_service_serializes_writes_across_event_loops）下更易暴露。
修正方向： 在 acquire 成功后立即用 try/finally 包裹整个持锁区间并将 release 置于最外层，使取消路径也必然释放；或在 session 关闭/close() 时清理 _store_locks 与 _stored_event_ids 对应条目。
| if not isinstance(response, str): | ||
| response = json.dumps( | ||
| response, | ||
| ensure_ascii=False, | ||
| default=str, | ||
| ) | ||
| messages.append({ | ||
| "role": "tool", | ||
| "tool_call_id": function_response.id or f"{event.id}-response-{index}", | ||
| "name": function_response.name or "tool", | ||
| "content": response, | ||
| }) | ||
|
|
||
| function_calls = event.get_function_calls() | ||
| if not text and not function_calls: | ||
| continue | ||
| role = "user" if event.author == "user" else "assistant" | ||
| message: dict[str, Any] = { | ||
| "role": role, | ||
| "content": text, | ||
| } | ||
| if function_calls: | ||
| message["role"] = "assistant" | ||
| message["tool_calls"] = [] | ||
| for index, function_call in enumerate(function_calls): | ||
| arguments = function_call.args or {} | ||
| message["tool_calls"].append({ | ||
| "id": function_call.id or f"{event.id}-call-{index}", | ||
| "type": "function", | ||
| "function": { | ||
| "name": function_call.name or "tool", | ||
| "arguments": json.dumps( | ||
| arguments, | ||
| ensure_ascii=False, | ||
| default=str, |
There was a problem hiding this comment.
问题： _events_to_messages 为缺失 id 的 function response 合成 {event.id}-response-{index}（第 376 行）、为对应调用合成 {event.id}-call-{index}（第 395 行），两套命名空间永不相同，SDK 侧无法把 tool 结果与 tool_calls 关联；且 tool 响应消息先于其调用消息写入列表（循环内先处理 function_responses 再处理 function_calls，与事件内实际顺序相反）。
触发条件： 任一事件含 function_call（插件/工具事件，part.is_thought=False 时）或其 function response 缺少 id（SDK 回放/重建事件时常见）。
实际影响： Devin SDK 的记忆生成管线会把 tool 消息与其调用分离甚至丢弃（ChatMessage.normalize 对未关联 tool_calls 的 "tool" 消息标记无效），工具调用痕迹在长期记忆中残缺或丢失，与 store_session 文档宣称的"preserving tool-call traces"相悖；顺序颠倒还可能让 SDK 因 tool 消息先于调用而拒绝该消息。
修正方向： 为同一事件内的调用与响应使用一致的合成 id 规则（如同一 {event.id}-call-{index} 的 call 与 response id），并按事件内容顺序（调用在前、响应在后）排列消息。
| if channel == "chain": | ||
| text = f"Code evolution context:\n{text}" | ||
| author = (metadata.get("author") or metadata.get("role") or metadata.get("source") or f"hy-memory:{channel}") | ||
| timestamp = cls._item_timestamp(metadata) | ||
| return MemoryEntry( | ||
| content=Content( | ||
| role="user", | ||
| parts=[Part.from_text(text=text)], |
There was a problem hiding this comment.
问题： _to_memory_entry 将搜索结果硬编码为 Content(role="user", ...)（第 470 行），未使用 metadata 中可能存在的角色/权威信息；搜出的是 assistant 生成的内容时（如经验链、重写文本），角色被错误标注为 user。
触发条件： async_search 命中 working/coding 管线中的经验、事实链或 assistant 生成内容（channel 为 "experience"、"chain" 等），而 metadata 无 author/role/source 字段时。
实际影响： 召回记忆注入 prompt 时角色失真，下游按"用户说过"的语义消费 assistant 内容，影响记忆回放的真实性；与仓库中 in_memory/mem0 适配器按事件作者保留 role 的实现不一致。
修正方向： 按 metadata 推导 role（如 role="assistant" 或 metadata.get("role") 存在时使用之），缺省时再回退 "user"。
| return response | ||
|
|
||
| app_name, user_id = self._identity_from_key(key) | ||
| options = self._effective_options(agent_context) | ||
| agent_id = options.get("agent_id") or self._config.agent_id or app_name | ||
| scene = self._normalize_scene(options.get("scene", self._config.scene)) | ||
| workspace_id = options.get("workspace_id", self._config.workspace_id) | ||
| branch = options.get("branch", self._config.branch) | ||
| if branch and not workspace_id: | ||
| branch = None | ||
|
|
||
| try: | ||
| result = await self._client.async_search( | ||
| query, | ||
| scene=scene, | ||
| search_mode=options.get("search_mode", self._config.search_mode), | ||
| user_ids=[user_id], | ||
| agent_ids=[agent_id] if agent_id else None, | ||
| include_sources=options.get( | ||
| "include_sources", | ||
| self._config.include_sources, | ||
| ), | ||
| exclude_sources=options.get( | ||
| "exclude_sources", | ||
| self._config.exclude_sources, | ||
| ), | ||
| limit=limit, | ||
| min_score=options.get("min_score", self._config.min_score), | ||
| schema_limit=options.get( | ||
| "schema_limit", | ||
| self._config.schema_limit, | ||
| ), | ||
| schema_min_score=options.get( | ||
| "schema_min_score", | ||
| self._config.schema_min_score, | ||
| ), | ||
| intention_limit=options.get( | ||
| "intention_limit", | ||
| self._config.intention_limit, | ||
| ), | ||
| workspace_id=workspace_id, | ||
| branch=branch, | ||
| ) | ||
| for item, channel in self._iter_search_items(result): | ||
| entry = self._to_memory_entry(item, channel) | ||
| if entry is not None: | ||
| response.memories.append(entry) | ||
| if len(response.memories) >= limit: | ||
| break | ||
| except Exception as exc: # pylint: disable=broad-except | ||
| logger.warning( | ||
| "Failed to search Hy-Memory. key=%s, query=%s, err=%s", | ||
| key, | ||
| query, | ||
| exc, | ||
| ) | ||
| return response |
There was a problem hiding this comment.
问题： search_memory 对 limit <= 0 静默返回空结果、对 async_search 失败仅 logger.warning 后返回空（第 292-297 行），调用方无法区分"无匹配记忆"与"检索失败"。
触发条件： SDK 连接异常、scene 归一化被抛出（第 248 行在 try 外）或远端返回错误时，任意一次用户查询触发检索。
实际影响： 检索失败被当作"没有记忆"吞掉，runner 不会告警或降级，用户得到看似正常的空记忆回复，掩盖了后端故障，排查困难；与 store_session 的失败告警（warning + return）一致但缺少任何状态信号。
修正方向： 失败时在 SearchMemoryResponse 增加错误标记或向调用方传播异常（由 runner 决定降级策略），并补充该路径的测试。
背景： 当前框架缺少 Hy-Memory 适配，无法直接使用其 Chat、Working 和 Coding记忆能力。
方案： 新增HyMemoryService，支持增量写入、工具轨迹转换、跨会话召回及请求级隔离；补充可选依赖、测试和中英文文档。
示例： 新增 Hy-Memory 示例，演示对接远程 BGE-M3 Embedding，并验证用户偏好可在不同 Session 间正确召回。