Repository navigation
Feature/advanced memory service - #332
CongkeChen wants to merge 6 commits into
Conversation
Please enter the commit message for your changes. Lines starting
54d9e9d to
11b3f0c
Compare
AI Code Review审查结论通过 本审查覆盖 base_commit..head_commit（108 个文件，+6012/-6201），主题为 advanced memory 服务化与 Session Compact 重构： 发现的问题中等
问题： 触发条件： 任何启用 token 模式（ 实际影响： usage-baseline 的指纹匹配保护形同虚设：上一次请求的 usage 事件可能被错误地当作当前请求的基线（即使两次请求的静态指纹不同），导致 token 估算偏高/偏低、 修正方向： 删除 中等
问题： 触发条件： 会话增量以独立的工具响应（function response，非工具调用链）结束时，例如用户在工具结果后回复，或单次工具调用 + 结果的短交互；此时 实际影响： 提取频率相对已发布的 1.1.20 行为降低：以 tool-response 收尾的常见增量不再满足工具条件，会话记忆更新被推迟到凑够工具调用数或下一次自然断点，长对话中记忆滞后（部分此类输入会被无限期挂起直至上下文被迫更新），用户可感知的会话记忆时效性下降。 修正方向： 若是有意的收紧，需在配置中说明该语义变更并同步更新文档；否则建议恢复以「ends-in-tool-call」作为唯一不满足自然断点的条件（ 中等
问题： 新增的 触发条件： 同一 实际影响： 会话状态（含 修正方向： 对 较低
问题： 触发条件： 配置 实际影响： SQL 后端为每次请求额外建立/释放一个 DB 连接与事务；local 后端反复执行文件系统与后台任务状态检查。在 QPS 较高时放大数据库连接压力与目录扫描开销，且 修正方向： 在 runtime 内以单一标志/计时器实现一次性或限频初始化（如首次调用后置位，或在 |
| @classmethod | ||
| def record_request_context( | ||
| self, | ||
| request: "LlmRequest", | ||
| ctx: "InvocationContext | None", | ||
| cls, | ||
| request: LlmRequest, | ||
| ctx: InvocationContext, | ||
| ) -> None: | ||
| """Stage the final request fingerprint for persistence on the response Event.""" | ||
| session = getattr(ctx, "session", None) if ctx is not None else None | ||
| state = getattr(session, "state", None) | ||
| session = ctx.session | ||
| state = session.state | ||
| if isinstance(state, dict): | ||
| state["advanced_memory_pending_request_context_fingerprint"] = _request_static_fingerprint(request) | ||
|
|
There was a problem hiding this comment.
问题： TokenContextTracker.record_request_context 把指纹写入 session.state["advanced_memory_pending_request_context_fingerprint"]，而 _latest_usage_baseline 读取的是事件 custom_metadata 中的 advanced_memory_request_context_fingerprint；全仓库（含测试）没有 producer 写该事件元数据，也没有任何代码读取该 state key，构成写入端与读取端两侧都悬挂的死机制。
触发条件： 任何启用 token 模式（token_context_tracker 配置了上下文窗口）并运行 create_session_summary_by_request（_auto_compact.py:581 调用 record_request_context）的会话；后续请求进入 _latest_usage_baseline 时，recorded_fingerprint 恒为 None，其过滤分支永远不生效。
实际影响： usage-baseline 的指纹匹配保护形同虚设：上一次请求的 usage 事件可能被错误地当作当前请求的基线（即使两次请求的静态指纹不同），导致 token 估算偏高/偏低、auto_compact 阈值判断与实际上下文大小不一致；相关测试通过硬编码「outdated」指纹模拟环境，掩盖了该机制从未接通的事实。
修正方向： 删除 record_request_context 及其在 _auto_compact.py:581 的调用，并简化 _latest_usage_baseline 中的指纹分支；若确实需要基线指纹，应在生成响应事件时把指纹持久化到事件 custom_metadata（与读取端对齐），并补充端到端测试验证指纹被写入并被读取端消费。
| if self._last_event_has_tool_call(pending): | ||
| return SessionMemoryExtractionResult(False, "unsafe-boundary") | ||
| natural_break = not self._event_has_tool_response(pending[-1]) | ||
| threshold_met = ((context_tokens >= threshold if checkpoint_context_tokens is None else | ||
| (context_tokens < checkpoint_context_tokens or context_tokens - | ||
| checkpoint_context_tokens >= threshold)) if token_mode else pending_chars >= threshold) | ||
| tool_condition_met = tool_calls >= config.session_memory_tool_calls_between_updates or natural_break | ||
| tool_condition_met = tool_calls >= config.tool_calls_between_updates or natural_break | ||
| if not force and (not threshold_met or not tool_condition_met): | ||
| return SessionMemoryExtractionResult(False, "threshold-not-met") | ||
| return SessionMemoryExtractionResult(reason="threshold-not-met") |
There was a problem hiding this comment.
问题： extract_if_needed 的 natural_break 语义被变更：现在以 tool-response 结束时不再算自然断点（natural_break = not self._event_has_tool_response(pending[-1])），而以 tool-call 结尾则直接返回「unsafe-boundary」。对比基线 f05797d 中 natural_break = not self._last_event_has_tool_call(pending)，只有以 tool-call 结尾时才算非自然断点。
触发条件： 会话增量以独立的工具响应（function response，非工具调用链）结束时，例如用户在工具结果后回复，或单次工具调用 + 结果的短交互；此时 tool_calls 计数小于 tool_calls_between_updates 且无自然断点，threshold-not-met 一直返回。
实际影响： 提取频率相对已发布的 1.1.20 行为降低：以 tool-response 收尾的常见增量不再满足工具条件，会话记忆更新被推迟到凑够工具调用数或下一次自然断点，长对话中记忆滞后（部分此类输入会被无限期挂起直至上下文被迫更新），用户可感知的会话记忆时效性下降。
修正方向： 若是有意的收紧，需在配置中说明该语义变更并同步更新文档；否则建议恢复以「ends-in-tool-call」作为唯一不满足自然断点的条件（natural_break = not self._last_event_has_tool_call(pending)），并为「增量以工具响应结尾」补充单元测试覆盖提取时机。
| if not config.enabled or not config.long_term_memory_injection_enabled: | ||
| return False | ||
| await self._runtime.initialize() | ||
| await runtime.initialize() |
There was a problem hiding this comment.
问题： LongTermMemoryContext.apply 在每个模型请求前无条件调用 await runtime.initialize()；_runtime.py 的 initialize() 在 SQL 后端会新建 DB session（create_db_session），在 local 后端会触发目录创建与 _local_cleanup.start() 幂等检查，属于每次请求的重复初始化开销。
触发条件： 配置 long_term_memory_injection_enabled=True 的常规多轮会话，每轮模型请求（含工具循环中的多次调用）都会经过该路径。
实际影响： SQL 后端为每次请求额外建立/释放一个 DB 连接与事务；local 后端反复执行文件系统与后台任务状态检查。在 QPS 较高时放大数据库连接压力与目录扫描开销，且 initialize() 内部若带 I/O（如 lock 等待）还会增加每请求延迟；功能不受影响。
修正方向： 在 runtime 内以单一标志/计时器实现一次性或限频初始化（如首次调用后置位，或在 initialize 内仅对缺失资源做前置检查），apply 仅在未初始化或显式失效时调用；补充一个高并发多轮请求的基准测试验证初始化不再重复执行。
AI Code Review审查结论不通过 审查范围：base f05797d..head 54d9e9d（"Feature/advanced memory service"），109 个文件，+6127/−6313。本次重构将 advanced memory 能力拆分为 sessions/compact/advanced（会话压缩与自动压缩）与 memory/advanced_memory（长期记忆运行时/存储/作用域），新增 Session.compact_events、三种后端 update_session_state 部分更新、AdvancedAutoCompactSummarizer 系列、作用域化存储与 SQL/Redis 记忆后端，同时删除旧 _summarizer_manager/TranscriptSessionService 及整套旧测试。 计划符合性：功能主体完整落地（压缩管线、记忆抽取、作用域存储、三后端均已实现），但存在多个高置信正确性与数据完整性缺陷：1) AdvancedAutoCompactSummarizer 默认构造（config=None）直接 AttributeError，默认用法崩溃；2) 会话记忆抽取的 natural_break 语义相对旧实现被反转（完成的 tool turn 不再视为自然断点），工具型会话记忆抽取被无限推迟，甚至把半成品工具回合落库；3) Redis 新增 update_session_state 是盲读改写整包覆盖，并发互相覆盖，且 store_historical_events=False 时抹掉已存 historical_events；4) 本地记忆索引 TTL 过期时直接删除 memory_dir 下全部 *.md 主题文件，纯读路径造成长期记忆整体丢失；5) 高级自动压缩"阻塞"路径把 LlmResponse 塞进过滤链，代理主循环按 Event 处理时缺 is_error()/is_final_response() 而崩溃；6) 抽取子代理失败被计入压缩失败，连续失败后可阻塞主模型请求；7) token 预算的请求指纹通道重构后无写入方，usage 基线静默退化。 安全性/并发：Redis 会话部分更新无乐观并发控制（读改写覆盖），SQL 侧通过 get_for_update 加锁但 update_session_state 未使用；本地存储删除无锁、与写入竞争。稳定性：阻塞/抽取失败路径存在两个崩溃点。兼容性：migrate_legacy 迁移函数零调用，旧平铺 memory 布局升级后不可见；DefaultSessionSummarizerManager 兼容保留。 测试充分性：本次删除约 2350 行旧 advanced_memory 测试（test_autocompact/test_history_snip/test_microcompact/test_tool_result_budget/test_session_memory_extractor/test_transcript_session_service 等），新增 test_session_compact.py 仅 268 行，未覆盖 config=None 构造、force 抽取失败、natural_break 语义、阻塞响应类型、Redis 部分更新并发、本地索引 TTL 剪枝等关键路径，存在明显测试缺口。 门禁结论：存在 5 个 SEVERE 问题，判定 FAILED，建议修复后再合入。 发现的问题严重
问题： 触发条件： 任何以默认参数调用 实际影响： 默认构造该压缩器无法使用，调用方必须显式创建 config 对象，与签名声明的默认值行为不符；这是本次重构引入的回归（旧 修正方向： 使用 严重
问题： 会话记忆抽取的 触发条件： 默认 实际影响： 长期记忆抽取频率错误（过少），且可能把半成品工具状态作为已完成的检查点固化，抽取出的记忆内容不可靠。 修正方向： 恢复旧语义： 严重
问题： 新增的 触发条件： (a) 实际影响： 已归档的会话历史被永久擦除；并发场景下状态与事件窗口互相覆盖，数据丢失。 修正方向： 用 Redis 原子结构（hash + HSET 局部更新 state 字段）或 WATCH/MULTI 事务 + 版本校验实现单字段更新，避免整包读改写； 严重
问题： 本地存储 触发条件： 写入频率低于 TTL（如 30 分钟一次写入、TTL 3600s）时，任何纯读取方（preload、read_topic、list_topics 触发 read_index）在索引过期后调用即触发整体删除； 实际影响： 长期记忆全部主题文件被永久删除，纯读路径造成数据整体丢失，事后 修正方向： TTL 过期只应使索引条目失效，不应删除主题文件；删除应由配置化的清理任务按主题级 TTL 逐个执行，并加写锁/串行化。 严重
问题： 触发条件： 连续压缩失败达到 实际影响： 本应优雅返回的"阻塞"提示变成请求运行异常（run loop 崩溃），用户得到 500/异常而非阻塞说明。 修正方向： 阻塞路径应产出 中等
问题： 触发条件： 会话接近自动压缩阈值时，抽取子代理偶发限流/模型超时/解析失败，连续几轮即可把健康的主请求硬阻断。 实际影响： 主对话被以 ADVANCED_AUTOCOMPACT_BLOCKED_MESSAGE 整体阻断，即使主模型本身完全正常；抽取失败与压缩失败未区分记账。 修正方向： 将抽取失败与压缩失败分开计数，或抽取失败降级（跳过记忆抽取但继续压缩），不参与阻塞判定。 中等
问题： token 预算的指纹通道失效： 触发条件： 每次请求重建后，护栏恒为空，usage 基线只能靠内容指纹匹配；当新请求片段与历史 usage 事件内容指纹重合时（压缩后的 summary/保留内容与旧请求相同），会误匹配到旧的、更小的 usage_tokens。 实际影响： token 估算静默偏低，auto_compact/blocking 阈值判定失真（可能延迟压缩或低估阻塞阈值），压缩触发时机不可靠；该通道从"有护栏"回归为"无护栏"。 修正方向： 恢复写入链路：在响应事件落库前把 中等
问题： 触发条件： 同会话内：压缩持久化成功但调用方崩溃后重试，或另一 worker 已先完成压缩，session-memory 来源的 record（boundary_event_id=None）重试时按签名+occurrence 在已替换的当前事件上解析不到边界；或会话从存储重载后事件窗口已不含边界事件。 实际影响： 幂等重放机制不可达，重试直接以 ValueError 中断 before-model 管线，请求失败；压缩记录无法自然去重。 修正方向： 在边界解析前先按 compaction_id 检查幂等（把幂等探测提前到 较低
问题： 触发条件： 已有旧版 SDK 平铺 实际影响： 旧长期记忆对新作用域路径（tenants///memory）不可见，升级用户静默丢失历史记忆可见性；迁移能力形同虚设。 修正方向： 在 较低
问题： 触发条件： 会话中相同内容（如同一用户消息文本）多次出现且中间夹带其他不可见事件时，occurrence 计数偏移，选中错误边界事件或解析失败。 实际影响： 可能把较新的用户消息一起归档（活动窗口数据错位），或触发前述 ValueError 让压缩失败。 修正方向： 边界解析直接使用压缩时记录的 |
| self._runtime = AdvancedAutoCompactSummarizerRuntime(config=config or AdvancedAutoCompactSummarizerConfig()) | ||
| self._auto_compact_config = config.auto_compact | ||
| self._session_memory_extractor = self._create_extractor(session_memory_extractor) |
There was a problem hiding this comment.
问题： AdvancedAutoCompactSummarizer.__init__ 中 self._auto_compact_config = config.auto_compact 在签名声明 config: AdvancedAutoCompactSummarizerConfig | None = None 时未处理 None，config or AdvancedAutoCompactSummarizerConfig() 的兜底只作用于 _runtime。默认构造（不传 config）即触发 AttributeError: 'NoneType' object has no attribute 'auto_compact'。
触发条件： 任何以默认参数调用 AdvancedAutoCompactSummarizer() 的路径（公开 API 默认用法、自定义集成），即构造期崩溃。
实际影响： 默认构造该压缩器无法使用，调用方必须显式创建 config 对象，与签名声明的默认值行为不符；这是本次重构引入的回归（旧 AutoCompactSummarizer 无此崩溃路径）。
修正方向： 使用 self._auto_compact_config = (config or AdvancedAutoCompactSummarizerConfig()).auto_compact，或把 config 解析为局部变量后复用。
| natural_break = not self._event_has_tool_response(pending[-1]) | ||
| threshold_met = ((context_tokens >= threshold if checkpoint_context_tokens is None else | ||
| (context_tokens < checkpoint_context_tokens or context_tokens - | ||
| checkpoint_context_tokens >= threshold)) if token_mode else pending_chars >= threshold) | ||
| tool_condition_met = tool_calls >= config.session_memory_tool_calls_between_updates or natural_break | ||
| tool_condition_met = tool_calls >= config.tool_calls_between_updates or natural_break |
There was a problem hiding this comment.
问题： 会话记忆抽取的 natural_break 判断被反转：新代码 natural_break = not self._event_has_tool_response(pending[-1])（最后一条事件不含 function_response 才算自然断点），而旧实现 natural_break = not self._last_event_has_tool_call(pending)（不处于未完成工具回合才算自然断点）。完成的工具回合以 function_response 事件结尾，因此"每轮完成一个工具调用"的典型会话现在 natural_break=False。
触发条件： 默认 tool_calls_between_updates=3 时，单工具调用会话在回合边界永不满足 tool_condition_met，会话记忆更新被无限推迟；反之用户在半成品工具回合中输入新消息时 pending[-1] 是普通文本，natural_break=True，会把未完成的工具回合状态提交进会话记忆。
实际影响： 长期记忆抽取频率错误（过少），且可能把半成品工具状态作为已完成的检查点固化，抽取出的记忆内容不可靠。
修正方向： 恢复旧语义：natural_break = not self._last_event_has_tool_call(pending)（以"是否处于未完成工具回合"为断点依据）。
| async def update_session_state( | ||
| self, | ||
| session: Session, | ||
| state_delta: dict[str, Any], | ||
| ) -> None: | ||
| """Persist session-scoped state without replacing the caller's Event window.""" | ||
| if not state_delta: | ||
| return | ||
| session.state.update(state_delta) | ||
|
|
||
| async with self._redis_storage.create_db_session() as redis_session: | ||
| key = session_key(session.app_name, session.user_id, session.id) | ||
| storage_session = await self._get_session(redis_session, key) | ||
| if not storage_session: | ||
| logger.warning( | ||
| "Session %s not found in Redis while updating state", |
There was a problem hiding this comment.
问题： 新增的 RedisSessionService.update_session_state 采用盲读改写：_get_session（其内 store_historical_events=False 时强制 session.historical_events = []）→ 原地改 state → _set_session（整包 SET key session_json）覆盖整个会话键，且无任何乐观并发控制；update_session 同样整包覆盖。
触发条件： (a) store_historical_events=False 时，记忆检查点通过 update_session_state(SESSION_MEMORY_STATE_KEY) 持久化即把已存的 historical_events 抹空；(b) 两个并发写者（如 update_session 与另一 worker 的 update_session_state），后者读到的旧快照覆盖前者的 events/state 变更。
实际影响： 已归档的会话历史被永久擦除；并发场景下状态与事件窗口互相覆盖，数据丢失。
修正方向： 用 Redis 原子结构（hash + HSET 局部更新 state 字段）或 WATCH/MULTI 事务 + 版本校验实现单字段更新，避免整包读改写；update_session_state 不得触发 _set_session 对 events/historical_events 的重写。
| def _read_index_sync(self) -> str: | ||
| if _is_expired(self.index_path, self._config.memory_ttl_seconds): | ||
| for path in self._paths.memory_dir.glob("*.md"): | ||
| path.unlink(missing_ok=True) | ||
| return "" |
There was a problem hiding this comment.
问题： 本地存储 _read_index_sync 中，当索引文件本身按 memory_ttl_seconds 过期（索引 mtime 只在写入时刷新）时，直接 for path in memory_dir.glob("*.md"): path.unlink() 删除全部主题文件，而主题文件自身的 TTL 未参与判断。
触发条件： 写入频率低于 TTL（如 30 分钟一次写入、TTL 3600s）时，任何纯读取方（preload、read_topic、list_topics 触发 read_index）在索引过期后调用即触发整体删除；LocalAdvancedMemoryCleanup._run（_storage.py:195-197）同样按索引过期整体删除。
实际影响： 长期记忆全部主题文件被永久删除，纯读路径造成数据整体丢失，事后 read_memory 返回 not found；删除与并发写入无锁竞争。
修正方向： TTL 过期只应使索引条目失效，不应删除主题文件；删除应由配置化的清理任务按主题级 TTL 逐个执行，并加写锁/串行化。
| invocation_ctx.end_invocation = True | ||
| rsp.rsp = result | ||
| rsp.is_continue = False | ||
| rsp.error = None |
There was a problem hiding this comment.
问题： AdvancedAutoCompactSummarizerFilter._before 在阻塞时设置 rsp.rsp = result（LlmResponse，内容为 ADVANCED_AUTOCOMPACT_BLOCKED_MESSAGE）+ is_continue=False。run_stream_filters 将该 LlmResponse 直接 yield 给代理主循环，而 _base_agent.py:305-312 与 _llm_agent.py:509 立即调用 event.has_content()/event.is_error()/event.is_final_response()——is_error() 与 is_final_response() 只存在于 Event（events/_event.py:256/180），LlmResponse（models/_llm_response.py）没有这些方法。
触发条件： 连续压缩失败达到 max_failures 且达到阻塞阈值时，过滤链输出阻塞响应，代理主循环收到 LlmResponse 后调用 event.is_error() 抛 AttributeError。
实际影响： 本应优雅返回的"阻塞"提示变成请求运行异常（run loop 崩溃），用户得到 500/异常而非阻塞说明。
修正方向： 阻塞路径应产出 Event（或把 LlmResponse 转成带错误码的 Event）再放入 rsp.rsp，使代理主循环按统一 Event 协议处理。
| original_contents = [content.model_copy(deep=True) for content in request.contents] | ||
| try: | ||
| compact_record: AdvancedAutoCompactRecord | None = None | ||
| if self._session_memory_extractor is not None: | ||
| await self._session_memory_extractor.extract_if_needed( | ||
| ctx, | ||
| force=True, |
There was a problem hiding this comment.
问题： _apply_scoped 把 await self._session_memory_extractor.extract_if_needed(ctx, force=True) 放在压缩 try 块内，抽取子代理（与主模型同模型、可能限流）的任何异常都被 except Exception 归入 state.consecutive_failures += 1（_auto_compact.py:776），与压缩失败混为一谈，一旦达到 max_failures（默认 3）且达到阻塞阈值即返回 blocked=True，主模型请求被阻断。
触发条件： 会话接近自动压缩阈值时，抽取子代理偶发限流/模型超时/解析失败，连续几轮即可把健康的主请求硬阻断。
实际影响： 主对话被以 ADVANCED_AUTOCOMPACT_BLOCKED_MESSAGE 整体阻断，即使主模型本身完全正常；抽取失败与压缩失败未区分记账。
修正方向： 将抽取失败与压缩失败分开计数，或抽取失败降级（跳过记忆抽取但继续压缩），不参与阻塞判定。
| session = ctx.session | ||
| events = session.events |
There was a problem hiding this comment.
问题： token 预算的指纹通道失效：_latest_usage_baseline 从事件 custom_metadata 读取 advanced_memory_request_context_fingerprint 作为过期 usage 匹配护栏，但全仓库已无任何写入方（唯一写入者 TranscriptSessionService 在本重构中删除）；record_request_context 把 advanced_memory_pending_request_context_fingerprint 写进 session.state，也没有任何代码把它拷贝到响应事件的 custom_metadata。
触发条件： 每次请求重建后，护栏恒为空，usage 基线只能靠内容指纹匹配；当新请求片段与历史 usage 事件内容指纹重合时（压缩后的 summary/保留内容与旧请求相同），会误匹配到旧的、更小的 usage_tokens。
实际影响： token 估算静默偏低，auto_compact/blocking 阈值判定失真（可能延迟压缩或低估阻塞阈值），压缩触发时机不可靠；该通道从"有护栏"回归为"无护栏"。
修正方向： 恢复写入链路：在响应事件落库前把 record_request_context 缓存的指纹写入该事件 custom_metadata，或移除该护栏并改用显式的 usage 事件关联字段。
|
|
||
| boundary_event_id = record.boundary_event_id or self._resolve_boundary_event_id( | ||
| ctx, | ||
| record.boundary_signature, | ||
| record.boundary_occurrence, | ||
| ) | ||
| if boundary_event_id is None: | ||
| raise ValueError("Cannot map the AdvancedAutoCompactSummarizer boundary to an active Session Event") | ||
|
|
There was a problem hiding this comment.
问题： _persist_session_compaction 在调用 compact_events 之前先做边界解析（_auto_compact.py:420-425）：boundary_event_id = record.boundary_event_id or self._resolve_boundary_event_id(...)，解析失败立刻 raise ValueError("Cannot map ...")；而 compact_events 的幂等保护（_session.py:155-156 按 session_compaction_id 元数据返回 False）位于调用之内。当重试/会话重载后边界事件已被先前成功的压缩替换/移入历史，ValueError 在幂等保护生效前抛出。
触发条件： 同会话内：压缩持久化成功但调用方崩溃后重试，或另一 worker 已先完成压缩，session-memory 来源的 record（boundary_event_id=None）重试时按签名+occurrence 在已替换的当前事件上解析不到边界；或会话从存储重载后事件窗口已不含边界事件。
实际影响： 幂等重放机制不可达，重试直接以 ValueError 中断 before-model 管线，请求失败；压缩记录无法自然去重。
修正方向： 在边界解析前先按 compaction_id 检查幂等（把幂等探测提前到 _persist_session_compaction 入口），解析失败时降级为跳过而非抛异常。
| def migrate_legacy(self, app_name: str, user_id: str) -> "ScopedAdvancedMemoryRuntime": | ||
| """Move an old flat Advanced Memory layout into one explicit tenant. | ||
|
|
||
| Refuses to overwrite a tenant that already contains data. | ||
| """ | ||
| scoped = self.for_scope(app_name, user_id) | ||
| legacy_paths = self.paths | ||
| target_root = scoped.paths.tenant_root_dir | ||
| if target_root.exists(): | ||
| raise FileExistsError(f"Target Advanced Memory tenant already exists: {target_root}") | ||
| if not legacy_paths.memory_dir.exists(): | ||
| raise FileNotFoundError("No legacy Advanced Memory directories exist") | ||
| target_root.mkdir(parents=True) | ||
| if legacy_paths.memory_dir.exists(): | ||
| shutil.move(str(legacy_paths.memory_dir), str(scoped.paths.memory_dir)) | ||
| return scoped |
There was a problem hiding this comment.
问题： AdvancedMemoryRuntime.migrate_legacy（将旧平铺 memory 布局迁移到 tenants 作用域布局）在本 SDK 全仓库无任何调用方（grep 确认零引用），for_scope/initialize 也未自动触发迁移。
触发条件： 已有旧版 SDK 平铺 MEMORY/ 存储的用户升级到本版本后继续使用（未手动调用未公开的 migrate_legacy）。
实际影响： 旧长期记忆对新作用域路径（tenants///memory）不可见，升级用户静默丢失历史记忆可见性；迁移能力形同虚设。
修正方向： 在 for_scope/initialize 首次发现平铺布局时自动调用迁移（带上现有防覆盖保护），或至少对外文档化迁移入口并触发告警。
| def _resolve_boundary_event_id( | ||
| self, | ||
| ctx: InvocationContext, | ||
| signature: str, | ||
| occurrence: int, | ||
| ) -> str | None: | ||
| """Map one request-content boundary back to an active Session Event.""" | ||
| seen = 0 | ||
| for event in ctx.session.events: | ||
| content = getattr(event, "content", None) | ||
| if content is None or content_signature(content) != signature: | ||
| continue | ||
| seen += 1 |
There was a problem hiding this comment.
问题： _resolve_boundary_event_id（_auto_compact.py:372-382）把请求内容层面计算的 boundary_occurrence 直接按"相同 signature 的 occurrence 次"映射到 ctx.session.events，但请求 contents 与 session.events 的构成不一致（会话含系统/部分/过滤事件，请求重建时跳过部分事件），同一 signature 两侧出现次数可不同。
触发条件： 会话中相同内容（如同一用户消息文本）多次出现且中间夹带其他不可见事件时，occurrence 计数偏移，选中错误边界事件或解析失败。
实际影响： 可能把较新的用户消息一起归档（活动窗口数据错位），或触发前述 ValueError 让压缩失败。
修正方向： 边界解析直接使用压缩时记录的 boundary_event_id（已存在但 session-memory 来源未保存），不依赖重新计数；或在请求与事件两侧使用同一过滤规则计数。
11b3f0c to
2814f74
Compare
2814f74 to
3c0f340
Compare
AI Code Review审查结论通过 本次审查范围：base_commit f05797d 至 head_commit 2814f74 的 109 个文件、+6117/-6342，核心为「Feature/advanced memory service」：advanced memory 服务化（local/redis/sql 三种后端）、Session Compact 重构（ 审查结论概述：计划符合度总体良好——重构保持了既有 summarizer 面（CompressSummarizer 语义、 主要风险（已提交 3 条评论）：(1) MODERATE—— 发现的问题中等
问题： 触发条件： 每次调用 实际影响： 长会话/大型工具集下每次请求前产生显著的 CPU 与内存开销（数 MB 载荷被重复 修正方向： 将一次请求的估算结果复用：每次 中等
问题： 触发条件： 多实例部署时，两个 worker 对同一会话并发处理：一方通过 before_model 压缩路径执行 实际影响： 事件级数据完整性受损——用户新输入或模型输出事件可能在压缩路径中丢失；由于 修正方向： 在 SQL 较低
问题： 触发条件： 每次 实际影响： 会话状态不断累积无效指纹数据（对 修正方向： 二选一：在写入 pending 指纹的事件持久化路径（如 |
| async with self._session_lock(session_id): | ||
| request.contents = [content.model_copy(deep=True) for content in request.contents] | ||
| state = await self._load_state(session_id) | ||
| reapplied = False | ||
| if state.latest_compaction is not None: | ||
| reapplied = self._apply_record(request, state.latest_compaction) | ||
|
|
||
| request_chars_before = estimate_request_chars(request) | ||
| token_budget_before = tracker.budget(request, ctx) | ||
| token_mode = token_budget_before.token_mode_enabled | ||
| request_tokens_before = token_budget_before.estimate.tokens | ||
| comparison_tokens_before = (tracker.estimate_request_tokens(request) if token_mode else None) | ||
| blocking_reached = (request_tokens_before >= token_budget_before.blocking_threshold_tokens | ||
| if token_mode else request_chars_before >= self._auto_compact_config.blocking_chars) | ||
| if state.consecutive_failures >= auto_compact_config.max_failures and blocking_reached: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=True, | ||
| source=None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) | ||
| if state.consecutive_failures >= auto_compact_config.max_failures: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=False, | ||
| source=None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) | ||
| auto_compact_reached = (request_tokens_before >= token_budget_before.auto_compact_threshold_tokens | ||
| if token_mode else request_chars_before >= auto_compact_config.trigger_chars) | ||
| if not force and not auto_compact_reached: | ||
| return AdvancedAutoCompactResult( | ||
| compacted=False, | ||
| reapplied=reapplied, | ||
| blocked=False, | ||
| source=state.latest_compaction.source if reapplied and state.latest_compaction else None, | ||
| request_chars_before=request_chars_before, | ||
| request_chars_after=request_chars_before, | ||
| consecutive_failures=state.consecutive_failures, | ||
| ) | ||
|
|
||
| original_contents = [content.model_copy(deep=True) for content in request.contents] | ||
| try: | ||
| compact_record: AdvancedAutoCompactRecord | None = None | ||
| if self._session_memory_extractor is not None: | ||
| await self._session_memory_extractor.extract_if_needed( | ||
| ctx, | ||
| force=True, | ||
| ) | ||
| session_memory = await self._latest_session_memory_record(ctx, ) | ||
| if session_memory is not None: | ||
| memory, boundary_signature, boundary_occurrence, boundary_event_id = session_memory | ||
| boundary_index = self._find_signature_index( | ||
| request.contents, | ||
| boundary_signature, | ||
| boundary_occurrence, | ||
| ) | ||
| if boundary_index is None and reapplied: | ||
| boundary_index = self._find_last_signature_index( | ||
| request.contents, | ||
| boundary_signature, | ||
| ) | ||
| if boundary_index is not None: | ||
| compact_record = self._compact_with_summary( | ||
| request, | ||
| summary=self._summary_with_recovery_path(memory), | ||
| boundary_index=boundary_index, | ||
| source="session-memory", | ||
| strict_boundary=True, | ||
| boundary_event_id=boundary_event_id, | ||
| ) | ||
| if token_mode: | ||
| target_reached = (tracker.budget(request, ctx).estimate.tokens | ||
| <= token_budget_before.warning_threshold_tokens) | ||
| else: | ||
| target_reached = estimate_request_chars(request) <= auto_compact_config.target_chars | ||
| if not target_reached: | ||
| request.contents = [content.model_copy(deep=True) for content in original_contents] | ||
| compact_record = None | ||
|
|
||
| if compact_record is None: | ||
| keep_count = min( | ||
| auto_compact_config.keep_recent_contents, | ||
| max(1, | ||
| len(request.contents) - 1), | ||
| ) | ||
| boundary_index = len(request.contents) - keep_count - 1 | ||
| if boundary_index < 0: | ||
| raise ValueError("Not enough model contents to compact") | ||
| summary = await self._legacy_summary( | ||
| request.contents[:boundary_index + 1], | ||
| ctx, | ||
| ) | ||
| compact_record = self._compact_with_summary( | ||
| request, | ||
| summary=self._summary_with_recovery_path(summary), | ||
| boundary_index=boundary_index, | ||
| source="legacy", | ||
| boundary_event_id=self._legacy_boundary_event_id(ctx), | ||
| ) | ||
|
|
||
| request_chars_after = estimate_request_chars(request) | ||
| if token_mode: | ||
| comparison_tokens_after = tracker.estimate_request_tokens(request) | ||
| if (comparison_tokens_after >= comparison_tokens_before | ||
| and request_chars_after >= request_chars_before): | ||
| raise ValueError("Advanced Auto Compact did not reduce request token estimate") |
There was a problem hiding this comment.
问题： _apply_scoped 在持锁状态下对请求进行多次全量深拷贝与全量序列化：request.contents 深拷贝发生在 644、690、726 行（最多 3 次），estimate_request_chars 在 650、750 行各做一次 request.model_dump(mode="python", by_alias=True, exclude_none=True) 加 json.dumps(sort_keys, separators, default=str) 全量序列化，estimate_request_tokens 在 654、752 行各做一次完整 model_dump，加上 tracker.budget（651 行）内部的估算以及 extract_if_needed(force=True)（694 行）中的内容估算，单次 before_model 压缩路径对一个包含全部工具定义、指令和长会话内容的大请求累计执行 6 次以上的全量序列化/深拷贝。
触发条件： 每次调用 AdvancedAutoCompactSummarizer.create_session_summary_by_request（经 AdvancedAutoCompactSummarizerFilter._before 在每次模型调用前触发）且自动压缩启用时都会完整走一遍；请求越大（工具多、会话长、指令多）放大越明显。
实际影响： 长会话/大型工具集下每次请求前产生显著的 CPU 与内存开销（数 MB 载荷被重复 json.dumps 和深拷贝 6 次以上），延长请求前处理耗时，并且全部发生在持锁范围内（_session_lock），会串行化同一会话的后续压缩并放大延迟。
修正方向： 将一次请求的估算结果复用：每次 _apply_scoped 只做一次 model_dump，estimate_request_chars 与 estimate_request_tokens 共用同一序列化结果；深拷贝仅在确实需要回滚快照时做一次（original_contents），锁内第 644 行的先行拷贝可在入口处用浅拷贝替代，待真正修改前再深拷贝，减少重复拷贝次数。
| changed = compact_events( | ||
| summary_event, | ||
| boundary_event_id, | ||
| compaction_id=compaction_id, | ||
| ) | ||
| if changed: | ||
| await ctx.session_service.update_session(ctx.session) |
There was a problem hiding this comment.
问题： _persist_session_compaction 在 compact_events 成功变更事件窗口后直接调用 ctx.session_service.update_session(ctx.session) 持久化。SQL 后端的 update_session（_sql_session_service.py:664）使用无行锁的普通 get 读取 StorageSession，随后删除该会话的全部 SessionStorageEvent 行并重新插入 session.events；而本次变更把 append_event（550 行）和 update_session_state（645 行）都改成了 get_for_update 行锁，唯独 update_session 保持无锁，形成防护不对称。
触发条件： 多实例部署时，两个 worker 对同一会话并发处理：一方通过 before_model 压缩路径执行 compact_events + update_session（全量删除/重写事件行），另一方同时 append_event 新增事件（带行锁）。全量删除的一方删除条件只按 app_name/user_id/session_id 匹配事件行，会把对方已提交的新事件行一并删除；或删除发生在对方插入之前，导致对方插入的行在后续全量重写中被覆盖。
实际影响： 事件级数据完整性受损——用户新输入或模型输出事件可能在压缩路径中丢失；由于 update_session 在删除事件行的同时还会用内存 session.state 整体替换存储 state，若内存对象已过期还会回退已由 update_session_state 行锁写好的持久化状态。
修正方向： 在 SQL update_session 中改用 get_for_update 读取 StorageSession（与 append_event/update_session_state 一致），使压缩重写与并发 append_event 之间互相串行；或让 _persist_session_compaction 改用按 compaction_id 幂等更新的增量接口（如事件级 upsert + boundary 标记），避免整窗删除重写。
| @classmethod | ||
| def record_request_context( | ||
| self, | ||
| request: "LlmRequest", | ||
| ctx: "InvocationContext | None", | ||
| cls, | ||
| request: LlmRequest, | ||
| ctx: InvocationContext, | ||
| ) -> None: | ||
| """Stage the final request fingerprint for persistence on the response Event.""" | ||
| session = getattr(ctx, "session", None) if ctx is not None else None | ||
| state = getattr(session, "state", None) | ||
| session = ctx.session | ||
| state = session.state | ||
| if isinstance(state, dict): | ||
| state["advanced_memory_pending_request_context_fingerprint"] = _request_static_fingerprint(request) | ||
|
|
There was a problem hiding this comment.
问题： TokenContextTracker.record_request_context 把 session.state["advanced_memory_pending_request_context_fingerprint"] 写入内存会话状态，但全代码库检索不到任何该键的读取者或清理者；同时 _latest_usage_baseline（178 行）读取的 advanced_memory_request_context_fingerprint 键全库没有任何写入点。二者构成一套永远无法触发的死状态：pending 指纹每轮 before_model 都被写入且永不消费/清理，而事件上的消费指纹永远不会存在。
触发条件： 每次 create_session_summary_by_request 非阻断路径（_auto_compact.py:581）都会执行 record_request_context，长会话中每轮对话都向 session.state 追加一个无消费者、无清理的格式为 JSON 字符串的键。
实际影响： 会话状态不断累积无效指纹数据（对 max_events/TTL 裁剪后的会话对象与持久化 JSON 均构成无意义增长）；_latest_usage_baseline 中的静态指纹校验分支因 recorded_fingerprint 永远为 None 而永不生效，token 匹配只依赖内容指纹，与设计意图不符，后续维护者难以分辨是未完成接线还是有意回退。
修正方向： 二选一：在写入 pending 指纹的事件持久化路径（如 append_event/usage metadata 落库处）真正把 advanced_memory_request_context_fingerprint 写入事件 custom_metadata，形成完整消费闭环；若该机制暂不启用，则移除 record_request_context 调用与 pending 指纹写入，避免死状态累积。
AI Code Review审查结论通过 对 f05797d..3c0f340（Feature/advanced memory service，109 文件，+6118/-6344）执行只读深度审查。逐段验证了 compact/advanced/ 压缩管线（AutoCompact、HistorySnip、MicroCompact、ToolResultBudget、TokenContextTracker）、过滤器框架（run_stream_filters、ModelCallbackFilter、internal_compaction_call）、长时记忆注入回调（LongTermMemoryContextCallback，stage 5）、SQL/Redis/文件存储与 compact_events 持久化路径、Runner 后置处理以及测试覆盖。确认 4 个由该变更引入的可操作问题：1) 发现的问题中等
问题： 触发条件： 启用 advanced auto-compact（ 实际影响： 1) 每次压缩内部调用都重复读取并可能重写记忆索引，在重试循环里放大成多次 I/O；2) 记忆索引与记忆维护指令进入压缩提示词，摘要模型会看到与任务无关的跨会话记忆内容，可能污染摘要/会话记忆提取结果（压缩本应只面对原始对话）；3) 索引指令占用的 token 计入每次内部请求，抬高成本与延迟；4) 修正方向： 在 较低
问题： 触发条件： 启用 token 模式（配置了 实际影响： 1) 设计意图的上下文一致性校验（响应事件指纹与本次请求静态指纹不一致则跳过该 usage 基线、回退全量估算）在生产中永远不会生效——基线始终按内容指纹匹配，可能把一个上下文已变化（如切换模型、更新指令/工具配置）的旧 usage 事件当作基线，导致 token 估算偏差，进而触发过晚或过早的压缩阈值判断；2) 修正方向： 在框架把响应 Event 落库/生成的路径上（如 Runner 的 中等
问题： 触发条件： 任意工具（含第三方 MCP 工具）在请求重建后复用同一 实际影响： 本应只是上下文预算处理跳过该条候选，却升级为整轮对话被 修正方向： 将同一请求内 id 复用冲突从抛异常改为按首次出现处理并跳过后续重复 id（或生成去重后缀），仅在跨请求的 较低
问题： 触发条件： 长会话中一次性有大量可剪枝的工具结果（ 实际影响： 每个被剪枝的工具结果都触发一次全请求重新序列化+token 重估算，随会话长度线性放大；该处理发生在每次模型调用前的 修正方向： 将 |
| request = LlmRequest() | ||
| request.append_instructions([LEGACY_SUMMARY_INSTRUCTION]) | ||
| prompt = ("Compress the following old conversation. The input may contain JSON representations " | ||
| "of tool calls and results:\n\n" | ||
| f"<history>\n{history}\n</history>") | ||
| request.contents.append(Content(role="user", parts=[Part.from_text(text=prompt)])) | ||
|
|
||
| output = "" | ||
| with internal_compaction_call(getattr(ctx, "agent_context", None)): | ||
| async for llm_response in self._resolve_model(ctx).generate_async( | ||
| request, | ||
| stream=False, | ||
| ctx=ctx, | ||
| ): | ||
| if llm_response.content and llm_response.content.parts: | ||
| for part in llm_response.content.parts: | ||
| if part.text: |
There was a problem hiding this comment.
问题： internal_compaction_call（_utils.py:22）只是在 ctx.agent_context 的 metadata 上设置 _trpc_agent_internal_compaction 标记，但全仓库没有任何消费方读取该标记（grep 仅命中定义与两处设置点）。LLMModel.generate_async 只要传入 ctx，就会把 ModelCallbackFilter（携带完整 before_model_callback 链）追加到 _run_stream_filters 的 extra_filters（_llm_model.py:156-159），而 LongTermMemoryContextCallback（stage 5，_memory_context.py:91）并不检查该标记。因此 _generate_summary（_auto_compact.py:142）和 SessionMemoryExtractor._call_llm_model（_compaction_memory_extractor.py:285）发起的每个内部 LLM 请求（含 summary_retries_count 与 max_retries+1 重试）都会命中该回调：先执行 read_index()（file 后端做目录 glob+剪枝+原子写回，redis 后端做 GET+逐主题 EXISTS+必要时 SET，sql 后端做查询+refresh+commit），再把完整 MEMORY.md 索引和主动维护记忆指令注入到本应是纯压缩/提取提示词的 system_instruction。
触发条件： 启用 advanced auto-compact（auto_compact.enabled）且启用长时记忆注入（long_term_memory_injection_enabled=True，默认值），并通过 setup_long_term_memory 安装了 stage-5 上下文回调；会话达到 trigger_chars/token 阈值触发任意一次压缩或会话记忆提取。
实际影响： 1) 每次压缩内部调用都重复读取并可能重写记忆索引，在重试循环里放大成多次 I/O；2) 记忆索引与记忆维护指令进入压缩提示词，摘要模型会看到与任务无关的跨会话记忆内容，可能污染摘要/会话记忆提取结果（压缩本应只面对原始对话）；3) 索引指令占用的 token 计入每次内部请求，抬高成本与延迟；4) runtime.initialize() 等副作用在每个压缩请求中重复执行。该标记机制本意就是阻止这种重入，但只挡住了 AdvancedAutoCompactSummarizerFilter 本身，未挡住其余回调链。
修正方向： 在 internal_compaction_call 的上下文管理器内，让 before_model_callback 链（至少 LongTermMemoryContextCallback）检查 ctx.agent_context.get_metadata(INTERNAL_COMPACTION_METADATA_KEY, False) 并直接跳过（或由 ModelCallbackFilter 在内部压缩时整体跳过 before_model_callback），保证内部压缩/提取请求不触发记忆注入与其余模型前置回调副作用。
| def record_request_context( | ||
| self, | ||
| request: "LlmRequest", | ||
| ctx: "InvocationContext | None", | ||
| cls, | ||
| request: LlmRequest, | ||
| ctx: InvocationContext, | ||
| ) -> None: | ||
| """Stage the final request fingerprint for persistence on the response Event.""" | ||
| session = getattr(ctx, "session", None) if ctx is not None else None | ||
| state = getattr(session, "state", None) | ||
| session = ctx.session | ||
| state = session.state | ||
| if isinstance(state, dict): | ||
| state["advanced_memory_pending_request_context_fingerprint"] = _request_static_fingerprint(request) |
There was a problem hiding this comment.
问题： record_request_context（_token_budget.py:242）把静态指纹写入 session.state["advanced_memory_pending_request_context_fingerprint"]，但在全仓库检索不到任何消费方：既没有代码把该临时值转写到响应事件的 custom_metadata["advanced_memory_request_context_fingerprint"]，也没有代码清理该 state 键；_latest_usage_baseline（_token_budget.py:178）读取的 advanced_memory_request_context_fingerprint 在真实事件上永远不会出现（唯一相关测试是手工构造的 SimpleNamespace）。
触发条件： 启用 token 模式（配置了 model_context_window_tokens）的 advanced auto-compact，任何一次 create_session_summary_by_request 都会执行 record_request_context；会话每次请求后该键都会被写入 session.state 并被持久化（SQL/InMemory 均持久化 session.state）。
实际影响： 1) 设计意图的上下文一致性校验（响应事件指纹与本次请求静态指纹不一致则跳过该 usage 基线、回退全量估算）在生产中永远不会生效——基线始终按内容指纹匹配，可能把一个上下文已变化（如切换模型、更新指令/工具配置）的旧 usage 事件当作基线，导致 token 估算偏差，进而触发过晚或过早的压缩阈值判断；2) advanced_memory_pending_request_context_fingerprint 成为一个永久残留且每次被覆盖为最新的陈旧 state 键，随会话 state 落库，污染持久化状态。
修正方向： 在框架把响应 Event 落库/生成的路径上（如 Runner 的 append_event 或模型的 _create_event_from_response）将 session.state 中的 pending 指纹转写到该 Event 的 custom_metadata["advanced_memory_request_context_fingerprint"] 并在写入后从 state 移除；若该机制无意启用，则应删除 record_request_context 及对应 state 写入，避免死代码与陈旧键。
| @@ -241,12 +229,11 @@ async def apply( | |||
| snipped_count = 0 | |||
| for candidate in eligible: | |||
| if not force and (current_tokens <= token_budget_before.warning_threshold_tokens | |||
| if token_mode else current_chars <= config.history_snip_target_chars): | |||
| if token_mode else current_chars <= history_snip_config.target_chars): | |||
| break | |||
| candidate_saving = max(0, candidate.original_size - replacement_size) | |||
| if candidate_saving == 0: | |||
| continue | |||
| await self._persist_snip(session_id, candidate, trigger) | |||
| candidate.part.function_response.response = self._snipped_response() | |||
| state.snipped_ids.add(candidate.result_id) | |||
| state.result_hashes[candidate.result_id] = candidate.original_sha256 | |||
| @@ -270,39 +257,20 @@ async def apply( | |||
| ) | |||
|
|
|||
|
|
|||
There was a problem hiding this comment.
问题： HistorySnip._apply_scoped 的剪枝循环（_history_snip.py:225-259）在每次删除一个候选后都调用 estimate_request_chars(request) 对完整请求做全量 JSON 序列化，并在 token 模式下再调用 tracker.budget(request, ctx) 重新做 usage 基线匹配与估算；_collect_candidates 本身又对每个 content/part 做 serialize_tool_response。删除 N 个候选即产生 O(N) 次整请求序列化，叠加一次遍历为 O(N²) 复杂度。
触发条件： 长会话中一次性有大量可剪枝的工具结果（allowed_tools 内的工具），上下文超过 trigger_chars/warning 阈值进入剪枝分支。
实际影响： 每个被剪枝的工具结果都触发一次全请求重新序列化+token 重估算，随会话长度线性放大；该处理发生在每次模型调用前的 _before 路径上，直接增加首 token 延迟，长会话下可达到数十至数百毫秒级额外开销。
修正方向： 将 current_chars 改为增量计算（仅在删除候选时减去该候选的 original_size - replacement_size 差值），token 估算改为仅在循环外做一次最终校验，避免每个循环迭代重新序列化完整请求。
3 participants
No description provided.