Repository navigation
Feature/advanced memory service - #335
CongkeChen wants to merge 7 commits into
Conversation
Please enter the commit message for your changes. Lines starting
AI Code Review审查结论不通过 审查范围：f05797d..debbecd（6 个提交，111 个文件，+6187/−6574 行），计划 Feature/advanced memory service。计划符合性：将 Advanced Memory 重构为 tools/advanced_memory/ 组件加 sessions/compact/（default+advanced）双 summarizer 管理器，新增组件、别名兼容与示例迁移基本符合计划。主要风险：本变更引入的高置信缺陷集中在 advanced 记忆链路 —— (1) 发现的问题严重
问题： 触发条件： 以 实际影响： 构造阶段立即抛出 修正方向： 改为 严重
问题： 触发条件： 会话第二次进入 实际影响： 每次成功 compaction 后的下一轮请求都会计入一次失败， 修正方向： replay 时对已摘要化的请求跳过再次 compaction 的尺寸校验（例如识别 严重
问题： 触发条件： 会话先完成一次 session-memory 提取（checkpoint 落在较早事件上），随后 advanced compaction 归档了该 checkpoint 事件；后续每次 实际影响： Session Memory 的增量游标永久停滞，之后的会话记忆不再更新（新记录与提取状态陈旧），同时每次提取都留下一条异常日志；该行为由本变更引入（旧实现以 修正方向： 归档 compaction 事件时保留边界映射（如给 严重
问题： 触发条件： 用户配置 实际影响： 过期记忆在 local 后端仍被当作有效长期记忆注入代理上下文，可能把陈旧/错误信息当作事实使用，且同一配置在 Redis/SQL 后端会正常过期删除、行为跨后端不一致。 修正方向： 在 中等
问题： 触发条件： 实际影响： 内存预算门控退化：每次请求都重新计算基线、无法跨请求保持稳定基线，预算控制与节流行为不可预期，且每次调用都丢弃历史、评估结果不稳定。 修正方向： 将 fingerprint 的写入方迁移到新的 advanced memory 会话服务/提取器写入路径（与 last_usage 同一写入点），或改为按会话 id 直接读取而不依赖 fingerprint。 中等
问题： 触发条件： 通过 实际影响： 正常触发压缩之后的下一次模型调用被该过滤器意外拦截/阻断，代理无法继续对话或任务执行，行为与设计意图（压缩后应放行模型调用）相悖。 修正方向： 在 |
| def __init__( | ||
| self, | ||
| config: AdvancedAutoCompactSummarizerConfig | None = None, | ||
| *, | ||
| model: LLMModel | None = None, | ||
| session_memory_extractor: SessionMemoryExtractor | None = None, | ||
| ) -> None: | ||
| """Initialize the compressor, summary generator, and session locks.""" | ||
| self._model = model | ||
| self._runtime = AdvancedAutoCompactSummarizerRuntime(config=config or AdvancedAutoCompactSummarizerConfig()) | ||
| self._auto_compact_config = config.auto_compact | ||
| self._session_memory_extractor = self._create_extractor(session_memory_extractor) | ||
| self._states: dict[str, AdvancedAutoCompactState] = {} | ||
| self._session_locks: dict[str, asyncio.Lock] = {} | ||
| self._scoped_processors: dict[object, "AdvancedAutoCompactSummarizer"] = {} | ||
|
|
There was a problem hiding this comment.
问题： AdvancedAutoCompactSummarizer.__init__ 将 config 声明为 config: AdvancedAutoCompactSummarizerConfig | None = None，但第 128 行 self._auto_compact_config = config.auto_compact 直接对可能为 None 的 config 解引用；config or AdvancedAutoCompactSummarizerConfig() 只用于 self._runtime。
触发条件： 以 AdvancedAutoCompactSummarizer(model=model) 或不传 config 的方式构造（公开签名与默认值均允许省略 config）。
实际影响： 构造阶段立即抛出 AttributeError: 'NoneType' object has no attribute 'auto_compact'，该新公开 API 在合法默认用法下直接不可用，迁移自旧 auto-summarizer 风格的调用方都会受影响。
修正方向： 改为 self._auto_compact_config = (config or AdvancedAutoCompactSummarizerConfig()).auto_compact，或在入口校验并抛出带指引的 ValueError。
| def _find_signature_index( | ||
| self, | ||
| contents: list[Content], | ||
| signature: str, | ||
| occurrence: int, | ||
| ) -> int | None: | ||
| """Locate a persisted compaction boundary by signature occurrence.""" | ||
| seen = 0 | ||
| for index, content in enumerate(contents): | ||
| if content_signature(contents[index]) == signature: | ||
| seen += 1 | ||
| if seen == occurrence: | ||
| return index | ||
| return None | ||
|
|
||
| def _find_last_signature_index( | ||
| self, | ||
| contents: list[Content], | ||
| signature: str, | ||
| ) -> int | None: | ||
| """Find the newest matching boundary after an earlier replay.""" | ||
| for index in range(len(contents) - 1, -1, -1): | ||
| if content_signature(contents[index]) == signature: | ||
| return index | ||
| return None | ||
|
|
||
| def _signature_occurrence( | ||
| self, | ||
| contents: list[Content], | ||
| signature: str, | ||
| boundary_index: int, | ||
| ) -> int: | ||
| """Count a boundary signature's occurrences from the request start.""" | ||
| return sum(1 for content in contents[:boundary_index + 1] if content_signature(content) == signature) | ||
|
|
||
| def _adjust_start_for_tool_pairing(self, contents: list[Content], start: int) -> int: | ||
| """Extend the retained range to keep calls paired with responses.""" | ||
| if start <= 0 or start >= len(contents): | ||
| return max(0, start) | ||
| response_ids = { | ||
| getattr(part.function_response, "id", None) | ||
| for content in contents[start:] | ||
| for part in content.parts or [] if part.function_response is not None | ||
| } | ||
| response_ids.discard(None) | ||
| if not response_ids: | ||
| return start | ||
| for index in range(start - 1, -1, -1): | ||
| call_ids = { | ||
| getattr(part.function_call, "id", None) | ||
| for part in contents[index].parts or [] if part.function_call is not None | ||
| } | ||
| if call_ids & response_ids: | ||
| start = index | ||
| response_ids -= call_ids | ||
| if not response_ids: | ||
| break | ||
| return start | ||
|
|
||
| def _compaction_start(self, contents: list[Content], boundary_index: int) -> int: | ||
| """Return the retained-content start for a legacy compaction.""" | ||
| start = min( | ||
| boundary_index + 1, | ||
| len(contents) - self._auto_compact_config.keep_recent_contents, | ||
| ) | ||
| return self._adjust_start_for_tool_pairing(contents, start) | ||
|
|
||
| def _session_memory_compaction_start(self, boundary_index: int) -> int: | ||
| """Drop everything through the session-memory checkpoint boundary.""" | ||
| return boundary_index + 1 | ||
|
|
||
| def _apply_record(self, request: LlmRequest, record: AdvancedAutoCompactRecord) -> bool: | ||
| """Replay a persisted compaction record into a rebuilt request.""" | ||
| boundary_index = self._find_signature_index( | ||
| request.contents, | ||
| record.boundary_signature, | ||
| record.boundary_occurrence, | ||
| ) | ||
| if boundary_index is None: | ||
| return False | ||
| start = (self._session_memory_compaction_start(boundary_index) | ||
| if record.source == "session-memory" else self._compaction_start(request.contents, boundary_index)) | ||
| request.contents = [ | ||
| self._summary_content(record.summary), | ||
| *request.contents[start:], | ||
| ] | ||
| return True |
There was a problem hiding this comment.
问题： _apply_record 用 _find_signature_index 按 boundary_occurrence 精确匹配旧 compaction 记录，但 session-memory 分支的 _persist_session_compaction 经 compact_events 已把 boundary 事件本身替换为 summary 事件，active 窗口与后续模型请求中不再存在该 boundary 签名；下次请求 _apply_record 必然返回 False（签名仅在第一次请求中可匹配一次）。
触发条件： 会话第二次进入 _apply_scoped 且 state.latest_compaction 非空（source 为 session-memory 或 legacy）；此时 request 已含摘要前缀，reapplied=False，_compact_with_summary 只替换摘要之后的少量内容，request_chars_after >= request_chars_before 触发 raise ValueError("Advanced Auto Compact did not reduce request size")。
实际影响： 每次成功 compaction 后的下一轮请求都会计入一次失败，consecutive_failures 递增，达到 max_failures 后会话被 ADVANCED_AUTOCOMPACT_BLOCKED_MESSAGE 硬阻断（阻断消息经模型过滤器直接返回给用户），正常对话流程被永久打断，且阻断状态仅保存在进程内 _states。
修正方向： replay 时对已摘要化的请求跳过再次 compaction 的尺寸校验（例如识别 reapplied==True 且请求前缀已是摘要时直接返回），并让 _find_signature_index 在匹配失败时退回 _find_last_signature_index/摘要锚点而非计入失败。
| @@ -634,39 +649,62 @@ async def _persist_checkpoint( | |||
| document.key_results, | |||
| document.worklog, | |||
| ) | |||
| await self._runtime.transcripts.append_unique( | |||
| session_id, | |||
| { | |||
| "schema_version": SESSION_MEMORY_CHECKPOINT_SCHEMA_VERSION, | |||
| "kind": "session-memory-checkpoint", | |||
| "checkpoint_id": f"session-memory:{last_event_id}", | |||
| "first_event_id": first_event_id, | |||
| "last_event_id": last_event_id, | |||
| "processed_events": len(included_records), | |||
| "non_empty_sections": sum(1 for value in values if value.strip()), | |||
| "session_memory_chars": len(document.to_markdown()), | |||
| "context_tokens": context_tokens, | |||
| }, | |||
| unique_key="checkpoint_id", | |||
| boundary = self._boundary_for_event(session, last_event_id) | |||
| if boundary is None: | |||
| raise ValueError(f"Session Memory boundary Event {last_event_id} has no visible content") | |||
| signature, occurrence = boundary | |||
| checkpoint = { | |||
| "first_event_id": first_event_id, | |||
| "last_event_id": last_event_id, | |||
| "recorded_at": included_records[-1].get("recorded_at"), | |||
| "last_event_timestamp": included_records[-1].get("event", {}).get("timestamp"), | |||
| "boundary_signature": signature, | |||
| "boundary_occurrence": occurrence, | |||
| "processed_events": len(included_records), | |||
| "non_empty_sections": sum(1 for value in values if value.strip()), | |||
| "updated_at": datetime.now(timezone.utc).isoformat(), | |||
| } | |||
| payload = build_session_memory_state( | |||
| document, | |||
| checkpoint=checkpoint, | |||
| context_tokens=context_tokens, | |||
| ) | |||
| state_delta = {SESSION_MEMORY_STATE_KEY: payload} | |||
| session_service = getattr(ctx, "session_service", None) | |||
| if session_service is None: | |||
| session.state.update(state_delta) | |||
| return | |||
There was a problem hiding this comment.
问题： SessionMemoryExtractor._persist_checkpoint 通过 _boundary_for_event 将 checkpoint 的 last_event_id 反查为 (boundary_signature, boundary_occurrence)，当 checkpoint 边界事件已被 compact_events 移入 historical_events 后（advanced compaction 正是把旧事件归档的手段），该事件在 active 窗口不可见，_boundary_for_event 返回 None 并抛出 ValueError；_event_records_after_checkpoint 的 timestamp 恢复分支返回的新事件又会被再次 _persist_checkpoint 依赖同一丢失的边界。
触发条件： 会话先完成一次 session-memory 提取（checkpoint 落在较早事件上），随后 advanced compaction 归档了该 checkpoint 事件；后续每次 extract_if_needed 都因边界不可解析而抛错（被 catch 仅记日志），checkpoint 永不前移。
实际影响： Session Memory 的增量游标永久停滞，之后的会话记忆不再更新（新记录与提取状态陈旧），同时每次提取都留下一条异常日志；该行为由本变更引入（旧实现以 last_event_id 直接比较、不依赖事件仍然可见）。
修正方向： 归档 compaction 事件时保留边界映射（如给 compact_events 增加返回被归档边界的元数据，或在 checkpoint 中直接持久化 boundary_signature/boundary_occurrence），使 _boundary_for_event 在事件不可见时也能解析 checkpoint。
| async def read_topic(self, topic_name: str) -> str | None: | ||
| path = self._paths.memory_topic_path(topic_name) | ||
| return await asyncio.to_thread(lambda: path.read_text(encoding=self._config.encoding) | ||
| if path.exists() else None) |
There was a problem hiding this comment.
问题： LongTermMemoryStore.read_topic（local 文件后端）直接 path.read_text 返回文件内容，完全不检查 memory_ttl_seconds 过期；而 Redis（_redis_stores.py）与 SQL（_sql_stores.py）后端在创建时都进行 TTL 过期检查，同一 read_topic 入口因后端不同行为不一致。
触发条件： 用户配置 memory_ttl_seconds 且以 StorageProvider.LOCAL 读取 topic 文件。
实际影响： 过期记忆在 local 后端仍被当作有效长期记忆注入代理上下文，可能把陈旧/错误信息当作事实使用，且同一配置在 Redis/SQL 后端会正常过期删除、行为跨后端不一致。
修正方向： 在 read_topic 返回前对照 memory_ttl_seconds 校验文件修改时间（或读取文件头部的 updated_at），过期时删除文件并返回 None，与 Redis/SQL 后端语义对齐。
| 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.
问题： _latest_usage_baseline 读取的 advanced_memory_request_context_fingerprint（写入方是旧 AdvancedMemorySessionService，本变更将其重构/删除后无任何代码再写入该 key），因此 fingerprint 与 last_usage 序列永远不会来自同一次写入，baseline 永久不命中，函数每次都会重建 RequestMetrics 并重新评估整个历史调用序列。
触发条件： advanced_memory 被启用且 metrics_threshold/baseline_window 类门控条件配置了需要 _latest_usage_baseline 生效的预算管理。
实际影响： 内存预算门控退化：每次请求都重新计算基线、无法跨请求保持稳定基线，预算控制与节流行为不可预期，且每次调用都丢弃历史、评估结果不稳定。
修正方向： 将 fingerprint 的写入方迁移到新的 advanced memory 会话服务/提取器写入路径（与 last_usage 同一写入点），或改为按会话 id 直接读取而不依赖 fingerprint。
|
|
||
| @override | ||
| async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult): | ||
| """Run the advanced auto compact summarizer filter.""" | ||
| if ctx.get_metadata(INTERNAL_COMPACTION_METADATA_KEY, False): | ||
| return None | ||
| invocation_ctx: InvocationContext = get_invocation_ctx() | ||
| summarizer_manager = self.get_summarizer_manager(invocation_ctx) | ||
| result = await summarizer_manager.create_session_summary_before_model( | ||
| req, | ||
| invocation_ctx, | ||
| ) | ||
| if not result: | ||
| return None | ||
| invocation_ctx.end_invocation = True | ||
| rsp.rsp = result | ||
| rsp.is_continue = False | ||
| rsp.error = None | ||
| session_memory_extractor = summarizer_manager.get_session_memory_extractor() | ||
| if session_memory_extractor is not None: | ||
| await session_memory_extractor.extract_if_needed( |
There was a problem hiding this comment.
问题： AdvancedAutoCompactSummarizerFilter._before 在压缩返回非空结果时设置 invocation_ctx.end_invocation = True 并短路模型调用，但没有设置 start_invocation；全仓库（含 _base.py 的断言与调用方过滤链）要求 start_invocation and end_invocation 校验通过才放行，短路路径使校验不通过，后续调用被过滤链继续拦截。
触发条件： 通过 AdvancedAutoCompactSummarizerFilter 且本次请求触发了压缩过滤链（压缩结果非空时短路）。
实际影响： 正常触发压缩之后的下一次模型调用被该过滤器意外拦截/阻断，代理无法继续对话或任务执行，行为与设计意图（压缩后应放行模型调用）相悖。
修正方向： 在 _before 里与 end_invocation 同时设置 start_invocation = True（或让短路不走过滤器、直接返回未设置状态），保证校验语义一致。
debbecd to
c4504e9
Compare
AI Code Review审查结论通过 审查综述范围与计划符合性： 审查覆盖 f05797d..debbecd（Feature/advanced memory service，111 文件 +6187/-6574，23 个测试文件，0 生成文件）。变更将旧 advanced_memory/* 代码迁移重构为 sessions/compact/{default,advanced}/ 与 tools/advanced_memory/，实现 Session 压缩双轨制（默认 AFTER_TURN 摘要 + 高级 BEFORE_MODEL 自动压缩）、Session Memory 提取、long-term memory 三存储后端（local/Redis/SQL）与请求指纹机制。整体结构与计划一致：compact_events、update_session_state、get_for_update、store_historical_events 等新接口均有配套实现与示例；旧 SessionSummarizer/SessionSummary 等以别名保留，ClawSummarizerSessionManager 兼容（继承别名且默认 AFTER_TURN）。 主要风险（由本次变更引入）：
测试充分性： 23 个测试文件覆盖 token budget、auto-compact、history snip、提取器、三个存储后端与默认压缩器；test_token_budget.py 手工写入 custom_metadata 恰好反向印证指纹断链（无自动写入方）；尚无针对 get_session_summary 注入体积、熔断跨重启持久化、conversation_count 失败重试语义、historical_events 默认配置下可恢复性的测试。 门禁结论： 无 SEVERE 级缺陷；存在 3 个 MODERATE（指纹断链、Session Memory 全量注入、历史事件不可恢复）与 6 个 LOW，均不影响主流程正确性但有真实成本/数据可追溯性影响，建议合入前优先修复 MODERATE 项。审查方式为静态只读分析（git 范围 diff + 调用链追踪 + 全局 grep），未运行测试（环境无法导入 pydantic 依赖），评论均锚定到本次变更的变更后行。 发现的问题中等
问题： 请求指纹写读断链： 触发条件： 启用 token 模式（ 实际影响： usage-baseline 退化为仅按内容指纹匹配，把不同上下文（不同 system 指令、不同历史）但同内容的请求错误归并为同一基线， 修正方向： 在响应 Event 持久化时把 state 中暂存的指纹 stamp 到 中等
问题： advanced 压缩器通过既有摘要 API 把 Session Memory 全文注入每次模型请求： 触发条件： 使用 实际影响： 与压缩后的 summary event、advanced-memory index 注入内容三重重叠，长会话下上下文窗口与 token 计费随请求数线性放大，且覆盖了原有 summary 的"简要历史"语义；全量记忆注入还引入与模型响应再压缩之间的循环干扰。 修正方向： 对 Session Memory 做截断/分层（如仅注入 index 或截取最相关 section），或在请求内容已含 memory 信息时跳过注入。 中等
问题： 本次新增的 触发条件： 使用 实际影响： 默认配置下压缩换出的原始事件无法恢复，与 SQL/Redis（强制 修正方向： 让 较低
问题： 触发条件： 会话进入自动压缩窗口（char 模式 ≥700k）后的每个请求，包括 实际影响： 每次压缩尝试都触发一轮 Session Memory 提取 LLM 调用（绕过阈值），提取失败后每轮请求重复触发，token 成本与延迟随失败次数线性增长，与 修正方向： 仅在检查点有新增事件时才用 较低
问题： 熔断状态仅存进程内存： 触发条件： 进程重启 / 进程内状态重置后，同一会话继续运行并再次达到 实际影响： 重启后 修正方向： 将 较低
问题： 工具结果剪枝状态仅存进程内存： 触发条件： 进程重启后同一会话再次请求（tool 结果 id 仍相同，如相同工具调用重放）并超过 snip 阈值。 实际影响： 重启后已剪枝的大段 tool 结果（可达数十万字符）被重新完整注入模型上下文，上下文窗口压力骤增、token 计费放大； 修正方向： 将 较低
问题： 默认摘要器的轮次计数语义被改变：base 版本（ 触发条件： 默认摘要器（AFTER_TURN）会话中每次总结失败（LLM 出错、摘要为空或 实际影响： 总结失败时计数不再归零，行为从"每 N 轮总结一次"退化为"失败后每轮都重试总结"，失败的场景下每轮触发一次昂贵的摘要 LLM 调用；计数语义与历史行为不兼容。 修正方向： 在 较低
问题： 触发条件： 外部调用方在服务方法返回后继续读取同一 ORM 对象（如 实际影响： 读到 stale 属性而非检测到过期重载：并发更新下读到旧 state/timestamp， 修正方向： 在每个改写路径统一 较低
问题： Redis 读路径逐 key 网络往返： 触发条件： 用户记忆条数 N 较大（数十/上百条）时每次 实际影响： 记忆读取热路径（每轮注入 advanced-memory index 前）延迟随记忆规模线性放大，长会话用户延迟明显，且每个请求都重复刷新 TTL，放大负载。 修正方向： 用 |
| 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-251）只把指纹写入 session.state["advanced_memory_pending_request_context_fingerprint"]，而 _latest_usage_baseline 读取的是 Event custom_metadata 中的 advanced_memory_request_context_fingerprint。全仓 grep 无任何自动写入方（旧 advanced_memory/_session_service.py 的 stamp 逻辑随本次迁移删除），仅测试文件手工写入该字段。
触发条件： 启用 token 模式（TokenContextTrackerConfig.enabled=True 且设置 model_context_window_tokens）后发起请求；凡是请求内容与历史 usage event 内容指纹相同（如重复提问、相同 tool 结果）的请求都会命中基线匹配。
实际影响： usage-baseline 退化为仅按内容指纹匹配，把不同上下文（不同 system 指令、不同历史）但同内容的请求错误归并为同一基线，ContextTokenEstimate 的 token 预算偏差，进而影响 warning/blocking 阈值判断：上下文被低估时可能越过阻塞阈值引发熔断误触发，或高估时压缩不及时。
修正方向： 在响应 Event 持久化时把 state 中暂存的指纹 stamp 到 event.custom_metadata["advanced_memory_request_context_fingerprint"]（恢复旧 _session_service.py 行为），或在 _latest_usage_baseline 中读取 state 暂存值。
| async def get_session_summary(self, session: Session) -> str | None: | ||
| """Read compact Session Memory through the existing summary API.""" | ||
| parsed = parse_session_memory_state(session.state.get(SESSION_MEMORY_STATE_KEY)) | ||
| if parsed is not None: | ||
| return parsed[0].to_markdown() | ||
| return None |
There was a problem hiding this comment.
问题： advanced 压缩器通过既有摘要 API 把 Session Memory 全文注入每次模型请求：get_session_summary（_manager.py:96-101）返回 session.state[SESSION_MEMORY_STATE_KEY] 的全部 section markdown（to_markdown 连空 section 也会渲染），而调用链上 _request_processor 每次请求都会把 get_session_summary 的结果拼进 system instructions。
触发条件： 使用 AdvancedAutoCompactSummarizerManager（BEFORE_MODEL）且提取器已产出过 Session Memory；每个请求（含 tool 调用后的再次请求）都会把整份 700k 字符量级的 memory 全文拼入 instructions。
实际影响： 与压缩后的 summary event、advanced-memory index 注入内容三重重叠，长会话下上下文窗口与 token 计费随请求数线性放大，且覆盖了原有 summary 的"简要历史"语义；全量记忆注入还引入与模型响应再压缩之间的循环干扰。
修正方向： 对 Session Memory 做截断/分层（如仅注入 index 或截取最相关 section），或在请求内容已含 memory 信息时跳过注入。
| @override | ||
| async def update_session_state( | ||
| self, | ||
| session: Session, | ||
| state_delta: dict[str, Any], | ||
| ) -> None: | ||
| """Patch stored session state without replacing its Event window.""" | ||
| if not state_delta: | ||
| return | ||
| session.state.update(state_delta) | ||
|
|
||
| app_sessions = self._sessions.get(session.app_name) | ||
| user_sessions = app_sessions.get(session.user_id) if app_sessions else None | ||
| stored = user_sessions.get(session.id) if user_sessions else None | ||
| if stored is None: | ||
| logger.warning("Session %s not found while updating state", session.id) | ||
| return | ||
| stored.session.state.update(state_delta) | ||
| stored.ttl.update_expired_at() | ||
|
|
There was a problem hiding this comment.
问题： 本次新增的 update_session_state（_in_memory_session_service.py:272-291）只 patch stored.session.state，未更新 stored.session.events/historical_events/conversation_count，而 compact_events 换入的原始事件依赖 historical_events 持久化；且服务内仍存在的 _set_session 在默认 store_historical_events=False 下每次写入清零 historical_events（update_session 亦经 _set_session），与本次示例 README 承诺"被替换的原始 events 移入 historical_events/原始事件仍可完整追溯"矛盾。
触发条件： 使用 InMemorySessionService 且未显式设置 SessionServiceConfig(store_historical_events=True)（examples/session_summarizer_with_advanced/run_agent.py 正是此默认配置）时触发一次压缩换出。
实际影响： 默认配置下压缩换出的原始事件无法恢复，与 SQL/Redis（强制 store_historical_events=True）后端行为不一致，用户无法追溯压缩前原始对话细节；同时 update_session_state 与 update_session 的语义分裂会让 state 与 events 更新路径不一致。
修正方向： 让 _set_session 在存储时保留已存在的 historical_events（仅初始化时空置），或使 update_session_state 与新压缩链路同步持久化 historical_events。
| 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 在每次压缩尝试时无条件执行 extract_if_needed(ctx, force=True)，force=True 会跳过 extractor 的 threshold/tool_condition 检查（_compaction_memory_extractor.py 中 if not force and (...) 分支），只保留 no-new-events/unsafe-boundary/coordination-timeout 守卫。
触发条件： 会话进入自动压缩窗口（char 模式 ≥700k）后的每个请求，包括 consecutive_failures 已超过 max_failures 的失败重试期（此时压缩尝试被提前返回，但 extraction 仍在循环内被执行）。
实际影响： 每次压缩尝试都触发一轮 Session Memory 提取 LLM 调用（绕过阈值），提取失败后每轮请求重复触发，token 成本与延迟随失败次数线性增长，与 max_failures 熔断设计的初衷（失败后停止重试）相悖。
修正方向： 仅在检查点有新增事件时才用 force=True 调用 LLM（或第一轮用 force=False、由 extraction 结果决定），熔断/无新事件时跳过 extraction。
| async def _load_state(self, session_id: str) -> AdvancedAutoCompactState: | ||
| """Restore process-local compaction state.""" | ||
| state_key = self._runtime.session_key(session_id) | ||
| state = self._states.get(state_key) | ||
| if state is not None: | ||
| return state | ||
| state = AdvancedAutoCompactState(latest_compaction=None, consecutive_failures=0) |
There was a problem hiding this comment.
问题： 熔断状态仅存进程内存：_load_state（_auto_compact.py:210-216）返回的 AdvancedAutoCompactState(latest_compaction, consecutive_failures) 只保存在 self._states 字典，从不持久化；_apply_record 重放与 consecutive_failures 计数均依赖该状态。
触发条件： 进程重启 / 进程内状态重置后，同一会话继续运行并再次达到 blocking_chars（默认 780k）。
实际影响： 重启后 consecutive_failures 归零、latest_compaction 丢失，之前连续失败的会话不再被 blocked=True 熔断保护，阻塞消息与重复压缩噪音重现，高成本压缩重试再次触发。
修正方向： 将 latest_compaction/consecutive_failures 持久化到 session.state，重启后从 state 恢复。
| async def _load_state(self, session_id: str) -> HistorySnipState: | ||
| """Restore prior history-snip decisions from the transcript.""" | ||
| state = self._states.get(session_id) | ||
| """Return process-local history-snip state.""" | ||
| state_key = self._runtime.session_key(session_id) if hasattr(self._runtime, "session_key") else session_id | ||
| state = self._states.get(state_key) | ||
| if state is not None: | ||
| return state | ||
| records = await self._runtime.transcripts.read_all(session_id) | ||
| snipped_ids: set[str] = set() | ||
| result_hashes: dict[str, str] = {} | ||
| for record in records: | ||
| result_id = record.get("result_id") | ||
| if record.get("kind") != "history-snip" or not isinstance(result_id, str): | ||
| continue | ||
| snipped_ids.add(result_id) | ||
| original_sha256 = record.get("original_sha256") | ||
| if isinstance(original_sha256, str): | ||
| result_hashes[result_id] = original_sha256 | ||
| state = HistorySnipState( | ||
| snipped_ids=snipped_ids, | ||
| result_hashes=result_hashes, | ||
| snipped_ids=set(), | ||
| result_hashes={}, | ||
| ) | ||
| self._states[session_id] = state | ||
| self._states[state_key] = state | ||
| return state |
There was a problem hiding this comment.
问题： 工具结果剪枝状态仅存进程内存：HistorySnipState(snipped_ids, result_hashes)（_history_snip.py:106-117）与 MicroCompactState.cleared_ids（_micro_compact.py:94-104）都只存于进程内 self._states 字典。
触发条件： 进程重启后同一会话再次请求（tool 结果 id 仍相同，如相同工具调用重放）并超过 snip 阈值。
实际影响： 重启后已剪枝的大段 tool 结果（可达数十万字符）被重新完整注入模型上下文，上下文窗口压力骤增、token 计费放大；result_hashes 丢失同时使"id 复用不同内容"防篡改校验无法跨重启生效。
修正方向： 将 snipped_ids/cleared_ids/result_hashes 持久化（写入 session.state 或 memory 索引），重启后恢复；或给占位响应打上可识别标记使重放可自愈。
|
|
||
| def _decorator(session: Session) -> bool: | ||
| if session.conversation_count > conversation_count: | ||
| session.conversation_count = 0 | ||
| return True | ||
| return False | ||
| return session.conversation_count > conversation_count | ||
|
|
There was a problem hiding this comment.
问题： 默认摘要器的轮次计数语义被改变：base 版本（f05797d9 的 sessions/_summarizer_checker.py:140-141）在 set_summarizer_conversation_threshold 命中阈值时执行 session.conversation_count = 0 后再返回 True，而本次迁移后的 _checker.py:140-141 只返回谓词结果不再重置；重置被移入 _summarizer_manager.py:118 且仅在 create_session_summary 成功产出摘要后执行。
触发条件： 默认摘要器（AFTER_TURN）会话中每次总结失败（LLM 出错、摘要为空或 should_summarize 之外的异常路径）。
实际影响： 总结失败时计数不再归零，行为从"每 N 轮总结一次"退化为"失败后每轮都重试总结"，失败的场景下每轮触发一次昂贵的摘要 LLM 调用；计数语义与历史行为不兼容。
修正方向： 在 should_summarize_session 判定后即重置计数（无论成败），或保留 threshold 谓词内的重置语义。
| # AsyncSession cannot perform an implicit refresh when an ORM | ||
| # attribute is accessed after commit. Keep committed values available | ||
| # because this service reads StorageSession state after committing. | ||
| kwargs.setdefault("expire_on_commit", False) |
There was a problem hiding this comment.
问题： SqlSessionService.__init__ 全局 kwargs.setdefault("expire_on_commit", False)（_sql_session_service.py:406）关闭了 ORM 每事务自动过期，但代码只在 append_event/update_session_state/_get_session/list 等少数内部路径补了 refresh()（约 783 行注释亦承认该约束），其余读写路径未覆盖。
触发条件： 外部调用方在服务方法返回后继续读取同一 ORM 对象（如 update_session 后读 storage_session.state、或经 _get_storage_session 复用对象），期间另一并发事务已提交新值。
实际影响： 读到 stale 属性而非检测到过期重载：并发更新下读到旧 state/timestamp，get_for_update 加锁路径的 stale 检测依赖调用方自行传对对象，容易漏掉更新。
修正方向： 在每个改写路径统一 refresh()，或在关键读路径显式 expire()/查询时 populate_existing。
| ttl: int | None, | ||
| skip_prefixes: tuple[str, ...] = (), | ||
| ) -> None: | ||
| """Track and refresh every key in one logical memory group.""" | ||
| if ttl is None: | ||
| return | ||
| if keys: | ||
| await self._command("sadd", registry, *keys) | ||
| tracked = await self._command("smembers", registry) or [] | ||
| tracked_keys = {self._text(value) for value in tracked} | ||
| tracked_keys.update(keys) | ||
| for key in tracked_keys: | ||
| if key and not key.startswith(skip_prefixes): | ||
| await self._command("expire", key, ttl) | ||
| await self._command("expire", registry, ttl) | ||
|
|
||
| async def _refresh_memory_ttl(self, *keys: str) -> None: |
There was a problem hiding this comment.
问题： Redis 读路径逐 key 网络往返：_refresh_ttl_group（_redis_stores.py:100-116）对组内每个 key 单独发 expire 命令（N+1 往返），read_index（_redis_stores.py:137-157）对索引逐条目 exists 检查再 prune。
触发条件： 用户记忆条数 N 较大（数十/上百条）时每次 read_index/read_topic/topic 列表刷新都触发 O(N)~O(N²) 次串行 Redis 往返；且本次实现未使用 pipeline。
实际影响： 记忆读取热路径（每轮注入 advanced-memory index 前）延迟随记忆规模线性放大，长会话用户延迟明显，且每个请求都重复刷新 TTL，放大负载。
修正方向： 用 pipeline 合并 expire/exists 批处理，或仅在写入路径维护 TTL、读路径用一次 EXISTS 查询剪枝。
AI Code Review审查结论不通过 本次审查覆盖 base_commit f05797d..head_commit c4504e9（计划：Feature/advanced memory service，111 个文件、+6182/-6578），重点检查 advanced memory 重构（advanced_memory 包拆分为 tools/advanced_memory 与 sessions/compact/advanced）及 Session Compact 的默认/高级两条路径。 审查方法：直接阅读变更前后代码对比（git show/diff）验证每个候选的触发条件与实际影响，并交叉核对各 finder 代理的输出，排除伪报（Redis list_topics 空、ToolResultBudget 重启后误抛 ValueError、SQL update_session_state 锁覆盖不足、circuit-breaker 死分支等均已被证据否定）。 核心结论：
测试影响：base 的 tests/advanced_memory/（454+516+334 行覆盖 circuit-breaker、重启重放、checkpoint 恢复、路径穿越防护）被删除，新 tests/sessions/compact/test_session_compact.py（268 行）未覆盖这些行为，重启重放与熔断回归无测试保护。 门禁结论：FAILED——存在已验证的数据丢失高风险缺陷（跨重启重复压缩），应予修复后合入。 发现的问题严重
问题： 新实现删除了 base 版本通过 transcript 持久化 compaction 状态的能力：base 的 触发条件： 进程重启/崩溃/多实例切换后继续同一 session： 实际影响： (a) 下一轮 修正方向： 恢复持久化恢复机制：将 compaction 成功/失败记录（boundary signature/occurrence、summary、compaction_id、失败计数）写入持久化会话状态（例如 中等
问题： 触发条件： 任何跨对象/跨进程的重复压缩场景（一次 实际影响： “使重试幂等”的文档承诺无法兑现；配合状态不持久化缺陷，重试/重建场景会对同一个前缀执行两次压缩，产生嵌套 summary 并丢失中间细节。 修正方向： 让 中等
问题： session-memory 提取的事件来源从 base 的 append-only transcript（含全部历史事件）改为仅构造 触发条件： BEFORE_MODEL 路径中 实际影响： checkpoint 之后、compaction 之前尚未提取的事件随 compaction 进入 historical_events 后永远无法被提取，会话记忆永久遗漏该段信息； 修正方向： 提取的增量边界不应依赖易失的活跃窗口：将 checkpoint（last_event_id 等）持久化到独立于 events 重排的源（如 session.state 中不与 compaction 冲突的键），并让 中等
问题： 触发条件： 会话轮数超过阈值后，summarize 失败（LLM 调用异常、返回空摘要或 实际影响： 每次失败后的每一轮对话都会重复触发一次 summarize LLM 调用（base 版本失败后计数被 checker 清零，需重新累积到阈值才再次尝试），模型持续故障时产生无界的重复调用与日志噪音。 修正方向： 在 manager 的 summarize 失败路径（ 中等
问题： 触发条件： 实际影响： 同一 session 的两次压缩/写操作并发执行， 修正方向： 与 较低
问题： 触发条件： 每次模型请求都会向 session.state 写入该键并随 state 持久化； 实际影响： (a) usage 基线可能匹配到模型/系统指令/工具配置不同时期的旧 usage 事件，token 预算估计偏差，导致提前或延迟触发压缩；(b) 修正方向： 在事件落库时把 pending fingerprint 写入事件 较低
问题： 本次变更给 触发条件： 多实例/多事件循环下，一个请求的 compaction 执行 实际影响： 事务交错时并发 append 的新事件行可能被 compaction 的 delete-all 删除，或 compaction 写入的窗口覆盖并发 append 的事件，活动窗口静默丢失事件（base 版本 advanced compact 只写 transcript、不重写 events，此写路径为本变更新引入）。 修正方向： |
| async def _load_state(self, session_id: str) -> AdvancedAutoCompactState: | ||
| """Restore process-local compaction state.""" | ||
| state_key = self._runtime.session_key(session_id) | ||
| state = self._states.get(state_key) | ||
| if state is not None: | ||
| return state | ||
| state = AdvancedAutoCompactState(latest_compaction=None, consecutive_failures=0) | ||
| self._states[state_key] = state | ||
| return state |
There was a problem hiding this comment.
问题： 新实现删除了 base 版本通过 transcript 持久化 compaction 状态的能力：base 的 AdvancedAutoCompact._load_state 从转录记录恢复 latest_compaction 与 consecutive_failures（_persist_success/_persist_failure 写盘），新代码 _load_state 仅查进程内 self._states，两次 persist 相关方法与 transcript 读写均被移除。
触发条件： 进程重启/崩溃/多实例切换后继续同一 session：state.latest_compaction 为空且无持久化记录可恢复；携带 session_compaction_id 的 summary 事件虽已持久化在 events 中，但不会被识别为已 compact。
实际影响： (a) 下一轮 _apply_scoped 无记录可重放（_apply_record 返回 False），基于已含 summary 的恢复窗口再次执行 compaction——对已摘要内容二次压缩，嵌套 summary，历史细节永久丢失；(b) consecutive_failures 归零，circuit-breaker 熔断保护失效，模型上下文超限的请求会被直接发给模型导致调用失败。
修正方向： 恢复持久化恢复机制：将 compaction 成功/失败记录（boundary signature/occurrence、summary、compaction_id、失败计数）写入持久化会话状态（例如 session.state 下的版本化键，或恢复 transcript 写入），_load_state 在 _states 未命中时从该持久化源重建。
|
|
||
| boundary_index = next( | ||
| (index for index, event in enumerate(self.events) if event.id == boundary_event_id), | ||
| None, | ||
| ) | ||
| if boundary_index is None: | ||
| raise ValueError(f"Session compaction boundary Event {boundary_event_id!r} " |
There was a problem hiding this comment.
问题： Session.compact_events 的幂等保护是死代码：_compact_with_summary 每次生成新 compaction_id = f"advanced_autocompact:{uuid.uuid4().hex}"，_persist_session_compaction 中 record.compaction_id or 新 uuid 恒取前者，因此每次压缩携带全新 id，compact_events 中 metadata.get("session_compaction_id") == compaction_id 不可能匹配任何已持久化 summary 事件。
触发条件： 任何跨对象/跨进程的重复压缩场景（一次 update_session 成功但调用方未观察到、或 session 从存储重建后再次 compact），幂等检查都不会拦截第二次压缩。
实际影响： “使重试幂等”的文档承诺无法兑现；配合状态不持久化缺陷，重试/重建场景会对同一个前缀执行两次压缩，产生嵌套 summary 并丢失中间细节。
修正方向： 让 compaction_id 从持久化且确定的源派生（例如 sha256(session_id + 边界签名 + 边界 occurrence)，或复用被替换前缀中首个事件的 id），使重试时生成与首次相同的 id，幂等分支才能真正命中。
| def _session_event_records(self, session: Session) -> list[dict[str, Any]]: | ||
| """Convert the authoritative Session Events into extraction records.""" | ||
| records: list[dict[str, Any]] = [] | ||
| seen: set[str] = set() | ||
| # Archived Events are no longer addressable in the active model | ||
| # request. Their information is already represented by the active | ||
| # summary Event included in the extraction context. | ||
| events = list(session.events or []) | ||
| for event in events: | ||
| if event.is_summary_event and event.is_summary_event(): | ||
| continue | ||
| if event.id in seen: | ||
| continue | ||
| seen.add(event.id) | ||
| timestamp = float(event.timestamp or 0.0) | ||
| records.append({ | ||
| "kind": | ||
| "event", | ||
| "event_id": | ||
| event.id, | ||
| "recorded_at": | ||
| datetime.fromtimestamp( | ||
| timestamp, | ||
| tz=timezone.utc, | ||
| ).isoformat(), | ||
| "event": | ||
| event.model_copy(deep=True).model_dump( | ||
| mode="json", |
There was a problem hiding this comment.
问题： session-memory 提取的事件来源从 base 的 append-only transcript（含全部历史事件）改为仅构造 session.events 活跃窗口（_session_event_records 遍历 session.events 并显式跳过 summary 事件），而 checkpoint 只存放在 session.state["_trpc_agent:summary"]。一旦 compaction 把 checkpoint 边界之前的事件移入 historical_events，_event_records_after_checkpoint 在活跃窗口找不到 checkpoint_event_id，只能退回按 recorded_at（实际是事件时间戳）近似恢复，找不到时直接返回空列表并跳过提取。
触发条件： BEFORE_MODEL 路径中 extract_if_needed(force=True) 提取失败（LLM 生成错误、预算超限等）后 compaction 仍继续执行，或 checkpoint 边界事件已被 compaction 移出活跃窗口的正常流程。
实际影响： checkpoint 之后、compaction 之前尚未提取的事件随 compaction 进入 historical_events 后永远无法被提取，会话记忆永久遗漏该段信息；recorded_at 时间戳近似恢复还可能重复提取已处理内容。
修正方向： 提取的增量边界不应依赖易失的活跃窗口：将 checkpoint（last_event_id 等）持久化到独立于 events 重排的源（如 session.state 中不与 compaction 冲突的键），并让 _session_event_records 同时纳入 historical_events 中 checkpoint 之后的事件，或恢复 transcript 机制。
| """ | ||
|
|
||
| def _decorator(session: Session) -> bool: | ||
| if session.conversation_count > conversation_count: | ||
| session.conversation_count = 0 | ||
| return True | ||
| return False | ||
| return session.conversation_count > conversation_count |
There was a problem hiding this comment.
问题： set_summarizer_conversation_threshold 从 base 的“超过阈值时把 session.conversation_count = 0 并返回 True”改为仅返回 session.conversation_count > conversation_count，触发时重置逻辑被删除；而 DefaultSessionSummarizerManager.create_session_summary 只在 summary_text 成功时重置 conversation_count = 0，runner 每轮无条件 conversation_count += 1。
触发条件： 会话轮数超过阈值后，summarize 失败（LLM 调用异常、返回空摘要或 create_session_summary_by_events 捕获异常返回 None）：计数不会归零，下一轮 should_summarize_session 依旧返回 True。
实际影响： 每次失败后的每一轮对话都会重复触发一次 summarize LLM 调用（base 版本失败后计数被 checker 清零，需重新累积到阈值才再次尝试），模型持续故障时产生无界的重复调用与日志噪音。
修正方向： 在 manager 的 summarize 失败路径（summary_text 为空/异常）也重置 conversation_count = 0，或恢复 checker 触发时的重置语义，保证失败后至少间隔一个阈值周期再重试。
| def _session_lock(self, session_id: str) -> asyncio.Lock: | ||
| """Return the unique compaction lock for a session.""" | ||
| key = self._runtime.session_key(session_id) | ||
| lock = self._session_locks.get(key) | ||
| if lock is None: | ||
| lock = asyncio.Lock() | ||
| self._session_locks[key] = lock | ||
| return lock |
There was a problem hiding this comment.
问题： _apply_scoped/ToolResultBudget/HistorySnip/MicroCompact 的 _session_lock 使用 asyncio.Lock，但新 manager 同时支持 BEFORE_MODEL（主事件循环内）与 AFTER_TURN（_PostTurnWorkerThread 独立线程+独立事件循环，见 runners.py L49-52）两条调用路径；base 版本 autocompact 只经 before_model_callback 单循环执行，不存在该跨循环共享。
触发条件： defer_post_turn_processing=True 时，上一个 turn 的 AFTER_TURN 压缩在 worker 循环异步执行期间，用户发起同一 session 的下一 turn，BEFORE_MODEL 压缩在主循环并发进入临界区（Python 3.10+ 的 asyncio.Lock 不绑定循环但状态非线程安全，3.6-3.9 则直接抛跨循环错误）。
实际影响： 同一 session 的两次压缩/写操作并发执行，_states/session.events 竞态，可能产生重复压缩或持久化事件窗口不一致。
修正方向： 与 SessionOperationCoordinator 一致改用跨循环安全的线程锁（如 CrossLoopLock）作为 session 级互斥，或仅在单一循环下执行压缩。
| 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 把 fingerprint 写入 session.state["advanced_memory_pending_request_context_fingerprint"]，但全仓库没有任何代码把这些 pending fingerprint 迁移到事件的 custom_metadata["advanced_memory_request_context_fingerprint"]；_latest_usage_baseline 读取的 recorded_fingerprint 恒为 None，静态 fingerprint 不一致过滤分支永远不会执行。
触发条件： 每次模型请求都会向 session.state 写入该键并随 state 持久化；_latest_usage_baseline 在匹配最近带 usage_metadata 的事件时跳过 fingerprint 校验。
实际影响： (a) usage 基线可能匹配到模型/系统指令/工具配置不同时期的旧 usage 事件，token 预算估计偏差，导致提前或延迟触发压缩；(b) .state 中残留无消费者的失效键，污染持久化状态。
修正方向： 在事件落库时把 pending fingerprint 写入事件 custom_metadata 并清理 state 中该键，或删除 record_request_context 及指纹过滤分支，仅依赖内容 fingerprint 匹配。
| storage_session: Optional[StorageSession] = await self._sql_storage.get_for_update( | ||
| sql_session, | ||
| session_key, | ||
| ) |
There was a problem hiding this comment.
问题： 本次变更给 append_event/update_session_state 新增了 get_for_update 行锁（本文件 553-556、629-662 行的变更），但同为会话写路径的 update_session 仍用普通 get 读取（不带锁），而 advanced compaction 通过 _persist_session_compaction → ctx.session_service.update_session 全量重写会话事件（delete 全部行后重新插入），与加了行锁的 append_event 并发执行时缺少同一把锁。
触发条件： 多实例/多事件循环下，一个请求的 compaction 执行 update_session（删除并重写 events 行）的同时，另一个请求对该 session 执行 append_event 插入新事件。
实际影响： 事务交错时并发 append 的新事件行可能被 compaction 的 delete-all 删除，或 compaction 写入的窗口覆盖并发 append 的事件，活动窗口静默丢失事件（base 版本 advanced compact 只写 transcript、不重写 events，此写路径为本变更新引入）。
修正方向： update_session 改用 get_for_update 并在事务内完成删除与重写，或为 compaction 持久化提供补丁式更新（仅替换被压缩前缀对应的事件行），避免全量重写与并发 append 竞争。
| from ._runtime import AdvancedMemoryRuntime | ||
|
|
||
|
|
||
| def install_staged_callback( |
| session_config: Optional[SessionServiceConfig] = None, | ||
| is_async: bool = False, | ||
| **kwargs: Any): | ||
| self._db_url = db_url |
| # AsyncSession cannot perform an implicit refresh when an ORM | ||
| # attribute is accessed after commit. Keep committed values available | ||
| # because this service reads StorageSession state after committing. | ||
| kwargs.setdefault("expire_on_commit", False) |
There was a problem hiding this comment.
kwargs.setdefault("expire_on_commit", False) 这个是否需要？
| from .utils import from_function_with_options | ||
| from .utils import get_required_fields | ||
| from .utils import parse_schema_from_parameter | ||
| from .utils import register_checker |
There was a problem hiding this comment.
建议导入“create_advanced_memory_toolset” 方便用户使用
| This module provides memory/RAG functionality including: | ||
| - Abstract memory service interfaces | ||
| - In-memory memory service implementation | ||
| - In-memory, Redis, and SQL memory implementations |
| from typing import Any | ||
| from uuid import uuid4 | ||
|
|
||
| from trpc_agent_sdk.storage import RedisCommand, RedisExpire, RedisStorage |
| import asyncio | ||
| from contextlib import asynccontextmanager | ||
| from dataclasses import replace | ||
| from datetime import datetime, timezone |
| from trpc_agent_sdk.types import Ttl | ||
|
|
||
| from ._config import AdvancedMemoryConfig | ||
| from ._formats import AdvancedMemoryDocument, AdvancedMemoryIndexEntry |
| from ._formats import AdvancedMemoryDocument, AdvancedMemoryIndexEntry | ||
| from ._formats import limit_memory_index | ||
| from ._storage_key import AdvancedMemoryPaths | ||
| from ._file_storage import parse_memory_index, prune_memory_index |
There was a problem hiding this comment.
按照字典序导入，这个放在_formats上面，
| from typing import Any | ||
|
|
||
| from ._config import AdvancedMemoryConfig | ||
| from trpc_agent_sdk.sessions.compact.advanced._coordination import CrossLoopLock |
AI Code Review审查结论通过 审查范围： 发现的问题较低
问题： base 版本向事件 触发条件： 启用 token-mode（ 实际影响： 指纹 guard 恒不生效，过期 usage 事件被当作有效基线，token 估算与真实上下文偏差，自动压缩/告警阈值误判； 修正方向： 在响应 Event 落盘路径上恢复消费 pending 指纹并写入事件元数据（或在 较低
问题： 本次变更删除了 触发条件： 在 CI 或本地运行 实际影响： 该测试确定性失败，测试套件变红；同时它已不能反映新行为，会误导后续维护者以为 summary 仍注入系统指令。 修正方向： 删除 |
|
|
||
| result = await self.apply(request, ctx=ctx, force=force) | ||
| if not result.blocked: | ||
| TokenContextTracker.record_request_context(request, ctx) |
There was a problem hiding this comment.
问题： base 版本向事件 custom_metadata 写入 advanced_memory_request_context_fingerprint 的唯一写入方是在本范围内被删除的 trpc_agent_sdk/advanced_memory/_session_service.py 的 append_event（从 state["advanced_memory_pending_request_context_fingerprint"] 取指纹写入事件元数据），head 全库已无任何代码写入该元数据键；因此本文件中 _latest_usage_baseline 的指纹 guard 和第 251 行 record_request_context 对 pending 指纹的写入均失去消费者，形成断裂的元数据链。
触发条件： 启用 token-mode（model_context_window_tokens 非空）且会话内模型/指令配置发生变化后，usage 事件仍会被 _latest_usage_baseline 接受为基线。
实际影响： 指纹 guard 恒不生效，过期 usage 事件被当作有效基线，token 估算与真实上下文偏差，自动压缩/告警阈值误判；record_request_context 的 pending 写入成为死代码。该功能默认关闭，影响有界。
修正方向： 在响应 Event 落盘路径上恢复消费 pending 指纹并写入事件元数据（或在 _auto_compact.py 第 581 行的 record_request_context 调用点直接写入），与删除写入方二选一；若该机制不再需要，则应同时删除本文件的指纹 guard 与 record_request_context 的 pending 写入，避免死代码。
| @@ -262,11 +262,6 @@ async def _add_instructions_to_request(self, agent: BaseAgent, ctx: InvocationCo | |||
| instructions_parts.append(code_executor_instruction) | |||
| logger.debug("Added code executor instruction for agent: %s", agent.name) | |||
|
|
|||
There was a problem hiding this comment.
问题： 本次变更删除了 _request_processor.py 中原有的 session summary 注入块（原先调用 ctx.session_service.get_session_summary 并把结果拼进系统指令），但未同步更新 tests/agents/core/test_request_processor_ext.py 中的 test_session_summary_appended，该测试仍断言 "previous summary text" in str(request.config.system_instruction)。
触发条件： 在 CI 或本地运行 tests/agents/core/test_request_processor_ext.py。
实际影响： 该测试确定性失败，测试套件变红；同时它已不能反映新行为，会误导后续维护者以为 summary 仍注入系统指令。
修正方向： 删除 test_session_summary_appended，或将其改为断言 summary 不再注入系统指令（例如断言结果中不含 summary 文本），以与新实现保持一致。
Advance Memory 服务化
基于 Session Service 的advanced compact实现
支持 local，sql，redis 的 advanced long-term memory