Repository navigation
Feat/tencent db - #347
Feat/tencent db#347
Conversation
Co-authored-by: Cursor <cursoragent@cursor.com>
AI Code Review审查结论通过 审查范围：e4b5fc3..f4c88b3（commit f4c88b3「支持TencentDB Agent Memory」），新增 TencentDBMemoryService 适配器（353行）、测试（356行）、可运行示例、中英文文档及 release.yml 的 twine>=7.0.0 固定。已核对 MemoryServiceABC 契约、Runner 的 _run_post_turn_processing 调用顺序（先 create_session_summary 后 store_session）、load_memory 工具经 user_key(app_name,user_id) 调 search_memory 的调用链、InMemory/SQL 等既有实现惯例、httpx/httpcore/anyio 跨事件循环语义（实证测试）与 summarizer 行为。计划符合性：功能完整，测试覆盖批量、重试、截断、错误降级等主要路径，文档与示例一致；未发现 SEVERE 级缺陷。主要风险：(1) 在 Runner(defer_post_turn_processing=True) 场景下，init 创建的单一 httpx.AsyncClient 与 asyncio.Lock 会被请求事件循环与后台工作线程事件循环共享，httpcore 连接池与 anyio 原语按事件循环绑定，可能造成挂起/异常或并发错乱（MODERATE）；(2) 会话压缩后新生成的摘要事件（author=system、新 uuid）会被当作 assistant L0 消息重复写入网关，污染记忆存储（MODERATE）；(3) 事件 ID 检查点与锁字典随会话无限增长、_user_id_from_key 对含 '/' 的 app_name 分区错误、网关持久性部分接受时每轮重复重发同一批次（LOW）。测试缺口：无跨循环/并发（defer 模式）、摘要事件过滤、含 '/' 键分区用例。门禁结论：PASSED（无 SEVERE，建议合入前处理 2 个 MODERATE）。 发现的问题未发现需要反馈的代码问题。 |
f4c88b3 to
935b5da
Compare
AI Code Review审查结论通过 审查范围：提交 935b5da「feat: 支持TencentDB Agent Memory」（base e4b5fc3..head 935b5da），共 13 个文件、1441 行新增、3 行删除，核心为 trpc_agent_sdk/memory/tencentdb_memory_service.py（TencentDBMemoryService，365 行）与 tests/memory/test_tencentdb_memory_service.py（394 行），另含示例、文档及 release.yml 改动。 计划符合性：功能实现与上游 TencentDB-Agent-Memory V3 网关契约逐项对齐（/v3/conversation/add 的 1-100 条消息、content 8192 UTF-16 单位、role 枚举；/v3/atomic/search 与 /v3/conversation/search 的响应形状、total_count/accepted_ids 校验），写入路径（store_session 分批、锁、at-least-once 去重、partial/error 过滤）经测试覆盖且与服务端 schema 一致；ttl 关闭、verify_ssl、Header 认证与文档描述一致，示例与文档配套完整，未发现与计划的偏差。 主要风险（本变更引入且有证据）：(1) search_memory 将 query 与 limit 逐字透传给网关而不做长度钳制（服务端 zod 校验 query≤2048、limit≤100），超限时网关返回 400 被宽泛 except 吞掉后静默返回空列表，记忆召回静默失败；(2) _user_id_from_key 用 partition("/") 从 "{app_name}/{user_id}" 中解析 user_id，当 user_id 自身含 "/" 时被截断（写入端用 session.user_id 完整值，搜索端用截断值），造成隔离字段不一致，可能召回同 team/agent 下其他用户的记忆或空召回；(3) _pending_messages 将非 "user" 作者事件（含压缩器生成的 author="system" 总结事件）一律映射为 "assistant" 写入 L0，总结文本被当作助手回复参与后续 L1 提取，语义污染；(4) 搜索路径缺少对超长 query、超限 limit、user_id 含 "/"、原子搜索异常回退等边界的测试覆盖。 测试充分性：写入路径覆盖较全（幂等去重、网关错误重试、不完整响应、分批 100、partial/error/空文本过滤、UTF-16 截断、close 语义），但搜索路径仅有正常映射与回退两条用例，未覆盖上述风险点；测试采用 mock 客户端，未做真实网关契约的集成验证。 门禁结论：PASSED（无 SEVERE 问题）。2 条 MODERATE（搜索参数未校验致静默空召回、user_id 解析破坏隔离）与 2 条 LOW（system 事件角色映射污染、测试缺口）建议在合入前修复或补测，不影响功能主路径的正确性。 发现的问题中等
问题： 触发条件： 当 LLM 生成的记忆检索 query 超过 2048 字符（中文/拼接长上下文很容易超过，KB 级提示词下的工具调用常见），或上层调用传入超过 100 的 实际影响： 400 被第 207 行 修正方向： 在构造 body 前对 中等
问题： 触发条件： 框架调用方传入的 实际影响： 搜索请求的 修正方向： 不要从 key 反解 较低
问题： 触发条件： 会话启用 compress/summarizer 配置（多轮长对话触发自动压缩）且事件流未开启 partial/error 时，总结事件残留在 实际影响： 总结文本（“Previous conversation summary: …”）被当作助手回复写入对话记忆，随后被网关 L1/L2 提取器当作真实对话内容参与记忆提取与去重（服务端对 session 消息会做摘要，总结文本会被重复压缩消费），污染长期记忆质量；同时该事件为非 修正方向： 在 较低
问题： 本提交的测试文件（tests/memory/test_tencentdb_memory_service.py，394 行）完整覆盖了写入路径（幂等去重、网关错误重试、不完整响应重试、分批 100、partial/error/空文本过滤、UTF-16 截断、close 语义），但搜索路径仅有两个正向用例（原子映射、会话搜索回退）和一个网关错误返回空用例，未覆盖本变更中实际引入风险的三条边界。 触发条件： 后续改动触及 实际影响： 上述四条路径正是本次审查发现的 MODERATE/LOW 缺陷所在——若在合入前补上这些测试，第 1、2 条缺陷会在 CI 中直接暴露；当前缺失使静默空召回与跨用户隔离错位等行为无任何防护网。 修正方向： 在测试文件中新增四类用例：对超长 |
| body = { | ||
| "query": query, | ||
| "limit": limit, | ||
| **self._isolation_body(user_id), | ||
| } |
There was a problem hiding this comment.
问题： search_memory 将 query 与 limit 原样写入请求体直接透传给网关，未做任何长度/上限钳制或本地校验。网关服务端契约（zod schema）对 query 校验为 min(1)..max(2048)、对 limit 校验为 min(1)..max(100)；上游官方 SDK 客户端在发送前用 query.slice(0, 2048) 截断以避免该问题，而本实现没有对应处理。
触发条件： 当 LLM 生成的记忆检索 query 超过 2048 字符（中文/拼接长上下文很容易超过，KB 级提示词下的工具调用常见），或上层调用传入超过 100 的 limit 时，网关对超限字段直接返回 400 校验错误。
实际影响： 400 被第 207 行 except Exception 宽泛捕获，仅打一条 warning 日志后返回空 SearchMemoryResponse，调用方（load_memory 工具及依赖记忆上下文作答的 LLM）只能拿到空结果且无任何异常信号——记忆召回静默失败，Agent 在长 query 场景下持续丢失长期记忆，且难以排查。
修正方向： 在构造 body 前对 query 做与上游一致的截断（如 query[:2048]，并配合与 _truncate_utf16 相同的代理对保护），对 limit 做 min(max(1, limit), 100) 钳制；同时在 except 分支中对 TencentDBMemoryError 的 code=400 情形单独记录请求参数，避免静默吞掉可诊断的错误。
| def _user_id_from_key(key: str) -> str: | ||
| _, separator, user_id = key.partition("/") | ||
| resolved = user_id if separator else key | ||
| if not resolved: | ||
| raise ValueError("Memory search key must contain a user ID") | ||
| return resolved |
There was a problem hiding this comment.
问题： _user_id_from_key 使用 key.partition("/") 取第一个 / 之后的所有内容作为 user_id。key 由框架的 user_key(app_name, user_id) = f'{app_name}/{user_id}' 生成（trpc_agent_sdk/utils/_hash_key.py），当 user_id 自身包含 / 时（如 A2A 服务器的 A2A_USER_{request.context_id} 之外的场景、自定义 user_id extractor 输出 URL/路径样式 ID，例如邮箱之外的 org/employee/123 或 u/abc 形式），本方法会把 myapp/org/employee/123 解析为 org/employee/123 的完整串没问题，但更典型地它会把关键部分截错方向——实际上该实现截取的是第一个 / 之后的全部内容，即 app_name 之后的部分全部当作 user_id，这与写入端不一致：store_session 直接使用 session.user_id 原值（第 160 行），而搜索端使用解析后的值。
触发条件： 框架调用方传入的 user_id 含 /（路径式 ID、带域名的标识、或框架内部 user_key 拼接出的含分隔符 ID），使解析结果与写入端 session.user_id 不一致。
实际影响： 搜索请求的 user_id 隔离字段与写入时不同，导致 (1) 无法召回本用户写入的记忆（空召回），或 (2) 在 resolveV3 隔离默认宽松（OFF）的网关配置下召回同 team/agent 下解析结果相同的其他用户的记忆，破坏 (team, agent, user) 三元的记忆隔离边界，产生跨用户信息泄露。
修正方向： 不要从 key 反解 user_id。search_memory 的调用方（InvocationContext.search_memory）已持有 self.user_id（_invocation_context.py:313-314），建议扩展 search_memory 签名直接传入 user_id，或在解析时改用 rsplit("/", 1) 取最后一个段并校验 key 格式，使解析结果与写入端严格一致。
| for event in events: | ||
| if event.id in stored_ids or event.partial or event.is_error(): | ||
| continue | ||
| text = event_to_text(event) | ||
| if not text: | ||
| continue | ||
| text = cls._truncate_utf16(text, _MAX_MESSAGE_UTF16_UNITS) | ||
| pending.append(( | ||
| event.id, | ||
| { | ||
| "role": "user" if event.author == "user" else "assistant", | ||
| "content": text, | ||
| "timestamp": cls._format_timestamp(event.timestamp), | ||
| }, | ||
| )) | ||
| return pending |
There was a problem hiding this comment.
问题： _pending_messages 的 role 映射为 "user" if event.author == "user" else "assistant"，把所有非 user 作者的事件全部当作助手消息。但框架的压缩器（sessions/compact/default/_summarizer.py:400 及 _claw_summarizer.py）会生成 author="system" 的总结事件，且事件带有 set_summary_event(True) 标记；这类事件既不是用户输入也不是助手回复。本服务对 system 作者事件没有按 is_summary_event/作者过滤，直接写入 L0。
触发条件： 会话启用 compress/summarizer 配置（多轮长对话触发自动压缩）且事件流未开启 partial/error 时，总结事件残留在 session.events 中并在每轮后轮处理被 store_session 提交；即使进程内 stored_ids 能去重同一条，新压缩周期产生的总结事件仍会再次提交，且跨会话重复。
实际影响： 总结文本（“Previous conversation summary: …”）被当作助手回复写入对话记忆，随后被网关 L1/L2 提取器当作真实对话内容参与记忆提取与去重（服务端对 session 消息会做摘要，总结文本会被重复压缩消费），污染长期记忆质量；同时该事件为非 user 作者但在 content.role 上并无区分，召回时也会把总结文本当作检索命中返回。
修正方向： 在 _pending_messages 过滤条件中增加 event.is_summary_event() 判断（或显式跳过 author == "system" 的事件），保证只有真实 user/assistant 对话事件进入 L0；或在 store_session 前由 Runner 侧过滤总结事件。
| async def search_memory( | ||
| self, | ||
| key: str, | ||
| query: str, | ||
| limit: int = 10, | ||
| agent_context: Optional[AgentContext] = None, | ||
| ) -> SearchMemoryResponse: | ||
| """Search V3 L1 atomic memories across the user's sessions.""" | ||
| del agent_context | ||
| response = SearchMemoryResponse() | ||
| user_id = self._user_id_from_key(key) | ||
| body = { | ||
| "query": query, | ||
| "limit": limit, | ||
| **self._isolation_body(user_id), | ||
| } | ||
| try: | ||
| # search atomic memories for the user L1 | ||
| data = await self._post(_ATOMIC_SEARCH_PATH, body) | ||
| items = data.get("items", []) | ||
| if not items: | ||
| # search conversation memories for the user L0 | ||
| data = await self._post(_CONVERSATION_SEARCH_PATH, body) | ||
| items = data.get("messages", []) | ||
| if not isinstance(items, list): | ||
| raise ValueError("TencentDB Agent Memory search response must contain a list") | ||
| for item in items: | ||
| entry = self._to_memory_entry(item) | ||
| if entry is not None: | ||
| response.memories.append(entry) | ||
| except Exception as exc: # pylint: disable=broad-except | ||
| logger.warning( | ||
| "Failed to search TencentDB Agent Memory. key=%s, query=%s, err=%s", | ||
| key, | ||
| query, | ||
| exc, | ||
| ) | ||
| return response | ||
|
|
||
| @override | ||
| async def close(self) -> None: | ||
| """Release the owned HTTP client and local delivery checkpoints.""" | ||
| self._stored_event_ids.clear() | ||
| self._store_locks.clear() | ||
| if self._owns_client: | ||
| await self._client.aclose() | ||
|
|
||
| async def _post( | ||
| self, | ||
| path: str, | ||
| body: dict[str, Any], | ||
| ) -> dict[str, Any]: | ||
| headers = { | ||
| "Authorization": f"Bearer {self._config.api_key}", | ||
| "x-tdai-service-id": self._config.service_id, | ||
| "Content-Type": "application/json", | ||
| } | ||
|
|
||
| response = await self._client.post( | ||
| f"{self._config.endpoint.rstrip('/')}{path}", | ||
| json=body, | ||
| headers=headers, | ||
| timeout=self._config.timeout, | ||
| ) | ||
| try: | ||
| payload = response.json() | ||
| except ValueError: | ||
| response.raise_for_status() | ||
| raise ValueError("TencentDB Agent Memory response must be valid JSON") | ||
| if not isinstance(payload, dict): | ||
| response.raise_for_status() | ||
| raise ValueError("TencentDB Agent Memory response must be a JSON object") | ||
|
|
||
| code = payload.get("code") | ||
| if response.is_error or code != 0: | ||
| request_id = (response.headers.get("x-qcloud-transaction-id") or payload.get("request_id") or "") | ||
| raise TencentDBMemoryError( | ||
| code=code if isinstance(code, int) else response.status_code, | ||
| message=str(payload.get("message", response.reason_phrase)), | ||
| request_id=str(request_id), | ||
| ) | ||
|
|
||
| data = payload.get("data", {}) | ||
| if data is None: | ||
| return {} | ||
| if not isinstance(data, dict): | ||
| raise ValueError("TencentDB Agent Memory response 'data' must be an object") | ||
| return data | ||
|
|
||
| def _isolation_body(self, user_id: str) -> dict[str, str]: | ||
| return { | ||
| "team_id": self._config.team_id, | ||
| "agent_id": self._config.agent_id, | ||
| "user_id": user_id, | ||
| } | ||
|
|
||
| @classmethod | ||
| def _pending_messages( | ||
| cls, | ||
| events: list[Event], | ||
| stored_ids: set[str], | ||
| ) -> list[tuple[str, dict[str, str]]]: | ||
| pending: list[tuple[str, dict[str, str]]] = [] | ||
| for event in events: | ||
| if event.id in stored_ids or event.partial or event.is_error(): | ||
| continue | ||
| text = event_to_text(event) | ||
| if not text: | ||
| continue | ||
| text = cls._truncate_utf16(text, _MAX_MESSAGE_UTF16_UNITS) | ||
| pending.append(( | ||
| event.id, | ||
| { | ||
| "role": "user" if event.author == "user" else "assistant", | ||
| "content": text, | ||
| "timestamp": cls._format_timestamp(event.timestamp), | ||
| }, | ||
| )) | ||
| return pending |
There was a problem hiding this comment.
问题： 本提交的测试文件（tests/memory/test_tencentdb_memory_service.py，394 行）完整覆盖了写入路径（幂等去重、网关错误重试、不完整响应重试、分批 100、partial/error/空文本过滤、UTF-16 截断、close 语义），但搜索路径仅有两个正向用例（原子映射、会话搜索回退）和一个网关错误返回空用例，未覆盖本变更中实际引入风险的三条边界。
触发条件： 后续改动触及 search_memory 的参数校验、_user_id_from_key 解析或 _pending_messages 的 role 映射时，现有测试无法捕获回归：(1) 超长 query 与超限 limit 的钳制/截断行为；(2) user_id 含 / 时隔离字段正确性；(3) author="system" 总结事件（is_summary_event）是否被错误写入；(4) 原子搜索请求抛异常（非返回空 items）时是否仍走会话搜索回退。
实际影响： 上述四条路径正是本次审查发现的 MODERATE/LOW 缺陷所在——若在合入前补上这些测试，第 1、2 条缺陷会在 CI 中直接暴露；当前缺失使静默空召回与跨用户隔离错位等行为无任何防护网。
修正方向： 在测试文件中新增四类用例：对超长 query 断言请求体被截断或字段被钳制（配合 _truncate_utf16 的代理对保护）；对 user_id="a/b" 断言请求体 user_id 与写入端一致；构造 summary_event=True 的事件断言不被提交；mock 原子搜索抛出 TencentDBMemoryError 断言回退到会话搜索。
2 participants
No description provided.