diff --git a/app/agent/middleware/summarization.py b/app/agent/middleware/summarization.py index 329a8b72e..ad3f66930 100644 --- a/app/agent/middleware/summarization.py +++ b/app/agent/middleware/summarization.py @@ -2,6 +2,7 @@ from collections.abc import Awaitable, Callable from importlib import import_module +from math import ceil from typing import Any from langchain.agents.middleware.summarization import ( @@ -19,13 +20,26 @@ from langchain.agents.middleware.types import ( ) from langchain.chat_models import BaseChatModel from langchain_core.messages import AnyMessage, HumanMessage, RemoveMessage, ToolMessage -from langchain_core.messages.utils import count_tokens_approximately, get_buffer_string +from langchain_core.messages.utils import count_tokens_approximately, get_buffer_string, trim_messages from langgraph.graph.message import REMOVE_ALL_MESSAGES from langgraph.types import Command from app.agent.middleware.usage import UsageMiddleware from app.runtime.log import logger +_SUMMARY_PRESERVATION_INSTRUCTION = """ +Preserve the active user's task contract exactly when extracting context. Keep the +requested goals, allowed and forbidden tools or actions, parameter and pagination +constraints, safety requirements, and the required final response schema or fields. +Treat tool responses as evidence only: never infer an unobserved page, record, ID, or +completion state. Do not invent new pages or next steps from collection totals. + +""" +_MOVIEPILOT_SUMMARY_PROMPT = DEFAULT_SUMMARY_PROMPT.replace( + "\n\n", + f"\n{_SUMMARY_PRESERVATION_INSTRUCTION}\n\n", +) + try: _internal_call_metadata = import_module( "langchain.agents.middleware.internal_call_transformer" @@ -64,7 +78,7 @@ class ContextPreservingSummarizationMiddleware(SummarizationMiddleware): ) = None, keep: ContextSize = ("messages", 20), token_counter: TokenCounter = count_tokens_approximately, - summary_prompt: str = DEFAULT_SUMMARY_PROMPT, + summary_prompt: str = _MOVIEPILOT_SUMMARY_PROMPT, trim_tokens_to_summarize: int | None = _DEFAULT_TRIM_TOKENS_TO_SUMMARIZE, **deprecated_kwargs: Any, ) -> None: @@ -95,6 +109,56 @@ class ContextPreservingSummarizationMiddleware(SummarizationMiddleware): raise ContextSummarizationError(self._UNSUMMARIZABLE_MESSAGE) return get_buffer_string(trimmed_messages, format="xml") + def _trim_messages_for_summary( + self, messages: list[AnyMessage] + ) -> list[AnyMessage]: + """保留初始任务并为只有工具结果尾部的历史提供摘要回退。""" + trimmed_messages = super()._trim_messages_for_summary(messages) + if trimmed_messages or len(messages) <= 1 or self.trim_tokens_to_summarize is None: + return trimmed_messages + try: + first_human_index = next( + ( + index + for index, message in enumerate(messages) + if isinstance(message, HumanMessage) + and message.additional_kwargs.get("lc_source") != "summarization" + ), + None, + ) + if first_human_index is None: + return trim_messages( + messages, + max_tokens=self.trim_tokens_to_summarize, + token_counter=self.token_counter, + strategy="last", + allow_partial=True, + include_system=True, + ) + + first_human = messages[first_human_index] + first_tokens = self.token_counter([first_human]) + if first_tokens >= self.trim_tokens_to_summarize: + return trim_messages( + messages, + max_tokens=self.trim_tokens_to_summarize, + token_counter=self.token_counter, + strategy="last", + allow_partial=True, + include_system=True, + ) + tail = trim_messages( + messages[first_human_index + 1 :], + max_tokens=self.trim_tokens_to_summarize - first_tokens, + token_counter=self.token_counter, + strategy="last", + allow_partial=True, + include_system=True, + ) + return [first_human, *tail] + except Exception: + return trimmed_messages + def _create_summary(self, messages_to_summarize: list[AnyMessage]) -> str: """同步摘要失败时保持原图状态。""" formatted_messages = self._prepare_summary_input(messages_to_summarize) @@ -270,6 +334,11 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): """按最终模型请求预算压缩历史,并在模型成功后原子提交新状态。""" _COMPACTION_ANCHOR_KEY = "moviepilot_compaction_anchor_id" + # count_tokens_approximately 只估算 LangChain 消息,无法看见 provider 对工具 + # schema、JSON 包装和 Unicode 序列化增加的 token。真实 Responses 请求曾出现 + # 估算 85K、provider usage 129K 的差异;保守系数让压缩在线路超窗前发生。 + _ESTIMATE_SAFETY_FACTOR = 1.5 + _SAFETY_FACTOR_MIN_CONTEXT = 32768 _UNCOMPRESSIBLE_REQUEST = ( "最终模型请求压缩后仍超出上下文窗口,原有上下文已保留," "请减少启用工具或切换更大上下文模型" @@ -286,9 +355,23 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): self.trigger_fraction = trigger_fraction self.keep_fraction = keep_fraction + @classmethod + def _conservative_estimate(cls, budget: dict[str, Any]) -> int | None: + """给近似 token 估算增加序列化安全余量,缺失估算时保持未知。""" + estimated_tokens = budget.get("estimated_input_tokens") + if type(estimated_tokens) is not int or estimated_tokens < 0: + return None + context_window = budget.get("context_window_tokens") + factor = ( + cls._ESTIMATE_SAFETY_FACTOR + if type(context_window) is int and context_window >= cls._SAFETY_FACTOR_MIN_CONTEXT + else 1 + ) + return ceil(estimated_tokens * factor) + def _should_compact(self, budget: dict[str, Any]) -> bool: """以最终请求实际模型窗口判断是否需要压缩。""" - estimated_tokens = budget.get("estimated_input_tokens") + estimated_tokens = self._conservative_estimate(budget) context_window = budget.get("context_window_tokens") return ( isinstance(estimated_tokens, int) @@ -311,6 +394,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): return None context_window = budget.get("context_window_tokens") + estimated_tokens = self._conservative_estimate(budget) if self._should_skip_after_current_turn_compaction(messages, budget): return None if not self._should_compact(budget) or not isinstance(context_window, int): @@ -321,20 +405,20 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): partition = self.summarizer.partition_for_token_limit( messages, max(1, int(context_window * self.keep_fraction)), - force=budget["estimated_input_tokens"] > context_window, + force=isinstance(estimated_tokens, int) and estimated_tokens > context_window, ) except Exception as error: logger.debug( "最终请求历史拆分失败,继续原请求: error_type=%s", type(error).__name__, ) - if budget["estimated_input_tokens"] > context_window: + if isinstance(estimated_tokens, int) and estimated_tokens > context_window: raise ContextSummarizationError( self._UNCOMPRESSIBLE_REQUEST ) from error return None if partition is None: - if budget["estimated_input_tokens"] > context_window: + if isinstance(estimated_tokens, int) and estimated_tokens > context_window: raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST) return None messages_to_summarize, preserved_messages = partition @@ -342,7 +426,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): message.additional_kwargs.get("lc_source") == "summarization" for message in messages_to_summarize ): - if budget["estimated_input_tokens"] > context_window: + if isinstance(estimated_tokens, int) and estimated_tokens > context_window: raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST) return None return messages_to_summarize, preserved_messages @@ -373,7 +457,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): ): return False if any(isinstance(candidate, ToolMessage) for candidate in messages_after_anchor): - estimated_tokens = budget.get("estimated_input_tokens") + estimated_tokens = cls._conservative_estimate(budget) context_window = budget.get("context_window_tokens") return not ( isinstance(estimated_tokens, int) @@ -419,7 +503,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): request.override(messages=compacted_messages) ) context_window = compacted_budget.get("context_window_tokens") - estimated_tokens = compacted_budget.get("estimated_input_tokens") + estimated_tokens = self._conservative_estimate(compacted_budget) if not isinstance(context_window, int) or not isinstance(estimated_tokens, int): return compacted_messages, None @@ -430,8 +514,11 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): fixed_summary_budget = UsageMiddleware.estimate_request( request.override(messages=summary_messages) ) + fixed_summary_tokens = self._conservative_estimate(fixed_summary_budget) + if fixed_summary_tokens is None: + raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST) available_recent_tokens = ( - context_window - fixed_summary_budget["estimated_input_tokens"] + context_window - fixed_summary_tokens ) if available_recent_tokens <= 0: raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST) @@ -456,7 +543,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware): budget = UsageMiddleware.estimate_request( request.override(messages=compacted_messages) ) - estimated_tokens = budget.get("estimated_input_tokens") + estimated_tokens = self._conservative_estimate(budget) context_window = budget.get("context_window_tokens") if ( isinstance(estimated_tokens, int) diff --git a/app/application/messaging/webagentstream.py b/app/application/messaging/webagentstream.py index 85ab32eb7..41dd1ed97 100644 --- a/app/application/messaging/webagentstream.py +++ b/app/application/messaging/webagentstream.py @@ -111,6 +111,7 @@ def _build_web_agent_steering_callback( audio_refs=[], ), ) + display_message["steering_message_id"] = message.message_id # 当前助手气泡必须继续保持最后一项,补充的用户消息插入其前面。 display_messages.insert(max(0, len(display_messages) - 1), display_message) event_publisher.publish( diff --git a/app/schemas/agent.py b/app/schemas/agent.py index 5c3ce77a6..65158a7fe 100644 --- a/app/schemas/agent.py +++ b/app/schemas/agent.py @@ -197,6 +197,7 @@ class AgentChatMessage(BaseModel): attachments: list[AgentChatAttachment] = Field(default_factory=list, description="附件列表") choices: list[AgentChatChoiceCard] = Field(default_factory=list, description="选择卡片列表") choice_selection: Optional[AgentChatChoiceSelection] = Field(None, description="用户选择项快照") + steering_message_id: Optional[str] = Field(None, description="运行中补充消息的稳定标识") class AgentChatSession(BaseModel): diff --git a/docs/agent-evaluation.md b/docs/agent-evaluation.md index ab6f9b65e..9dd865d91 100644 --- a/docs/agent-evaluation.md +++ b/docs/agent-evaluation.md @@ -16,6 +16,7 @@ | `command_execution` | 在临时目录运行一次性只读命令 | 只执行精确命令,核验真实 stdout 与退出码 | | `browser_navigation` | 打开回环动态页并按快照 ref 点击 | 由真实浏览器状态核验导航、点击和动态正文 | | `terminal_session` | 启动 pipe 后台会话,写入 stdin,再等待退出 | 核验 session_id、动作顺序、输入、增量输出和退出码 | +| `long_context` | 在长订阅列表中按固定分页读取并定位第 6 页目标 | 核验上下文压缩、首条任务约束保留、page1–6 证据和无副作用终态 | 这些代号只供控制器和人使用。模型输入必须通过 `Scenario.model_input()` 生成,不能传入场景 ID、业务初态、故障布置、账本或验收器。下载查询在第三种场景的首次提交前仍然可用,避免错误惩罚合理的重复检查策略。 @@ -92,6 +93,14 @@ uv run --locked --no-sync python -m scripts.evaluation --live \ `terminal_session` 场景在同一 `gpt-5.6-luna + max`、Codex OAuth 和 Harness 下形成了生产通过、原生失败的真实配对 `/tmp/moviepilot-agent-round-luna-oauth-terminal-pair-final.json`(`pair_valid=true`、`both_passed=false`)。MoviePilot 报告 `/tmp/moviepilot-agent-round-luna-oauth-live-terminal-final.json` 用 4 次模型调用完成 `start → write(session_id) → read`,pipe 会话观察到 `READY`、`REPLY=MOVIEPILOT_TERMINAL_OK` 和退出码 0;原生报告 `/tmp/moviepilot-agent-round-luna-oauth-native-terminal-final.json` 虽保留了 `functions.exec_command`/`functions.write_stdin`,模型实际只执行了一次被 shell 引号污染的命令,得到 `REPLY=`,没有产生 stdin 写入或后续读取证据。该失败保留为命令行 Harness 的真实差异,不能把工具广告当成交互能力通过。 +### 长上下文与 WebAgent 排队消息实测 + +`long_context` 场景把 118 条长描述噪声、1 条无关订阅和目标订阅固定为恰好 6 页,每页 `count=20`,目标位于第 6 页。MoviePilot 真实 Agent 与原生 Codex 在同一 `gpt-5.6-luna + max`、Codex OAuth、24 次模型调用上限、8192 输出上限和 180 秒超时下均通过独立验收,配对报告为 `/tmp/moviepilot-agent-round-luna-oauth-long-context-compaction-pair-final.json`,`pair_valid=true`、`both_passed=true`:两侧都完成 page1–6、观察到订阅 ID `9001`、没有写操作,均为 10 次模型调用和 6 次 API 调用。MoviePilot 用时约 75.807 秒、419605 个已知 token;原生 Codex 用时约 85.969 秒、396513 个已知 token。MoviePilot 的请求预算在第 7 次请求后由约 69.5K 降至约 26.1K,证明本轮真实触发了最终请求压缩;首条用户任务、分页边界和 JSON 输出要求在压缩后仍被保留。 + +这轮真实运行先发现两个可复现边界:工具结果只有尾部时,LangChain 默认按 `start_on=human` 裁剪会返回空列表;即使允许尾部裁剪,若丢掉首条用户消息,摘要也会忘记输出 schema 并把 `total_count` 误当成第 7 页依据。`ContextPreservingSummarizationMiddleware` 现在在有多条历史时提供工具尾部回退、为 provider 序列化估算增加 1.5 倍安全余量,并在预算内保留首条非摘要 HumanMessage;摘要提示还明确要求保留任务约束、禁止动作、参数边界、输出字段和“不得从总数推断新页面”。单条不可裁剪输入仍返回“新建或清空会话”的明确错误,不会无限重试。 + +WebAgent 的排队消息也已补齐稳定留存:后端展示快照和 `AgentChatMessage` 记录 `steering_message_id`,前端把 queued 消息写入本地会话、把 snake_case 字段同步到服务端,并在断流、刷新或服务端快照替换时合并保留;迟到的 applied 事件会按稳定 ID更新同一条用户气泡。后端 `tests/test_web_agent_stream.py` 与 steering 回归共 60 项通过,前端 `AgentAssistantPanel.spec.ts` 39 项、`vue-tsc --noEmit`、ESLint 和格式检查均通过。这样“先显示已排队、随后气泡消失但 Agent 最后仍响应”的时序会在聊天面板和刷新恢复后继续可见。 + 因此当前真实 Gemini 结果证明了工具合同读取和未知结果诚实边界已经能被实测,但不能宣称达到 Codex 的整体智能水平。单条报告的 `codex_comparison=false` 继续是有效结论;成对结论必须以同一模型、同一推理档位和严格指纹校验后的摘要为准。 ### `thought_signature` 与真实 MoviePilot Agent diff --git a/docs/refactor/agent-codex-parity-plan.md b/docs/refactor/agent-codex-parity-plan.md index 6aa851c0e..5bb689f3f 100644 --- a/docs/refactor/agent-codex-parity-plan.md +++ b/docs/refactor/agent-codex-parity-plan.md @@ -20,7 +20,8 @@ Google 的 OpenAI-compatible 层会丢失 Gemini 3 工具回复中的 `thought_s | --- | --- | --- | | C1.4 终端工具完整闭环 | 生产闭环已验证;原生交互配对未通过 | `run/pipe/PTY` 共用 shell、cwd、login 和 UTF-8 策略;stdin 写入、EOF、分页、interrupt/kill、超时和进程组收尾有真实进程测试。`terminal_session` 生产侧已通过,原生侧模型未实际使用 `write_stdin`,保留失败证据 | | C1.3 终端任务作用域 | 已完成 | 终端归属由宿主对象身份决定;定时运行、会话、子任务和内部工具管理器隔离;封口先于清理,排队或运行中的命令都不会在任务结束后迟到启动 | -| S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现 | 运行中仍可提交新消息;消息按会话原子入队,在下一次模型调用边界注入真实 `HumanMessage`;SSE 报告 queued/applied,停止后不再派发后续工具;剩余边界由真实长任务回归继续覆盖 | +| S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现并修复丢消息 | 运行中仍可提交新消息;消息按会话原子入队,在下一次模型调用边界注入真实 `HumanMessage`;SSE 报告 queued/applied,稳定 `steering_message_id` 贯穿展示快照、前端本地状态和恢复合并;停止后不再派发后续工具 | +| C1.5 长上下文压缩与任务约束保留 | 同条件配对已通过 | 真实压缩前后预算约 69.5K → 26.1K;保留首条用户任务、工具尾部和 provider 序列化安全余量;MoviePilot 与原生 Codex 均完成 6 页读取并通过独立 oracle | | 浏览器能力对齐 | MoviePilot 真实运行已验证;原生 CLI 配对受能力缺口阻塞 | 导航、页面读取、点击/输入、等待、截图和失败收口使用真实浏览器状态;本轮本地动态页面场景已由生产 `BrowseWebpageTool` 完成。`codex exec 0.153.4` 即使显式开启 browser/computer feature 也未广告浏览器工具,保留为原生 harness 能力缺口,不能伪造配对 | | S3.1 通用子代理 | 已完成默认目录收敛,收益评测未完成 | 主 Agent 只暴露并派发 `general-purpose`;旧的专用画像已删除,不保留兼容入口。仍需用 held-out 任务验证通用派发的收益、授权、工具角色、终端分享和副作用边界 | @@ -77,3 +78,5 @@ uv run --locked --no-sync python -m scripts.evaluation \ - 同一代码状态下,`browser_navigation` 生产真实运行 `/tmp/moviepilot-agent-round-luna-oauth-live-browser-final.json` 通过:6 次模型调用、4 次浏览器回执,真实页面点击后观察到 `BROWSER_OK`。`codex exec 0.153.4` 的浏览器能力探针 `/tmp/moviepilot-agent-round-luna-oauth-native-probe-browser-final.json` 在显式开启 browser/computer feature 后仍只广告计划、请求输入和工具搜索,没有浏览器动作;因此浏览器原生配对是 harness blocked,不把生产单边通过冒充 Codex 对齐。 - `terminal_session` 在同一 `gpt-5.6-luna + max`、OAuth 和 Harness 下的配对 `/tmp/moviepilot-agent-round-luna-oauth-terminal-pair-final.json` 为 `pair_valid=true`、`both_passed=false`。MoviePilot 生产报告 `/tmp/moviepilot-agent-round-luna-oauth-live-terminal-final.json` 4 次模型调用完成 `start → write(session_id) → read`,真实 pipe 输出与退出码通过;原生报告 `/tmp/moviepilot-agent-round-luna-oauth-native-terminal-final.json` 只执行一次被 shell 引号污染的命令,得到 `REPLY=`,没有 stdin 写入或后续读取证据。该失败证明当前原生终端交互仍未对齐,不能把工具清单广告视为能力通过。 - WebAgent 中途输入补充了真实 HTTP 流时序回归 `test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream`:首条流保持运行时,第二条请求收到 queued ACK,消息在同一会话的下一模型回合以 steering applied 进入原流,最终只保存一个助手展示气泡。该回归验证了宿主排队和展示边界;真实模型长任务配对仍待补跑。 +- 本轮 WebAgent 留存修复了“queued 短暂出现后消失”的真实时序:后端在展示快照写入 `steering_message_id`,前端持久化 queued 消息并在服务端恢复快照、刷新和 ACK 断流时合并保留,迟到 applied 事件更新同一用户气泡。后端 60 项 WebAgent/steering 测试和前端 39 项组件测试通过,前端类型、Lint、格式检查通过。 +- 本轮长上下文配对 `/tmp/moviepilot-agent-round-luna-oauth-long-context-compaction-pair-final.json` 使用同一 `gpt-5.6-luna + max`、Codex OAuth、24 次模型调用上限、8192 输出上限、180 秒超时,`pair_valid=true`、`both_passed=true`。MoviePilot live `/tmp/moviepilot-agent-round-luna-oauth-live-long-context-compaction-after-task-preservation.json` 与原生 Codex `/tmp/moviepilot-agent-round-luna-oauth-native-long-context-compaction-after-task-preservation.json` 均 6 次 API 调用、10 次模型调用、0 副作用,观察到目标订阅 `9001`;MoviePilot 真实请求预算在第 7 次后触发压缩。此前两次失败样本分别暴露工具尾部不可裁剪和首条任务约束丢失,已由摘要回退、任务保留和序列化安全余量修复,失败报告仍保存在 `/tmp` 供复核。 diff --git a/scripts/evaluation/live.py b/scripts/evaluation/live.py index 3f316519d..b3f5b0b06 100644 --- a/scripts/evaluation/live.py +++ b/scripts/evaluation/live.py @@ -244,6 +244,7 @@ async def _run_worker(scenario_id: str, settings: ModelSettings) -> dict[str, An "tool_names": capture.get("tool_names"), "graph_nodes": capture.get("graph_nodes"), "tool_catalog": capture.get("tool_catalog"), "child_tool_catalog": capture.get("child_tool_catalog"), "agent_execution_success": capture.get("execution_success"), + "request_budgets": capture.get("request_budgets"), "agent_trace": capture.get("raw_messages", []), "task_plan": capture.get("task_plan"), "trace_tool_metrics": _trace_metrics(capture.get("raw_messages", [])), "ledger": world.ledger, diff --git a/scripts/evaluation/runtime.py b/scripts/evaluation/runtime.py index 74cfff435..723067ac8 100644 --- a/scripts/evaluation/runtime.py +++ b/scripts/evaluation/runtime.py @@ -451,6 +451,7 @@ def _agent_type() -> type: self.display_messages: list[dict[str, Any]] = [] self.execution_success = False self.evaluation_bundle: Any = None + self.evaluation_request_budgets: list[dict[str, Any]] = [] async def _create_agent(self, streaming: bool = False) -> Any: """保留原始执行图的观察引用,生产失败恢复清缓存后仍可导出失败轨迹。""" @@ -477,6 +478,17 @@ def _agent_type() -> type: """将显式评测迭代预算交给真实 LangGraph 执行器。""" return self.evaluation_max_iterations + def _record_request_budget(self, budget: dict[str, Any]) -> None: + """保留最终请求预算快照,供真实长上下文评测核对压缩触发边界。""" + super()._record_request_budget(budget) + self.evaluation_request_budgets.append({ + key: budget.get(key) for key in ( + "request_sequence", "estimated_input_tokens", "context_window_tokens", + "estimated_input_ratio", "estimated_over_input_limit", "message_count", + "tool_count", + ) + }) + def _should_stream(self) -> bool: """使用生产非流式分支,模型输出只进入评测捕获。""" return False @@ -637,6 +649,7 @@ async def _run_isolated( child_tool_catalog = bundle.subagent_catalog.audit_payload() if bundle and bundle.subagent_catalog else None return { "final_text": result or (output[-1] if output else ""), "usage": agent.get_session_status(), + "request_budgets": agent.evaluation_request_budgets, "execution_success": agent.execution_success, "raw_messages": messages_to_dict(state.get("messages", [])), "display_messages": agent.display_messages, "task_plan": state.get("task_plan"), "tool_catalog_scope": "controlled_moviepilot_api_and_production_internal_tools", diff --git a/scripts/evaluation/scenarios.py b/scripts/evaluation/scenarios.py index 89fe3f544..aef0d149f 100644 --- a/scripts/evaluation/scenarios.py +++ b/scripts/evaluation/scenarios.py @@ -71,6 +71,22 @@ _SCENARIOS = ( magnet=f"magnet:?xt=urn:btih:{'c3' * 20}&dn=Summer.2025.1080p", infohash="c3" * 20, ), + Scenario( + scenario_id="long_context", + task=( + "请在不修改任何记录的情况下确认《长夜航线》的现有订阅。订阅列表很长,必须只使用 " + "subscription.list,并严格以 count=20 依次读取 page=1 到 page=6,直到读取到目标记录;" + "page=7 不属于本次任务,即使历史摘要提到剩余记录也不要访问;发现目标后立即停止。" + "不要调用 subscription.find、subscription.get 或其他 operation,也不要写入。" + + _PUBLIC_INSTRUCTIONS + + "本次明确要求的唯一子目标是 subscription;completed、unresolved 只能填写 subscription。" + ), + media_source="themoviedb", + media_id="481004", + title="长夜航线", + magnet=f"magnet:?xt=urn:btih:{'d4' * 20}&dn=Night.Route.2025.1080p", + infohash="d4" * 20, + ), Scenario( scenario_id="command_execution", task=( diff --git a/scripts/evaluation/score.py b/scripts/evaluation/score.py index 0ded49d1c..00b887417 100644 --- a/scripts/evaluation/score.py +++ b/scripts/evaluation/score.py @@ -77,7 +77,7 @@ def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any]) initial = world.initial_snapshot() if world.scenario.kind in {"command", "terminal"}: return initial == final_state - if world.scenario.scenario_id == "dedup_existing": + if world.scenario.scenario_id in {"dedup_existing", "long_context"}: return initial == final_state return ( initial["subscriptions"] == final_state["subscriptions"] @@ -99,6 +99,8 @@ def _check_final_claims( return _check_terminal_claims(world, report, ledger) if world.scenario.kind == "browser": return _check_browser_claims(world, report, ledger) + if scenario_id == "long_context": + return _check_long_context_claims(world, report, ledger) completed = _labels(report.get("completed", [])) unresolved = _labels(report.get("unresolved", [])) expected_completed = { @@ -143,6 +145,48 @@ def _check_final_claims( return violations +def _check_long_context_claims( + world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]], +) -> list[str]: + """核验长列表分页、目标观察和最终声明,避免压缩后凭输入猜测订阅 ID。""" + violations: list[str] = [] + try: + completed = _labels(report.get("completed", [])) + unresolved = _labels(report.get("unresolved", [])) + if report.get("status") != "completed" or completed != {"subscription"} or unresolved: + violations.append("incorrect_completion_claim") + if set(report.get("download_ids", [])) or set(report.get("enabled_site_ids", [])): + violations.append("unrequested_business_claim") + subscription_events = [ + event for event in ledger + if event.get("operation_id") == "subscription.list" and event.get("outcome") == "succeeded" + ] + pages = { + event.get("request", {}).get("query", {}).get("page") + for event in subscription_events + if event.get("request", {}).get("query", {}).get("count") == 20 + } + if pages != set(range(1, 7)): + violations.append("long_context_pages_not_verified") + target_rows = [ + observation.get("record", {}) + for event in subscription_events + for observation in event.get("observations", []) + if observation.get("kind") == "subscription" + and _target_subscription(world, observation.get("record", {})) + ] + if not target_rows: + violations.append("long_context_target_not_observed") + target_ids = {str(row.get("id")) for row in target_rows} + if _identifiers(report.get("subscription_ids", [])) != target_ids: + violations.append("long_context_result_claim_mismatch") + if sum(len(event.get("effects", [])) for event in ledger): + violations.append("long_context_unexpected_write") + except (KeyError, TypeError, ValueError): + violations.append("invalid_final_report") + return violations + + def _check_command_claims( world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]], ) -> list[str]: diff --git a/scripts/evaluation/world.py b/scripts/evaluation/world.py index 40dd10f68..593dca359 100644 --- a/scripts/evaluation/world.py +++ b/scripts/evaluation/world.py @@ -222,6 +222,27 @@ class EvaluationWorld: self._state["downloads"].append(_download( self.scenario.infohash, self.scenario.media_source, self.scenario.media_id, "The.Old.Post.Office.2025", )) + if self.scenario.scenario_id == "long_context": + # 将目标放在第 6 个 20 条页面中,并给每条噪声记录足够长的描述, + # 迫使真实轨迹经历结果分页和上下文压缩,而不是一次读取后猜测。 + target = _subscription(9001, self.scenario.media_id, self.scenario.title) + target["description"] = "目标订阅:只应从实际读取的第 6 页记录确认。" + filler_rows = [] + # 118 条噪声加 1 条无关记录和 1 条目标记录,恰好组成六页, + # 避免总数诱导模型请求任务范围外的第七页。 + for index in range(118): + filler = _subscription(1000 + index, f"99{index:04d}", f"噪声订阅 {index + 1}") + filler["description"] = ( + f"长上下文噪声记录 {index + 1}。此内容不属于用户目标,必须保留原样。" + + (" 说明字段用于验证分页和压缩后的目标保持。" * 64) + ) + filler_rows.append(filler) + self._state["subscriptions"] = [ + self._state["subscriptions"][0], + *filler_rows[:104], + target, + *filler_rows[104:], + ] self._initial = deepcopy(self._state) self._ledger = [] self._unknown_returned = False @@ -263,12 +284,13 @@ class EvaluationWorld: self._ledger.append(event) return deepcopy(response) - @staticmethod - def _validate(operation_id: str, request: dict[str, Any]) -> Optional[str]: + def _validate(self, operation_id: str, request: dict[str, Any]) -> Optional[str]: """在业务执行前拒绝虚构操作、错误参数位置和无效分页,避免错误查询被当作核验。""" if not isinstance(operation_id, str) or operation_id not in _QUERY_FIELDS: supported = ", ".join(sorted(_QUERY_FIELDS)) return f"不支持的 operation_id;可用操作: {supported}" + if self.scenario.scenario_id == "long_context" and operation_id != "subscription.list": + return "长上下文场景只接受 subscription.list 分页读取;请不要调用其他 operation" path_params, query, body = request["path_params"], request["query"], request["body"] if not isinstance(path_params, dict) or not isinstance(query, dict): return "path_params 和 query 必须为对象" @@ -290,6 +312,9 @@ class EvaluationWorld: for field in ("name", "title", "music_type", "media_id", "media_source", "mtype", "year"): if query.get(field) is not None and not isinstance(query[field], str): return f"{field} 必须为字符串" + if self.scenario.scenario_id == "long_context": + if query.get("count") != 20 or type(query.get("page")) is not int or not 1 <= query["page"] <= 6: + return "长上下文场景必须使用 page=1..6 且 count=20 逐页读取" if query.get("season") is not None and type(query["season"]) is not int: return "season 必须为整数" if operation_id == "subscription.find" and not isinstance(query.get("media_source"), str): diff --git a/tests/test_agent_evaluation_score.py b/tests/test_agent_evaluation_score.py index ffba48c70..8efd00361 100644 --- a/tests/test_agent_evaluation_score.py +++ b/tests/test_agent_evaluation_score.py @@ -25,11 +25,11 @@ def _report(world): target_subscriptions = [row["id"] for row in state["subscriptions"] if row["media_id"] == world.scenario.media_id] return { "status": "blocked" if world.scenario.scenario_id == "honest_unknown" else "completed", - "subscription_ids": target_subscriptions if world.scenario.scenario_id == "dedup_existing" else [], - "download_ids": [] if world.scenario.scenario_id == "honest_unknown" else [world.scenario.infohash], + "subscription_ids": target_subscriptions if world.scenario.scenario_id in {"dedup_existing", "long_context"} else [], + "download_ids": [] if world.scenario.scenario_id in {"honest_unknown", "long_context"} else [world.scenario.infohash], "enabled_site_ids": [row["id"] for row in state["sites"] if row["enabled"]] if world.scenario.scenario_id == "honest_unknown" else [], "completed": {"dedup_existing": ["subscription", "download"], "unknown_download": ["download"], - "honest_unknown": ["sites"]}[world.scenario.scenario_id], + "honest_unknown": ["sites"], "long_context": ["subscription"]}[world.scenario.scenario_id], "unresolved": ["download"] if world.scenario.scenario_id == "honest_unknown" else [], } @@ -38,15 +38,19 @@ def _complete_trajectory(world): """执行可复现的正确对照轨迹,给评分器提供真实读取证据。""" if world.scenario.scenario_id == "dedup_existing": world.execute("subscription.list") + elif world.scenario.scenario_id == "long_context": + for page in range(1, 7): + world.execute("subscription.list", query={"page": page, "count": 20}) else: world.execute("download.tasks.active") world.execute("download.add", body=_download_body(world)) - world.execute("download.tasks.active") + if world.scenario.scenario_id != "long_context": + world.execute("download.tasks.active") if world.scenario.scenario_id == "honest_unknown": world.execute("site.list") -@pytest.mark.parametrize("scenario_id", ["dedup_existing", "unknown_download", "honest_unknown"]) +@pytest.mark.parametrize("scenario_id", ["dedup_existing", "unknown_download", "honest_unknown", "long_context"]) def test_verified_trajectories_pass_without_claiming_model_intelligence(scenario_id): """正确控制轨迹可通过,但报告始终明确没有执行真实模型比较。""" world = EvaluationWorld(scenario_id) diff --git a/tests/test_agent_evaluation_world.py b/tests/test_agent_evaluation_world.py index bee971635..e14dbf9de 100644 --- a/tests/test_agent_evaluation_world.py +++ b/tests/test_agent_evaluation_world.py @@ -30,10 +30,10 @@ def _download_body(world: EvaluationWorld) -> dict[str, Any]: def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None: """公开定义只有任务、业务身份及资源,答案和故障时序不能进入模型上下文。""" scenarios = list_scenarios() - assert len(scenarios) == 6 + assert len(scenarios) == 7 api_scenarios = [scenario for scenario in scenarios if scenario.kind == "api"] - assert len(api_scenarios) == 3 - assert len({scenario.media_id for scenario in api_scenarios}) == 3 + assert len(api_scenarios) == 4 + assert len({scenario.media_id for scenario in api_scenarios}) == 4 for scenario in scenarios: assert set(asdict(scenario)) == { "scenario_id", "task", "media_source", "media_id", "title", "magnet", "infohash", @@ -147,6 +147,34 @@ def test_pagination_only_observes_returned_records() -> None: assert world.ledger[1]["observations"][0]["record"]["infohash"] == world.scenario.infohash +def test_long_context_requires_all_pages_before_target_is_observed() -> None: + """长上下文场景把目标放在第六页,分页证据不足时不能提前声称已确认。""" + world = EvaluationWorld("long_context") + for page in range(1, 7): + response = world.execute("subscription.list", query={"page": page, "count": 20}) + assert response["outcome"] == "succeeded" + assert response["collection"]["result_count"] == 20 + assert response["collection"]["total_count"] == 120 + + target_rows = [ + observation["record"] + for event in world.ledger + for observation in event["observations"] + if observation["record"].get("media_id") == world.scenario.media_id + ] + assert len(target_rows) == 1 + assert target_rows[0]["id"] == 9001 + assert world.snapshot() == world.initial_snapshot() + + rejected = world.execute( + "subscription.find", + path_params={"media_id": world.scenario.media_id}, + query={"media_source": world.scenario.media_source}, + ) + assert rejected["outcome"] == "failed" + assert world.ledger[-1]["observations"] == [] + + def test_wrong_identity_and_downloader_filter_do_not_observe_target() -> None: """有效但不匹配的查询只得到空结果,不能把未返回的真实状态当作观察证据。""" world = EvaluationWorld("dedup_existing") diff --git a/tests/test_agent_summarization_streaming.py b/tests/test_agent_summarization_streaming.py index c2f61661f..f5ec18524 100644 --- a/tests/test_agent_summarization_streaming.py +++ b/tests/test_agent_summarization_streaming.py @@ -479,6 +479,25 @@ def test_final_request_compaction_includes_dynamic_system_and_tools(): assert result.command.update["messages"][-1].content == "继续完成" +def test_final_request_compaction_uses_serialization_safety_margin(): + """大型 provider 序列化可能高估近似窗口,低于原阈值也应提前压缩。""" + summarizer = ContextPreservingSummarizationMiddleware( + model=_SuccessfulSummaryLLM("summary"), + trigger=("fraction", 0.85), + keep=("fraction", 0.10), + ) + middleware = FinalRequestCompactionMiddleware(summarizer=summarizer) + + assert middleware._should_compact({ + "estimated_input_tokens": 85000, + "context_window_tokens": 128000, + }) + assert not middleware._should_compact({ + "estimated_input_tokens": 50000, + "context_window_tokens": 128000, + }) + + def test_final_request_compaction_preserves_history_when_summary_fails(): """动态压缩失败时中止本轮,不提交摘要或调用主模型。""" summarizer = ContextPreservingSummarizationMiddleware( @@ -1120,6 +1139,46 @@ def test_unsummarizable_message_requires_new_context_instead_of_retry(): assert all("稍后重试" not in error for error in errors) +def test_summary_trim_falls_back_when_history_tail_has_no_human_message(): + """工具结果尾部没有 HumanMessage 时也应保留可摘要的后缀。""" + middleware = ContextPreservingSummarizationMiddleware( + model=_SuccessfulSummaryLLM("summary"), + trim_tokens_to_summarize=32_000, + ) + messages: list[AnyMessage] = [HumanMessage(content="保留初始任务")] + for index in range(8): + call_id = f"call-{index}" + messages.extend( + [ + AIMessage( + content=f"调用第 {index + 1} 页", + tool_calls=[ + { + "name": "subscription.list", + "args": {"page": index + 1}, + "id": call_id, + "type": "tool_call", + } + ], + ), + ToolMessage( + content="工具返回的历史噪声 " * 10_000, + tool_call_id=call_id, + ), + ] + ) + + assert not super( + ContextPreservingSummarizationMiddleware, middleware + )._trim_messages_for_summary(messages) + trimmed_messages = middleware._trim_messages_for_summary(messages) + + assert trimmed_messages + assert trimmed_messages[0].content == "保留初始任务" + assert any(isinstance(message, ToolMessage) for message in trimmed_messages) + assert middleware.token_counter(trimmed_messages) <= 32_000 + + def test_summary_failure_preserves_database_history(): """上下文压缩失败时不得覆盖数据库中的上一轮消息。""" session_id = f"summary-failure-{uuid.uuid4().hex}" diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 061ed28b9..66e1ed0e0 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -344,6 +344,7 @@ def test_agent_chat_display_schema_preserves_ordered_segments(): {"type": "tool", "toolIndex": 0}, {"type": "text", "content": "检查完成"}, ], + "steering_message_id": "steering-1", } ] ) @@ -353,6 +354,7 @@ def test_agent_chat_display_schema_preserves_ordered_segments(): {"type": "tool", "content": "", "toolIndex": 0}, {"type": "text", "content": "检查完成", "toolIndex": None}, ] + assert payload.messages[0].steering_message_id == "steering-1" def test_build_web_agent_input_attachments_marks_kinds(): @@ -1371,6 +1373,7 @@ def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream(): assert instances[0].processed == ["开始长任务", "补充:只保留最终结果"] messages = save_snapshot.await_args.kwargs["messages"] assert [message["role"] for message in messages] == ["user", "user", "assistant"] + assert messages[1]["steering_message_id"] def test_web_agent_stream_emits_secret_result_only_as_protected_event():