diff --git a/research-wiki/designs/core-agent-adapters-llm.md b/research-wiki/designs/core-agent-adapters-llm.md deleted file mode 100644 index 184cbac..0000000 --- a/research-wiki/designs/core-agent-adapters-llm.md +++ /dev/null @@ -1,9 +0,0 @@ ---- -type: design -node_id: design:core-agent-adapters-llm -title: "core/agent/ + adapters/llm 基础设施设计" -date: 2026-07-07 ---- - -# core/agent/ + adapters/llm 基础设施设计 - diff --git a/research-wiki/graph/edges.json b/research-wiki/graph/edges.json index 9c852c1..b01d9e6 100644 --- a/research-wiki/graph/edges.json +++ b/research-wiki/graph/edges.json @@ -10,7 +10,20 @@ "id": "design:core-agent-adapters-llm", "label": "core/agent/ + adapters/llm 基础设施设计", "type": "design" + }, + { + "id": "plan:core-agent-adapters-llm", + "label": "core/agent/ + adapters/llm 基础设施实现计划", + "type": "plan" } ], - "links": [] + "links": [ + { + "source": "plan:core-agent-adapters-llm", + "target": "design:core-agent-adapters-llm", + "relation": "implements", + "evidence": "计划实现设计文档中定义的全异步 AgentLoop + 四层治理栈", + "added": "2026-07-07T02:25:32.349931+00:00" + } + ] } \ No newline at end of file diff --git a/research-wiki/index.md b/research-wiki/index.md index 867eaaf..4cbf94a 100644 --- a/research-wiki/index.md +++ b/research-wiki/index.md @@ -1,10 +1,11 @@ # Research Wiki 索引 -> 自动生成，更新时间：2026-07-07 02:06 UTC +> 自动生成，更新时间：2026-07-07 02:25 UTC -## design (2) +## design (1) - [2026-07-06-core-agent-adapters-llm-design](designs/2026-07-06-core-agent-adapters-llm-design.md) `design:2026-07-06-core-agent-adapters-llm-design` -- [core/agent/ + adapters/llm 基础设施设计](designs/core-agent-adapters-llm.md) `design:core-agent-adapters-llm` -## plan (1) +## plan (3) +- [2026-07-06-core-agent-adapters-llm](plans/2026-07-06-core-agent-adapters-llm.md) `plan:2026-07-06-core-agent-adapters-llm` +- [core/agent/ + adapters/llm 基础设施实现计划](plans/core-agent-adapters-llm.md) `plan:core-agent-adapters-llm` - [项目基础设施初始化计划](plans/infrastructure-setup.md) `plan:infrastructure-setup` diff --git a/research-wiki/log.md b/research-wiki/log.md index 7688d17..2f6567f 100644 --- a/research-wiki/log.md +++ b/research-wiki/log.md @@ -5,3 +5,6 @@ - [2026-07-06 15:13 UTC] 重建索引: 1 篇页面 - [2026-07-07 02:06 UTC] 新增 design: core/agent/ + adapters/llm 基础设施设计 (design:core-agent-adapters-llm) - [2026-07-07 02:06 UTC] 重建索引: 3 篇页面 +- [2026-07-07 02:25 UTC] 新增 plan: core/agent/ + adapters/llm 基础设施实现计划 (plan:core-agent-adapters-llm) +- [2026-07-07 02:25 UTC] 新增边: plan:core-agent-adapters-llm --implements--> design:core-agent-adapters-llm +- [2026-07-07 02:25 UTC] 重建索引: 4 篇页面 diff --git a/research-wiki/plans/2026-07-06-core-agent-adapters-llm.md b/research-wiki/plans/2026-07-06-core-agent-adapters-llm.md new file mode 100644 index 0000000..624a8ea --- /dev/null +++ b/research-wiki/plans/2026-07-06-core-agent-adapters-llm.md @@ -0,0 +1,3155 @@ +# core/agent/ + adapters/llm 基础设施实现计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 实现全异步 AgentLoop 可提取内核 + GovernedLLMClient 四层治理栈（熔断→缓存→流式看门狗→重试），含流式三层看门狗和遥测集成。 + +**前置条件（执行前完成）：** +```bash +conda activate Video-Tree-TRM & pip install pluggy json_repair httpx "redis[hiredis]" fakeredis pytest-asyncio +``` +并将新依赖添加到 `pyproject.toml` 的 dev dependencies 中。 + +**Architecture:** core/protocols.py 定义共享端口（LLMProvider, VLMProvider, TelemetryRecorder），core/agent/ 实现算法保真 #11 的 AgentLoop 引擎（pluggy hooks + json_repair），adapters/ 组合式实现治理栈（streaming 看门狗 + CircuitBreaker + RedisResponseCache + SQLiteTelemetryRecorder → GovernedLLMClient）。 + +**Tech Stack:** Python 3.11, asyncio, pluggy, json_repair, httpx, redis.asyncio, loguru, pytest + pytest-asyncio + +**设计文档:** `research-wiki/designs/2026-07-06-core-agent-adapters-llm-design.md` + +--- + +### Task 1: 规范同步 — 更新 ARCHITECTURE.md + CLAUDE.md + .env.example + +**Files:** +- Modify: `research-wiki/ARCHITECTURE.md` +- Modify: `CLAUDE.md` +- Modify: `.env.example` + +本任务无测试，纯文档变更。按设计文档 §8 的清单逐项更新。 + +- [ ] **Step 1: 更新 ARCHITECTURE.md §2.3 目录结构** + +在 `core/` 下新增 `protocols.py`，`adapters/` 下新增 `streaming.py`、`breaker.py`： + +```text +├── core/ +│ ├── protocols.py # 共享端口：LLMProvider, VLMProvider, TelemetryRecorder +│ ├── agent/ +│ │ ├── loop.py +│ │ ├── types.py +│ │ └── protocols.py # Agent 专属：ToolDispatcher, AgentLoopSpec +│ ├── evolution/ +│ │ └── protocols.py # Evolution 专属：SkillStore, PromptStore, RunLog +│ └── types.py +``` + +```text +├── adapters/ +│ ├── llm.py # GovernedLLMClient +│ ├── streaming.py # 三层看门狗 +│ ├── breaker.py # CircuitBreaker +│ ├── vlm.py +│ ├── embedding.py +│ ├── redis_cache.py +│ ├── ocr.py +│ ├── asr.py +│ └── telemetry.py +``` + +- [ ] **Step 2: 更新 ARCHITECTURE.md §2.4 依赖方向** + +将表格中 `core/` 的"可依赖"列从 `标准库、typing、pluggy` 改为 `标准库、typing、pluggy、json_repair`。 + +- [ ] **Step 3: 更新 ARCHITECTURE.md §3.1 核心端口** + +将 §3.1 表格重构为两部分： + +**共享端口（`core/protocols.py`，跨子包）：** + +| Protocol | 关键方法 | 职责 | +|----------|---------|------| +| `LLMProvider` | `chat()` | LLM 文本调用 | +| `VLMProvider` | `chat_with_images()` | VLM 图文调用 | +| `TelemetryRecorder` | `record_llm_call()` | LLM 调用遥测 | + +**Agent 专属端口（`core/agent/protocols.py`）：** + +| Protocol | 关键方法 | 职责 | +|----------|---------|------| +| `ToolDispatcher` | `dispatch(tool_name, args, context)` | Agent 工具调度 | +| `AgentLoopSpec` | `before_step/after_tool/after_step/on_finish` | pluggy 生命周期 | + +**Evolution 专属端口（`core/evolution/protocols.py`）：** + +| Protocol | 关键方法 | 职责 | +|----------|---------|------| +| `SkillStore` | `read_skill()`, `write_skill()`, `list_versions()` | 版本化技能存储 | +| `PromptStore` | `read_prompt()`, `write_prompt()` | 版本化提示词存储 | +| `RunLog` | `insert()`, `query()` | 实验日志 | + +- [ ] **Step 4: 更新 ARCHITECTURE.md §4 遥测规范** + +在遥测字段表中新增三行： + +| 字段 | 类型 | 说明 | +|------|------|------| +| `thinking` | str | thinking/reasoning 内容 | +| `ttft_ms` | float? | 首 token 延迟（流式测量） | +| `max_inter_token_ms` | float? | 最大 token 间隔（流式测量） | + +- [ ] **Step 5: 更新 ARCHITECTURE.md §5 韧性治理** + +五层表格改为四层（删除 ARQ 行），新增流式三层看门狗段落： + +| 层 | 机制 | 说明 | +|---|------|------| +| 1 | 熔断器 | 连续 N 失败 → 短路 M 秒 → 探针恢复（`adapters/breaker.py`） | +| 2 | Redis 响应缓存 | content-addressed：`hash(model + messages)` → response | +| 3 | 流式三层看门狗 | TTFT / inter_token / total 超时保护（`adapters/streaming.py`） | +| 4 | 指数退避重试 | `max_retries`、`base_delay`、`max_delay`（可配置） | + +- [ ] **Step 6: 更新 CLAUDE.md §4.8 遥测 + §4.9 韧性 + §5 结构** + +§4.8 遥测必录字段表新增 `thinking`、`ttft_ms`、`max_inter_token_ms`。 + +§4.9 韧性表格同步为四层（与 ARCHITECTURE.md §5 一致），删除 ARQ 行，新增流式看门狗说明。 + +§5 项目结构树 `core/` 下新增 `protocols.py` 行。 + +- [ ] **Step 7: 更新 .env.example** + +在 `# ── Redis ──` 段落注释从 `响应缓存 + ARQ 任务队列` 改为 `响应缓存`。 + +在文件末尾 `# ── LLM 韧性参数 ──` 段落中新增： + +```bash +LLM_TTFT_TIMEOUT=30 +LLM_INTER_TOKEN_TIMEOUT=15 +LLM_RETRY_MAX_DELAY=30.0 +REDIS_CACHE_TTL=86400 +``` + +- [ ] **Step 8: 提交** + +```bash +git add research-wiki/ARCHITECTURE.md CLAUDE.md .env.example +git commit -m "docs: 同步 core/protocols.py 分层、四层治理栈、遥测新字段 + +ARCHITECTURE.md §2.3/§2.4/§3.1/§4/§5 + CLAUDE.md §4.8/§4.9/§5 + .env.example +对应设计 research-wiki/designs/2026-07-06-core-agent-adapters-llm-design.md §8" +``` + +--- + +### Task 2: core/types.py — LLMResponse + +**Files:** +- Modify: `core/types.py` +- Test: `tests/unit/test_core_types.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_core_types.py +"""core/types.py 单元测试。""" +from __future__ import annotations + +import copy + +import pytest + +from core.types import LLMResponse + + +class TestLLMResponse: + """LLMResponse 不可变性与字段完整性。""" + + @pytest.fixture() + def sample_response(self) -> LLMResponse: + return LLMResponse( + content="回答内容", + thinking="思考过程", + model="deepseek-v4-pro", + provider="deepseek", + prompt_tokens=100, + completion_tokens=50, + latency_ms=1200, + ttft_ms=350.0, + max_inter_token_ms=45.0, + cache_hit=False, + call_id="test-uuid-001", + ) + + def test_frozen_prevents_mutation(self, sample_response: LLMResponse) -> None: + """frozen=True 阻止属性赋值。""" + with pytest.raises(AttributeError): + sample_response.content = "篡改" # type: ignore[misc] + + def test_all_fields_accessible(self, sample_response: LLMResponse) -> None: + """所有字段均可读取。""" + assert sample_response.content == "回答内容" + assert sample_response.thinking == "思考过程" + assert sample_response.model == "deepseek-v4-pro" + assert sample_response.provider == "deepseek" + assert sample_response.prompt_tokens == 100 + assert sample_response.completion_tokens == 50 + assert sample_response.latency_ms == 1200 + assert sample_response.ttft_ms == 350.0 + assert sample_response.max_inter_token_ms == 45.0 + assert sample_response.cache_hit is False + assert sample_response.call_id == "test-uuid-001" + + def test_cache_hit_response_has_none_ttft(self) -> None: + """缓存命中时 ttft_ms 和 max_inter_token_ms 为 None。""" + resp = LLMResponse( + content="cached", + thinking="", + model="m", + provider="p", + prompt_tokens=0, + completion_tokens=0, + latency_ms=1, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=True, + call_id="c", + ) + assert resp.ttft_ms is None + assert resp.max_inter_token_ms is None + assert resp.cache_hit is True +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_core_types.py -v` +Expected: FAIL — `ImportError: cannot import name 'LLMResponse' from 'core.types'` + +- [ ] **Step 3: 实现 LLMResponse** + +```python +# core/types.py +"""跨模块共享类型。""" +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class LLMResponse: + """LLM/VLM 调用的统一返回值。 + + 由 adapters 层生成，core 层消费。frozen=True 确保响应不可变。 + """ + + content: str + thinking: str + model: str + provider: str + prompt_tokens: int + completion_tokens: int + latency_ms: int + ttft_ms: float | None + max_inter_token_ms: float | None + cache_hit: bool + call_id: str +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_core_types.py -v` +Expected: 3 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add core/types.py tests/unit/test_core_types.py +git commit -m "feat(core): 添加 LLMResponse frozen dataclass + +含 content/thinking/ttft_ms/max_inter_token_ms/call_id 等 11 个字段。" +``` + +--- + +### Task 3: core/protocols.py — 共享端口 + +**Files:** +- Create: `core/protocols.py` +- Test: `tests/unit/test_core_protocols.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_core_protocols.py +"""core/protocols.py 单元测试 — 验证 Protocol 可 runtime_checkable。""" +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from core.protocols import LLMProvider, TelemetryRecorder, VLMProvider +from core.types import LLMResponse + + +class _FakeLLM: + async def chat( + self, + messages: list[dict[str, Any]], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + return LLMResponse( + content="ok", + thinking="", + model="m", + provider="p", + prompt_tokens=1, + completion_tokens=1, + latency_ms=1, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + call_id="c", + ) + + +class _FakeVLM: + async def chat_with_images( + self, + messages: list[dict[str, Any]], + images: list[str | Path], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + return LLMResponse( + content="ok", + thinking="", + model="m", + provider="p", + prompt_tokens=1, + completion_tokens=1, + latency_ms=1, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + call_id="c", + ) + + +class _FakeTelemetry: + async def record_llm_call( + self, + *, + call_id: str, + parent_call_id: str | None, + session_id: str | None, + model_name: str, + provider: str, + messages: str, + response: str, + thinking: str, + prompt_tokens: int, + completion_tokens: int, + latency_ms: int, + ttft_ms: float | None, + max_inter_token_ms: float | None, + cache_hit: bool, + error: str | None, + ) -> None: + pass + + +def test_fake_llm_satisfies_protocol() -> None: + assert isinstance(_FakeLLM(), LLMProvider) + + +def test_fake_vlm_satisfies_protocol() -> None: + assert isinstance(_FakeVLM(), VLMProvider) + + +def test_fake_telemetry_satisfies_protocol() -> None: + assert isinstance(_FakeTelemetry(), TelemetryRecorder) + + +def test_plain_object_does_not_satisfy() -> None: + assert not isinstance(object(), LLMProvider) + assert not isinstance(object(), VLMProvider) + assert not isinstance(object(), TelemetryRecorder) +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_core_protocols.py -v` +Expected: FAIL — `ImportError: cannot import name 'LLMProvider' from 'core.protocols'` + +- [ ] **Step 3: 实现 core/protocols.py** + +```python +# core/protocols.py +"""共享 Protocol 端口定义。 + +LLMProvider / VLMProvider / TelemetryRecorder 是跨子包共享接口， +被 core/agent/、core/evolution/、app/ 各模块引用。 +adapters/ 提供具体实现。 +""" +from __future__ import annotations + +from pathlib import Path +from typing import Any, Protocol, runtime_checkable + +from core.types import LLMResponse + + +@runtime_checkable +class LLMProvider(Protocol): + """LLM 文本调用端口。""" + + async def chat( + self, + messages: list[dict[str, Any]], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: ... + + +@runtime_checkable +class VLMProvider(Protocol): + """VLM 图文调用端口。""" + + async def chat_with_images( + self, + messages: list[dict[str, Any]], + images: list[str | Path], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: ... + + +@runtime_checkable +class TelemetryRecorder(Protocol): + """LLM 调用遥测记录端口。""" + + async def record_llm_call( + self, + *, + call_id: str, + parent_call_id: str | None, + session_id: str | None, + model_name: str, + provider: str, + messages: str, + response: str, + thinking: str, + prompt_tokens: int, + completion_tokens: int, + latency_ms: int, + ttft_ms: float | None, + max_inter_token_ms: float | None, + cache_hit: bool, + error: str | None, + ) -> None: ... +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_core_protocols.py -v` +Expected: 4 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add core/protocols.py tests/unit/test_core_protocols.py +git commit -m "feat(core): 添加共享 Protocol 端口 + +LLMProvider / VLMProvider / TelemetryRecorder，全部 runtime_checkable。" +``` + +--- + +### Task 4: core/agent/types.py — Step + LoopResult + +**Files:** +- Create: `core/agent/types.py` +- Test: `tests/unit/test_agent_types.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_agent_types.py +"""core/agent/types.py 单元测试。""" +from __future__ import annotations + +from core.agent.types import LoopResult, Step + + +class TestStep: + def test_creation(self) -> None: + step = Step( + thought="thinking...", + reflect={"key": "value"}, + plan={"next": "do something"}, + tool_call={"tool": "search", "args": {"query": "test"}}, + tool_output="result text", + raw_content='{"reflect": {}, "plan": {}, "action": {}}', + call_id="uuid-1", + ) + assert step.thought == "thinking..." + assert step.tool_call["tool"] == "search" + assert step.call_id == "uuid-1" + + +class TestLoopResult: + def test_defaults(self) -> None: + lr = LoopResult() + assert lr.result is None + assert lr.steps == [] + assert lr.steps_used == 0 + assert lr.token_usage == {"prompt_tokens": 0, "completion_tokens": 0} + assert lr.stop_reason == "finished" + + def test_with_steps(self) -> None: + step = Step( + thought="t", + reflect={}, + plan={}, + tool_call={"tool": "t", "args": {}}, + tool_output="o", + raw_content="r", + call_id="c", + ) + lr = LoopResult( + result={"answer": "42"}, + steps=[step], + steps_used=1, + token_usage={"prompt_tokens": 100, "completion_tokens": 50}, + stop_reason="finished", + ) + assert lr.result == {"answer": "42"} + assert len(lr.steps) == 1 + assert lr.steps_used == 1 + + def test_separate_instances_have_independent_lists(self) -> None: + """default_factory 确保每个实例有独立的 steps 列表。""" + lr1 = LoopResult() + lr2 = LoopResult() + lr1.steps.append( + Step("t", {}, {}, {"tool": "x", "args": {}}, "o", "r", "c") + ) + assert len(lr2.steps) == 0 +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_agent_types.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 core/agent/types.py** + +```python +# core/agent/types.py +"""AgentLoop 数据类型。""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class Step: + """Agent 单步决策记录。""" + + thought: str + reflect: dict[str, Any] + plan: dict[str, Any] + tool_call: dict[str, Any] + tool_output: str + raw_content: str + call_id: str + + +@dataclass +class LoopResult: + """AgentLoop 完整运行结果。""" + + result: dict[str, Any] | None = None + steps: list[Step] = field(default_factory=list) + steps_used: int = 0 + token_usage: dict[str, int] = field( + default_factory=lambda: {"prompt_tokens": 0, "completion_tokens": 0} + ) + stop_reason: str = "finished" +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_agent_types.py -v` +Expected: 4 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add core/agent/types.py tests/unit/test_agent_types.py +git commit -m "feat(core/agent): 添加 Step 和 LoopResult 数据类 + +保真 TRM4 算法 #11，Step 新增 call_id 字段。" +``` + +--- + +### Task 5: core/agent/protocols.py — ToolDispatcher + AgentLoopSpec + +**Files:** +- Create: `core/agent/protocols.py` +- Test: `tests/unit/test_agent_protocols.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_agent_protocols.py +"""core/agent/protocols.py 单元测试。""" +from __future__ import annotations + +from typing import Any + +import pluggy + +from core.agent.protocols import AgentLoopSpec, ToolDispatcher + + +class _FakeDispatcher: + async def dispatch( + self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any] + ) -> str: + return f"executed {tool_name}" + + +def test_fake_dispatcher_satisfies_protocol() -> None: + assert isinstance(_FakeDispatcher(), ToolDispatcher) + + +def test_plain_object_not_dispatcher() -> None: + assert not isinstance(object(), ToolDispatcher) + + +def test_hookspec_can_register() -> None: + """验证 AgentLoopSpec 可被 pluggy 注册为 hookspec。""" + pm = pluggy.PluginManager("agent_loop") + pm.add_hookspecs(AgentLoopSpec) + assert pm.hook.before_step is not None + assert pm.hook.after_tool is not None + assert pm.hook.after_step is not None + assert pm.hook.on_finish is not None +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_agent_protocols.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 core/agent/protocols.py** + +```python +# core/agent/protocols.py +"""Agent 专属 Protocol 端口。""" +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + +import pluggy + +from core.agent.types import LoopResult, Step + +hookspec = pluggy.HookspecMarker("agent_loop") +hookimpl = pluggy.HookimplMarker("agent_loop") + + +@runtime_checkable +class ToolDispatcher(Protocol): + """Agent 工具调度端口。无效工具名抛 ValueError。""" + + async def dispatch( + self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any] + ) -> str: ... + + +class AgentLoopSpec: + """AgentLoop 生命周期扩展点。 + + 每个 hookimpl 可选择观察（返回 None）或变换（返回值）。 + """ + + @hookspec + async def before_step( + self, iteration: int, messages: list[dict[str, Any]] + ) -> None: ... + + @hookspec + async def after_tool( + self, iteration: int, step: Step + ) -> str | None: ... + + @hookspec + async def after_step( + self, iteration: int, messages: list[dict[str, Any]] + ) -> None: ... + + @hookspec + async def on_finish(self, result: LoopResult) -> None: ... +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_agent_protocols.py -v` +Expected: 3 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add core/agent/protocols.py tests/unit/test_agent_protocols.py +git commit -m "feat(core/agent): 添加 ToolDispatcher Protocol 和 AgentLoopSpec hookspec + +ToolDispatcher async + context 参数。AgentLoopSpec 四个 async 生命周期 hook。" +``` + +--- + +### Task 6: core/agent/loop.py — AgentLoop 引擎 + +**Files:** +- Create: `core/agent/loop.py` +- Test: `tests/unit/test_agent_loop.py` + +**保真校验：本任务迁移 TRM4 核心算法 #11（Agent Loop）。实现时必须逐行比对 `/home/iomgaa/Projects/Video-Tree-TRM4/core/loop.py` 的以下逻辑：json_repair 解析、解析失败纠正 prompt、submit_answer 终止、pluggy hook 调用时机（4 个 before_step + 4 个 after_step + 4 个 on_finish 调用点）、无效工具调用不计入 step_count、messages 组装格式。** + +- [ ] **Step 1: 写失败测试 — 正常 submit_answer 终止** + +```python +# tests/unit/test_agent_loop.py +"""core/agent/loop.py 单元测试。""" +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from core.agent.loop import AgentLoop +from core.agent.types import LoopResult +from core.types import LLMResponse + + +def _make_response(content: str, thinking: str = "") -> LLMResponse: + """构造测试用 LLMResponse。""" + return LLMResponse( + content=content, + thinking=thinking, + model="test-model", + provider="test", + prompt_tokens=10, + completion_tokens=10, + latency_ms=100, + ttft_ms=50.0, + max_inter_token_ms=10.0, + cache_hit=False, + call_id="test-call-id", + ) + + +class _StubDispatcher: + """总是返回固定输出的工具调度器。""" + + async def dispatch( + self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any] + ) -> str: + if tool_name == "submit_answer": + return "答案已提交" + if tool_name == "search_tree": + return "搜索结果: 找到节点 L2-3" + raise ValueError(f"未知工具: {tool_name}") + + +@pytest.mark.asyncio +async def test_submit_answer_terminates_loop() -> None: + """submit_answer 应终止循环并返回 finished。""" + llm = AsyncMock() + llm.chat = AsyncMock( + return_value=_make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {"answer": "42"}}}' + ) + ) + loop = AgentLoop(llm=llm, max_steps=10) + result = await loop.run( + system_prompt="你是助手", + user_prompt="问题", + tool_dispatcher=_StubDispatcher(), + ) + assert result.stop_reason == "finished" + assert result.result == {"answer": "42"} + assert result.steps_used == 1 + + +@pytest.mark.asyncio +async def test_budget_exceeded() -> None: + """超过 max_steps 应返回 budget_exceeded。""" + call_count = 0 + + async def fake_chat(messages, *, session_id=None, parent_call_id=None): + nonlocal call_count + call_count += 1 + return _make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "search_tree", "args": {"query": "test"}}}' + ) + + llm = AsyncMock() + llm.chat = fake_chat + loop = AgentLoop(llm=llm, max_steps=3) + result = await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + ) + assert result.stop_reason == "budget_exceeded" + assert result.steps_used == 3 + + +@pytest.mark.asyncio +async def test_invalid_tool_not_counted_as_step() -> None: + """无效工具调用（ValueError）不计入 steps_used。""" + responses = [ + _make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "unknown_tool", "args": {}}}' + ), + _make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {"answer": "ok"}}}' + ), + ] + call_idx = 0 + + async def fake_chat(messages, *, session_id=None, parent_call_id=None): + nonlocal call_idx + resp = responses[call_idx] + call_idx += 1 + return resp + + llm = AsyncMock() + llm.chat = fake_chat + loop = AgentLoop(llm=llm, max_steps=10) + result = await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + ) + assert result.stop_reason == "finished" + assert result.steps_used == 1 # unknown_tool 不计 + + +@pytest.mark.asyncio +async def test_parse_error_after_max_retries() -> None: + """连续 JSON 解析失败超过 max_retries 应返回 parse_error。""" + llm = AsyncMock() + llm.chat = AsyncMock( + return_value=_make_response("这不是JSON，完全无法解析") + ) + loop = AgentLoop(llm=llm, max_steps=10, max_retries=2) + result = await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + ) + assert result.stop_reason == "parse_error" + assert result.steps_used == 0 + + +@pytest.mark.asyncio +async def test_json_repair_handles_malformed() -> None: + """json_repair 应修复轻微 JSON 问题。""" + llm = AsyncMock() + # 缺少末尾引号 — json_repair 可修复 + llm.chat = AsyncMock( + return_value=_make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {"answer": "repaired}}}' + ) + ) + loop = AgentLoop(llm=llm, max_steps=10) + result = await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + ) + assert result.stop_reason == "finished" + + +@pytest.mark.asyncio +async def test_thinking_content_captured_in_step() -> None: + """LLMResponse.thinking 应透传到 Step.thought。""" + llm = AsyncMock() + llm.chat = AsyncMock( + return_value=_make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {}}}', + thinking="我在深度思考这个问题", + ) + ) + loop = AgentLoop(llm=llm, max_steps=10) + result = await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + ) + assert result.steps[0].thought == "我在深度思考这个问题" + + +@pytest.mark.asyncio +async def test_token_usage_accumulated() -> None: + """多步调用 token 应累加。""" + responses = [ + _make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "search_tree", "args": {"query": "q"}}}' + ), + _make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {}}}' + ), + ] + idx = 0 + + async def fake_chat(messages, *, session_id=None, parent_call_id=None): + nonlocal idx + r = responses[idx] + idx += 1 + return r + + llm = AsyncMock() + llm.chat = fake_chat + loop = AgentLoop(llm=llm, max_steps=10) + result = await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + ) + assert result.token_usage["prompt_tokens"] == 20 # 10 * 2 + assert result.token_usage["completion_tokens"] == 20 + + +@pytest.mark.asyncio +async def test_call_id_propagated_to_step() -> None: + """LLMResponse.call_id 应记录到 Step.call_id。""" + llm = AsyncMock() + llm.chat = AsyncMock( + return_value=_make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {}}}' + ) + ) + loop = AgentLoop(llm=llm, max_steps=10) + result = await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + ) + assert result.steps[0].call_id == "test-call-id" + + +@pytest.mark.asyncio +async def test_pluggy_hooks_called() -> None: + """pluggy 生命周期 hook 应被调用。""" + from core.agent.protocols import hookimpl + + class TrackingPlugin: + def __init__(self): + self.events = [] + + @hookimpl + async def before_step(self, iteration, messages): + self.events.append(("before_step", iteration)) + + @hookimpl + async def after_tool(self, iteration, step): + self.events.append(("after_tool", iteration)) + return None + + @hookimpl + async def after_step(self, iteration, messages): + self.events.append(("after_step", iteration)) + + @hookimpl + async def on_finish(self, result): + self.events.append(("on_finish", result.stop_reason)) + + llm = AsyncMock() + llm.chat = AsyncMock( + return_value=_make_response( + '{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {}}}' + ) + ) + plugin = TrackingPlugin() + loop = AgentLoop(llm=llm, max_steps=10) + await loop.run( + system_prompt="s", + user_prompt="u", + tool_dispatcher=_StubDispatcher(), + plugins=[plugin], + ) + assert ("before_step", 0) in plugin.events + assert ("after_tool", 0) in plugin.events + assert ("after_step", 0) in plugin.events + assert ("on_finish", "finished") in plugin.events +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_agent_loop.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 core/agent/loop.py** + +**保真校验检查点**：实现前必须打开 `/home/iomgaa/Projects/Video-Tree-TRM4/core/loop.py` 逐行比对以下逻辑： +- `_parse_response` 方法（TRM4 lines 259-293）：`repair_json` → `json.loads` → 校验 `action`/`tool`/`args` 存在 +- `_execute_tool` 方法（TRM4 lines 295-315）：`ValueError` 捕获 → `(output, False)` +- `_build_feedback` 方法（TRM4 lines 317-337）：`[工具执行结果: {name}]` 格式 +- `run` 方法（TRM4 lines 88-224）：messages 格式、retry prompt 文案、`step_count` 计数逻辑、四个终止路径 + +```python +# core/agent/loop.py +"""AgentLoop 推理循环引擎。 + +保真 TRM4 算法 #11：Thinking+JSON 格式、json_repair 兜底解析、 +pluggy hook 生命周期、submit_answer 终止。 +""" +from __future__ import annotations + +import json +from typing import Any + +import pluggy +from json_repair import repair_json +from loguru import logger + +from core.agent.protocols import AgentLoopSpec, ToolDispatcher +from core.agent.types import LoopResult, Step +from core.protocols import LLMProvider +from core.types import LLMResponse + +_RETRY_PROMPT = ( + "你的输出不是合法 JSON。请严格输出 JSON 格式：" + '{"reflect": {...}, "plan": {...}, ' + '"action": {"tool": "...", "args": {...}}}' +) + + +class AgentLoop: + """Thinking+JSON Agent 推理循环。 + + Args: + llm: LLM 调用端口（由 adapters 层实现治理栈）。 + max_steps: 最大有效工具调用步数。 + max_retries: 连续 JSON 解析失败上限。 + """ + + def __init__( + self, + llm: LLMProvider, + max_steps: int, + max_retries: int = 3, + ) -> None: + self._llm = llm + self._max_steps = max_steps + self._max_retries = max_retries + + async def run( + self, + system_prompt: str, + user_prompt: str, + tool_dispatcher: ToolDispatcher, + plugins: list[Any] | None = None, + *, + session_id: str | None = None, + ) -> LoopResult: + """执行推理循环直到终止。 + + Args: + system_prompt: 系统提示词。 + user_prompt: 用户问题。 + tool_dispatcher: 工具调度器（无效工具名抛 ValueError）。 + plugins: pluggy 插件列表。 + session_id: 关联到遥测的会话 ID。 + + Returns: + LoopResult 包含所有步骤和终止原因。 + """ + pm = self._create_plugin_manager(plugins) + messages: list[dict[str, Any]] = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ] + steps: list[Step] = [] + step_count = 0 + token_usage: dict[str, int] = {"prompt_tokens": 0, "completion_tokens": 0} + retry_count = 0 + iteration = 0 + last_call_id: str | None = None + + while step_count < self._max_steps: + # Phase 1: before_step hook + await _call_hook(pm.hook.before_step, iteration=iteration, messages=messages) + + # Phase 2: LLM 调用 + try: + response = await self._llm.chat( + messages, + session_id=session_id, + parent_call_id=last_call_id, + ) + except Exception as exc: + logger.error("LLM 调用异常: {}", exc) + result = LoopResult( + steps=steps, + steps_used=step_count, + token_usage=token_usage, + stop_reason="error", + ) + await _call_hook(pm.hook.on_finish, result=result) + return result + + token_usage["prompt_tokens"] += response.prompt_tokens + token_usage["completion_tokens"] += response.completion_tokens + last_call_id = response.call_id + + # Phase 3: 解析响应 + parsed = self._parse_response(response) + if parsed is None: + retry_count += 1 + raw = response.content or "" + messages.append({"role": "assistant", "content": raw}) + messages.append({"role": "user", "content": _RETRY_PROMPT}) + if retry_count >= self._max_retries: + logger.warning( + "连续 {} 次 JSON 解析失败，终止循环", retry_count + ) + result = LoopResult( + steps=steps, + steps_used=step_count, + token_usage=token_usage, + stop_reason="parse_error", + ) + await _call_hook( + pm.hook.after_step, iteration=iteration, messages=messages + ) + await _call_hook(pm.hook.on_finish, result=result) + return result + await _call_hook( + pm.hook.after_step, iteration=iteration, messages=messages + ) + iteration += 1 + continue + + retry_count = 0 + thought, reflect, plan, raw_content, action = parsed + tool_name = action["tool"] + tool_args = action["args"] + + # Phase 4: 执行工具 + output, is_valid = await self._execute_tool( + tool_dispatcher, tool_name, tool_args, + context={"iteration": iteration, "session_id": session_id}, + ) + if not is_valid: + messages.append({ + "role": "user", + "content": f"[工具调用无效: {tool_name}] {output}", + }) + await _call_hook( + pm.hook.after_step, iteration=iteration, messages=messages + ) + iteration += 1 + continue + + step_count += 1 + step = Step( + thought=thought, + reflect=reflect, + plan=plan, + tool_call={"tool": tool_name, "args": tool_args}, + tool_output=output, + raw_content=raw_content, + call_id=response.call_id, + ) + steps.append(step) + + # Phase 4.5: after_tool hook + 反馈组装 + hints = await _call_hook(pm.hook.after_tool, iteration=iteration, step=step) + feedback = self._build_feedback(tool_name, output, hints or []) + messages.append({"role": "assistant", "content": raw_content}) + messages.append(feedback) + + await _call_hook( + pm.hook.after_step, iteration=iteration, messages=messages + ) + + # Phase 5: 终止检查 + if tool_name == "submit_answer": + result = LoopResult( + result=tool_args, + steps=steps, + steps_used=step_count, + token_usage=token_usage, + stop_reason="finished", + ) + await _call_hook(pm.hook.on_finish, result=result) + return result + + iteration += 1 + + # 预算耗尽 + result = LoopResult( + steps=steps, + steps_used=step_count, + token_usage=token_usage, + stop_reason="budget_exceeded", + ) + await _call_hook(pm.hook.on_finish, result=result) + return result + + def _create_plugin_manager( + self, plugins: list[Any] | None + ) -> pluggy.PluginManager: + pm = pluggy.PluginManager("agent_loop") + pm.add_hookspecs(AgentLoopSpec) + for plugin in plugins or []: + pm.register(plugin) + return pm + + def _parse_response( + self, response: LLMResponse + ) -> tuple[str, dict[str, Any], dict[str, Any], str, dict[str, Any]] | None: + """解析 LLM 响应为 (thought, reflect, plan, raw_content, action)。 + + 解析失败返回 None。使用 json_repair 兜底修复。 + """ + content = response.content or "" + if not content.strip(): + return None + + repaired = repair_json(content) + try: + data = json.loads(repaired) + except (json.JSONDecodeError, ValueError): + return None + + if not isinstance(data, dict) or "action" not in data: + return None + + action = data["action"] + if not isinstance(action, dict) or "tool" not in action or "args" not in action: + return None + + thought = response.thinking or "" + reflect = data.get("reflect", {}) + plan = data.get("plan", {}) + return thought, reflect, plan, content, action + + async def _execute_tool( + self, + dispatcher: ToolDispatcher, + name: str, + args: dict[str, Any], + *, + context: dict[str, Any], + ) -> tuple[str, bool]: + """执行工具调用。返回 (output, is_valid)。""" + try: + output = await dispatcher.dispatch(name, args, context=context) + return output, True + except ValueError as e: + return f"工具调用失败: {e}", False + + def _build_feedback( + self, tool_name: str, tool_output: str, hints: list[str | None] + ) -> dict[str, Any]: + """组装工具执行反馈消息。""" + parts = [f"[工具执行结果: {tool_name}]", tool_output] + for hint in hints: + if hint is not None: + parts.append(hint) + return {"role": "user", "content": "\n".join(parts)} + + +async def _call_hook(hook: Any, **kwargs: Any) -> Any: + """调用 pluggy hook，支持 async hookimpl。""" + results = hook(**kwargs) + if results is not None: + resolved = [] + for r in results: + if hasattr(r, "__await__"): + resolved.append(await r) + else: + resolved.append(r) + return resolved + return [] +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_agent_loop.py -v` +Expected: 9 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add core/agent/loop.py tests/unit/test_agent_loop.py +git commit -m "feat(core/agent): 实现 AgentLoop 推理循环引擎 + +保真 TRM4 算法 #11: json_repair 兜底、submit_answer 终止、 +pluggy hook 生命周期、无效工具不计步。" +``` + +--- + +### Task 7: adapters/streaming.py — 三层看门狗 + +**Files:** +- Create: `adapters/streaming.py` +- Test: `tests/unit/test_streaming.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_streaming.py +"""adapters/streaming.py 三层看门狗单元测试。""" +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator + +import pytest + +from adapters.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts + + +async def _items_with_delays( + items: list[tuple[bool, str]], delays: list[float] +) -> AsyncIterator[tuple[bool, str]]: + """按指定延迟逐个 yield items。""" + for item, delay in zip(items, delays): + await asyncio.sleep(delay) + yield item + + +@pytest.mark.asyncio +async def test_all_items_yielded_within_budget() -> None: + """全部 token 在预算内应正常 yield。""" + items = [(True, "a"), (True, "b"), (True, "c")] + delays = [0.01, 0.01, 0.01] + result = [] + async for chunk in stream_with_liveness_timeouts( + _items_with_delays(items, delays), + ttft_s=1.0, + inter_token_s=1.0, + total_s=5.0, + ): + result.append(chunk) + assert result == items + + +@pytest.mark.asyncio +async def test_ttft_timeout_fires() -> None: + """首 token 超过 ttft_s 应触发 ttft 超时。""" + items = [(True, "a")] + delays = [2.0] # 远超 ttft + with pytest.raises(StreamLivenessTimeout) as exc_info: + async for _ in stream_with_liveness_timeouts( + _items_with_delays(items, delays), + ttft_s=0.1, + inter_token_s=5.0, + total_s=10.0, + ): + pass + assert exc_info.value.kind == "ttft" + assert exc_info.value.first_token_seen is False + + +@pytest.mark.asyncio +async def test_inter_token_timeout_fires() -> None: + """token 间隔超过 inter_token_s 应触发 inter_token 超时。""" + items = [(True, "a"), (True, "b")] + delays = [0.01, 2.0] # 第二个 token 延迟过大 + collected = [] + with pytest.raises(StreamLivenessTimeout) as exc_info: + async for chunk in stream_with_liveness_timeouts( + _items_with_delays(items, delays), + ttft_s=5.0, + inter_token_s=0.1, + total_s=10.0, + ): + collected.append(chunk) + assert exc_info.value.kind == "inter_token" + assert exc_info.value.first_token_seen is True + assert len(collected) == 1 # 第一个 token 已 yield + + +@pytest.mark.asyncio +async def test_total_timeout_fires() -> None: + """总时间超过 total_s 应触发 total 超时。""" + items = [(True, str(i)) for i in range(100)] + delays = [0.05] * 100 # 总计 5s，远超 total_s=0.2 + with pytest.raises(StreamLivenessTimeout) as exc_info: + async for _ in stream_with_liveness_timeouts( + _items_with_delays(items, delays), + ttft_s=5.0, + inter_token_s=5.0, + total_s=0.2, + ): + pass + assert exc_info.value.kind == "total" + + +@pytest.mark.asyncio +async def test_thinking_tokens_refresh_watchdog() -> None: + """thinking token (is_content=False) 应刷新看门狗但不影响内容判断。""" + items = [(False, "think1"), (False, "think2"), (True, "content")] + delays = [0.01, 0.01, 0.01] + result = [] + async for chunk in stream_with_liveness_timeouts( + _items_with_delays(items, delays), + ttft_s=1.0, + inter_token_s=1.0, + total_s=5.0, + ): + result.append(chunk) + assert result == items # 全部透传，包括 thinking tokens + + +@pytest.mark.asyncio +async def test_empty_stream_no_error() -> None: + """空流应正常结束。""" + async def empty() -> AsyncIterator[tuple[bool, str]]: + return + yield # noqa: make it an async generator + + result = [] + async for chunk in stream_with_liveness_timeouts( + empty(), + ttft_s=1.0, + inter_token_s=1.0, + total_s=5.0, + ): + result.append(chunk) + assert result == [] +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_streaming.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 adapters/streaming.py** + +```python +# adapters/streaming.py +"""流式 LLM 响应三层看门狗。 + +参考 CHSAnalyzer2 app/providers/streaming.py。 +三层超时：TTFT（首 token）、inter_token（token 间隔）、total（总时长）。 +只 wrap __anext__，不 wrap yield，防取消泄漏。 +""" +from __future__ import annotations + +import asyncio +import contextlib +import time +from collections.abc import AsyncIterator +from typing import TypeVar + +_T = TypeVar("_T") + + +class StreamLivenessTimeout(Exception): + """流活性超时异常。""" + + def __init__( + self, kind: str, elapsed_s: float, first_token_seen: bool + ) -> None: + self.kind = kind + self.elapsed_s = elapsed_s + self.first_token_seen = first_token_seen + super().__init__( + f"流活性超时({kind}, elapsed={elapsed_s:.1f}s)" + ) + + +async def _anext_within( + it: AsyncIterator[_T], + timeout_s: float, + *, + kind: str, + start: float, + first: bool, +) -> _T: + """在 timeout_s 内获取下一个元素，超时抛 StreamLivenessTimeout。""" + try: + async with asyncio.timeout(timeout_s) as cm: + return await it.__anext__() + except TimeoutError: + if not cm.expired(): + raise + raise StreamLivenessTimeout( + kind=kind, + elapsed_s=time.monotonic() - start, + first_token_seen=not first, + ) from None + + +async def stream_with_liveness_timeouts( + source: AsyncIterator[_T], + *, + ttft_s: float, + inter_token_s: float, + total_s: float, +) -> AsyncIterator[_T]: + """三层看门狗包装异步迭代器。 + + Args: + source: 被包装的异步迭代器。 + ttft_s: 首 token 超时（秒）。 + inter_token_s: token 间隔超时（秒）。 + total_s: 总时长超时（秒）。 + + Yields: + 源迭代器的元素。 + + Raises: + StreamLivenessTimeout: 任一层超时触发。 + """ + it = source.__aiter__() + start = time.monotonic() + deadline = start + total_s + first = True + + try: + while True: + remaining_total = deadline - time.monotonic() + if remaining_total <= 0: + raise StreamLivenessTimeout( + kind="total", + elapsed_s=time.monotonic() - start, + first_token_seen=not first, + ) + + budget = ttft_s if first else inter_token_s + if remaining_total <= budget: + actual_timeout = remaining_total + kind = "total" + else: + actual_timeout = budget + kind = "ttft" if first else "inter_token" + + try: + item = await _anext_within( + it, actual_timeout, kind=kind, start=start, first=first + ) + except StopAsyncIteration: + return + + first = False + yield item + finally: + with contextlib.suppress(Exception): + if hasattr(it, "aclose"): + await it.aclose() +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_streaming.py -v` +Expected: 6 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add adapters/streaming.py tests/unit/test_streaming.py +git commit -m "feat(adapters): 实现三层流式看门狗 + +TTFT / inter_token / total 超时保护。 +参考 CHSAnalyzer2 streaming.py。" +``` + +--- + +### Task 8: adapters/breaker.py — 熔断器 + +**Files:** +- Create: `adapters/breaker.py` +- Test: `tests/unit/test_breaker.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_breaker.py +"""adapters/breaker.py 熔断器单元测试。""" +from __future__ import annotations + +from adapters.breaker import CircuitBreaker + + +class TestCircuitBreaker: + def test_closed_by_default(self) -> None: + cb = CircuitBreaker(fail_threshold=3, cooldown_s=60.0) + assert cb.is_open(now=0.0) is False + + def test_opens_after_threshold_failures(self) -> None: + cb = CircuitBreaker(fail_threshold=3, cooldown_s=60.0) + cb.record_failure(now=1.0) + cb.record_failure(now=2.0) + assert cb.is_open(now=2.5) is False # 2 < 3 + cb.record_failure(now=3.0) + assert cb.is_open(now=3.5) is True # 3 >= 3 + + def test_half_open_after_cooldown(self) -> None: + cb = CircuitBreaker(fail_threshold=2, cooldown_s=10.0) + cb.record_failure(now=0.0) + cb.record_failure(now=1.0) + assert cb.is_open(now=5.0) is True # 5 < 11 + assert cb.is_open(now=11.0) is False # 11 >= 11, 半开 + + def test_success_resets(self) -> None: + cb = CircuitBreaker(fail_threshold=2, cooldown_s=10.0) + cb.record_failure(now=0.0) + cb.record_failure(now=1.0) + assert cb.is_open(now=2.0) is True + cb.record_success() + assert cb.is_open(now=2.0) is False + + def test_force_open_immediate(self) -> None: + cb = CircuitBreaker(fail_threshold=5, cooldown_s=30.0) + cb.force_open(now=10.0) + assert cb.is_open(now=10.5) is True + assert cb.is_open(now=40.0) is False # 40 >= 40, 半开 + + def test_force_open_probe_failure_reopens(self) -> None: + """force_open 后半开探针失败应重新熔断。""" + cb = CircuitBreaker(fail_threshold=3, cooldown_s=10.0) + cb.force_open(now=0.0) + # 半开后再失败一次 + assert cb.is_open(now=10.0) is False # 半开 + cb.record_failure(now=10.0) # fail count 已经是 3，+1 = 4 >= 3 + assert cb.is_open(now=10.5) is True # 重新熔断 +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_breaker.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 adapters/breaker.py** + +```python +# adapters/breaker.py +"""内存级熔断器。 + +参考 CHSAnalyzer2 app/providers/breaker.py。 +注入 now 参数，纯确定性可测试。 +""" +from __future__ import annotations + + +class CircuitBreaker: + """单进程内存级熔断器。 + + 连续失败达到阈值后熔断，冷却后半开允许一次探针请求。 + + Args: + fail_threshold: 连续失败触发熔断的次数。 + cooldown_s: 熔断后冷却秒数。 + """ + + def __init__(self, fail_threshold: int, cooldown_s: float) -> None: + self._fail_threshold = fail_threshold + self._cooldown_s = cooldown_s + self._fails: int = 0 + self._open_until: float | None = None + + def is_open(self, now: float) -> bool: + """检查熔断器是否开启。冷却过期后自动半开。""" + if self._open_until is None: + return False + return now < self._open_until + + def record_failure(self, now: float) -> None: + """记录一次失败。达到阈值时熔断。""" + self._fails += 1 + if self._fails >= self._fail_threshold: + self._open_until = now + self._cooldown_s + + def record_success(self) -> None: + """记录一次成功。重置失败计数，关闭熔断器。""" + self._fails = 0 + self._open_until = None + + def force_open(self, now: float) -> None: + """强制熔断（用于 401/403 等致命错误）。 + + 预设失败计数为阈值，确保半开探针失败后重新熔断。 + """ + self._fails = self._fail_threshold + self._open_until = now + self._cooldown_s +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_breaker.py -v` +Expected: 6 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add adapters/breaker.py tests/unit/test_breaker.py +git commit -m "feat(adapters): 实现 CircuitBreaker 内存级熔断器 + +注入 now 纯确定性，force_open 支持 401/403 直接熔断。" +``` + +--- + +### Task 9: adapters/redis_cache.py — Redis 响应缓存 + +**Files:** +- Create: `adapters/redis_cache.py` +- Test: `tests/unit/test_redis_cache.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_redis_cache.py +"""adapters/redis_cache.py 单元测试。使用 fakeredis 替代真实 Redis。""" +from __future__ import annotations + +import pytest + +from adapters.redis_cache import RedisResponseCache +from core.types import LLMResponse + +pytest_plugins = ["pytest_asyncio"] + + +def _sample_response() -> LLMResponse: + return LLMResponse( + content="answer", + thinking="thought", + model="deepseek-v4-pro", + provider="deepseek", + prompt_tokens=100, + completion_tokens=50, + latency_ms=1200, + ttft_ms=350.0, + max_inter_token_ms=45.0, + cache_hit=False, + call_id="original-call-id", + ) + + +@pytest.fixture() +def fake_redis(): + """使用 fakeredis 提供内存 Redis。""" + try: + import fakeredis.aioredis + except ImportError: + pytest.skip("fakeredis not installed") + return fakeredis.aioredis.FakeRedis(decode_responses=True) + + +@pytest.mark.asyncio +async def test_cache_miss_returns_none(fake_redis) -> None: + cache = RedisResponseCache(redis=fake_redis, ttl_s=3600) + result = await cache.get(model="m", messages=[{"role": "user", "content": "q"}]) + assert result is None + + +@pytest.mark.asyncio +async def test_cache_roundtrip(fake_redis) -> None: + cache = RedisResponseCache(redis=fake_redis, ttl_s=3600) + messages = [{"role": "user", "content": "hello"}] + resp = _sample_response() + await cache.set(model="deepseek-v4-pro", messages=messages, response=resp) + cached = await cache.get(model="deepseek-v4-pro", messages=messages) + assert cached is not None + assert cached.content == "answer" + assert cached.thinking == "thought" + assert cached.prompt_tokens == 100 + + +@pytest.mark.asyncio +async def test_different_messages_different_keys(fake_redis) -> None: + cache = RedisResponseCache(redis=fake_redis, ttl_s=3600) + resp = _sample_response() + await cache.set(model="m", messages=[{"role": "user", "content": "q1"}], response=resp) + result = await cache.get(model="m", messages=[{"role": "user", "content": "q2"}]) + assert result is None + + +@pytest.mark.asyncio +async def test_different_models_different_keys(fake_redis) -> None: + cache = RedisResponseCache(redis=fake_redis, ttl_s=3600) + resp = _sample_response() + await cache.set(model="model-a", messages=[{"role": "user", "content": "q"}], response=resp) + result = await cache.get(model="model-b", messages=[{"role": "user", "content": "q"}]) + assert result is None + + +@pytest.mark.asyncio +async def test_graceful_degradation_on_error() -> None: + """Redis 不可用时 get 返回 None、set 不报错。""" + + class _BrokenRedis: + async def get(self, key): + raise ConnectionError("redis down") + + async def set(self, key, value, ex=None): + raise ConnectionError("redis down") + + cache = RedisResponseCache(redis=_BrokenRedis(), ttl_s=3600) # type: ignore[arg-type] + result = await cache.get(model="m", messages=[]) + assert result is None + await cache.set(model="m", messages=[], response=_sample_response()) +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_redis_cache.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 adapters/redis_cache.py** + +```python +# adapters/redis_cache.py +"""Redis 响应缓存。content-addressed，key = sha256(model + messages)。""" +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict +from typing import Any + +from loguru import logger + +from core.types import LLMResponse + + +class RedisResponseCache: + """Redis LLM 响应缓存。 + + Redis 不可用时静默降级，不阻断主流程。 + + Args: + redis: redis.asyncio.Redis 实例（或兼容 duck-typed 对象）。 + ttl_s: 缓存条目 TTL 秒数。 + """ + + def __init__(self, redis: Any, ttl_s: int) -> None: + self._redis = redis + self._ttl_s = ttl_s + + async def get( + self, model: str, messages: list[dict[str, Any]] + ) -> LLMResponse | None: + """查询缓存。未命中或 Redis 不可用返回 None。""" + key = self._make_key(model, messages) + try: + raw = await self._redis.get(key) + except Exception as exc: + logger.warning("Redis 缓存读取失败，降级跳过: {}", exc) + return None + if raw is None: + return None + try: + data = json.loads(raw) + return LLMResponse(**data) + except (json.JSONDecodeError, TypeError) as exc: + logger.warning("Redis 缓存反序列化失败: {}", exc) + return None + + async def set( + self, + model: str, + messages: list[dict[str, Any]], + response: LLMResponse, + ) -> None: + """写入缓存。Redis 不可用时静默跳过。""" + key = self._make_key(model, messages) + try: + value = json.dumps(asdict(response), ensure_ascii=False) + await self._redis.set(key, value, ex=self._ttl_s) + except Exception as exc: + logger.warning("Redis 缓存写入失败，降级跳过: {}", exc) + + @staticmethod + def _make_key(model: str, messages: list[dict[str, Any]]) -> str: + """生成 content-addressed 缓存键。""" + payload = json.dumps( + {"model": model, "messages": messages}, + sort_keys=True, + ensure_ascii=False, + ) + digest = hashlib.sha256(payload.encode()).hexdigest() + return f"llm:cache:{digest}" +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_redis_cache.py -v` +Expected: 5 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add adapters/redis_cache.py tests/unit/test_redis_cache.py +git commit -m "feat(adapters): 实现 RedisResponseCache + +content-addressed sha256 缓存键，Redis 不可用时静默降级。" +``` + +--- + +### Task 10: adapters/telemetry.py — SQLite 遥测记录 + +**Files:** +- Create: `adapters/telemetry.py` +- Test: `tests/unit/test_telemetry.py` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_telemetry.py +"""adapters/telemetry.py 单元测试。""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from adapters.telemetry import SQLiteTelemetryRecorder +from core.protocols import TelemetryRecorder + + +@pytest.fixture() +def db_path(tmp_path: Path) -> Path: + return tmp_path / "telemetry.db" + + +@pytest.mark.asyncio +async def test_satisfies_protocol(db_path: Path) -> None: + recorder = SQLiteTelemetryRecorder(db_path=db_path) + assert isinstance(recorder, TelemetryRecorder) + + +@pytest.mark.asyncio +async def test_record_creates_table_and_inserts(db_path: Path) -> None: + recorder = SQLiteTelemetryRecorder(db_path=db_path) + await recorder.record_llm_call( + call_id="c1", + parent_call_id=None, + session_id="s1", + model_name="deepseek-v4-pro", + provider="deepseek", + messages='[{"role": "user", "content": "hi"}]', + response="hello", + thinking="let me think", + prompt_tokens=10, + completion_tokens=5, + latency_ms=500, + ttft_ms=100.0, + max_inter_token_ms=25.0, + cache_hit=False, + error=None, + ) + conn = sqlite3.connect(db_path) + rows = conn.execute("SELECT * FROM llm_calls").fetchall() + conn.close() + assert len(rows) == 1 + + +@pytest.mark.asyncio +async def test_record_with_error(db_path: Path) -> None: + recorder = SQLiteTelemetryRecorder(db_path=db_path) + await recorder.record_llm_call( + call_id="c2", + parent_call_id="c1", + session_id="s1", + model_name="m", + provider="p", + messages="[]", + response="", + thinking="", + prompt_tokens=0, + completion_tokens=0, + latency_ms=0, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + error="TimeoutError: stream stalled", + ) + conn = sqlite3.connect(db_path) + row = conn.execute( + "SELECT error FROM llm_calls WHERE call_id = 'c2'" + ).fetchone() + conn.close() + assert row[0] == "TimeoutError: stream stalled" + + +@pytest.mark.asyncio +async def test_record_cache_hit(db_path: Path) -> None: + recorder = SQLiteTelemetryRecorder(db_path=db_path) + await recorder.record_llm_call( + call_id="c3", + parent_call_id=None, + session_id=None, + model_name="m", + provider="p", + messages="[]", + response="cached", + thinking="", + prompt_tokens=10, + completion_tokens=5, + latency_ms=1, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=True, + error=None, + ) + conn = sqlite3.connect(db_path) + row = conn.execute( + "SELECT cache_hit FROM llm_calls WHERE call_id = 'c3'" + ).fetchone() + conn.close() + assert row[0] == 1 # SQLite boolean as int +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_telemetry.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 adapters/telemetry.py** + +```python +# adapters/telemetry.py +"""SQLite 遥测记录实现。 + +实现 core/protocols.py 的 TelemetryRecorder Protocol。 +SQLite 写入通过 asyncio.to_thread() 桥接。 +""" +from __future__ import annotations + +import asyncio +import sqlite3 +from pathlib import Path + +_CREATE_TABLE = """ +CREATE TABLE IF NOT EXISTS llm_calls ( + call_id TEXT PRIMARY KEY, + parent_call_id TEXT, + session_id TEXT, + model_name TEXT NOT NULL, + provider TEXT NOT NULL, + messages TEXT NOT NULL, + response TEXT NOT NULL, + thinking TEXT NOT NULL DEFAULT '', + prompt_tokens INTEGER NOT NULL, + completion_tokens INTEGER NOT NULL, + latency_ms INTEGER NOT NULL, + ttft_ms REAL, + max_inter_token_ms REAL, + cache_hit INTEGER NOT NULL DEFAULT 0, + error TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')) +) +""" + +_INSERT = """ +INSERT INTO llm_calls ( + call_id, parent_call_id, session_id, model_name, provider, + messages, response, thinking, prompt_tokens, completion_tokens, + latency_ms, ttft_ms, max_inter_token_ms, cache_hit, error +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +""" + + +class SQLiteTelemetryRecorder: + """SQLite 遥测记录器。 + + Args: + db_path: SQLite 数据库文件路径。 + """ + + def __init__(self, db_path: Path) -> None: + self._db_path = db_path + self._initialized = False + + def _ensure_table(self, conn: sqlite3.Connection) -> None: + if not self._initialized: + conn.execute(_CREATE_TABLE) + conn.commit() + self._initialized = True + + def _write( + self, + *, + call_id: str, + parent_call_id: str | None, + session_id: str | None, + model_name: str, + provider: str, + messages: str, + response: str, + thinking: str, + prompt_tokens: int, + completion_tokens: int, + latency_ms: int, + ttft_ms: float | None, + max_inter_token_ms: float | None, + cache_hit: bool, + error: str | None, + ) -> None: + conn = sqlite3.connect(self._db_path) + try: + self._ensure_table(conn) + conn.execute( + _INSERT, + ( + call_id, + parent_call_id, + session_id, + model_name, + provider, + messages, + response, + thinking, + prompt_tokens, + completion_tokens, + latency_ms, + ttft_ms, + max_inter_token_ms, + int(cache_hit), + error, + ), + ) + conn.commit() + finally: + conn.close() + + async def record_llm_call( + self, + *, + call_id: str, + parent_call_id: str | None, + session_id: str | None, + model_name: str, + provider: str, + messages: str, + response: str, + thinking: str, + prompt_tokens: int, + completion_tokens: int, + latency_ms: int, + ttft_ms: float | None, + max_inter_token_ms: float | None, + cache_hit: bool, + error: str | None, + ) -> None: + """记录一次 LLM 调用到 SQLite。""" + await asyncio.to_thread( + self._write, + call_id=call_id, + parent_call_id=parent_call_id, + session_id=session_id, + model_name=model_name, + provider=provider, + messages=messages, + response=response, + thinking=thinking, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + latency_ms=latency_ms, + ttft_ms=ttft_ms, + max_inter_token_ms=max_inter_token_ms, + cache_hit=cache_hit, + error=error, + ) +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_telemetry.py -v` +Expected: 4 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add adapters/telemetry.py tests/unit/test_telemetry.py +git commit -m "feat(adapters): 实现 SQLiteTelemetryRecorder + +asyncio.to_thread 桥接 SQLite，字段与 TelemetryRecorder Protocol 一一对应。" +``` + +--- + +### Task 11: adapters/llm.py — GovernedLLMClient + +**Files:** +- Create: `adapters/llm.py` +- Test: `tests/unit/test_governed_llm.py` + +这是最复杂的文件——组合四层治理栈 + 流式 SSE 消费 + provider 差异处理 + 遥测集成。 + +- [ ] **Step 1: 写失败测试** + +```python +# tests/unit/test_governed_llm.py +"""adapters/llm.py GovernedLLMClient 单元测试。 + +使用 mock httpx 响应模拟流式 SSE，不依赖真实 LLM API。 +""" +from __future__ import annotations + +import asyncio +import json +import time +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from adapters.breaker import CircuitBreaker +from adapters.llm import CircuitOpenError, GovernedLLMClient +from core.types import LLMResponse + + +class _FakeRedisCache: + """内存缓存替身。""" + + def __init__(self): + self._store: dict[str, LLMResponse] = {} + + async def get(self, model: str, messages: list[dict]) -> LLMResponse | None: + key = f"{model}:{json.dumps(messages, sort_keys=True)}" + return self._store.get(key) + + async def set( + self, model: str, messages: list[dict], response: LLMResponse + ) -> None: + key = f"{model}:{json.dumps(messages, sort_keys=True)}" + self._store[key] = response + + +class _FakeTelemetry: + def __init__(self): + self.calls: list[dict] = [] + + async def record_llm_call(self, **kwargs) -> None: + self.calls.append(kwargs) + + +def _make_sse_lines( + content_chunks: list[str], + reasoning_chunks: list[str] | None = None, + model: str = "test-model", + usage: dict | None = None, +) -> list[str]: + """构造 SSE 文本行列表。""" + lines = [] + for text in (reasoning_chunks or []): + chunk = { + "choices": [{"delta": {"reasoning_content": text}}], + "model": model, + } + lines.append(f"data: {json.dumps(chunk)}") + for text in content_chunks: + chunk = { + "choices": [{"delta": {"content": text}}], + "model": model, + } + lines.append(f"data: {json.dumps(chunk)}") + if usage: + chunk = {"choices": [], "model": model, "usage": usage} + lines.append(f"data: {json.dumps(chunk)}") + lines.append("data: [DONE]") + return lines + + +def _build_client( + *, + breaker: CircuitBreaker | None = None, + cache: _FakeRedisCache | None = None, + telemetry: _FakeTelemetry | None = None, +) -> GovernedLLMClient: + return GovernedLLMClient( + model="test-model", + base_url="https://api.test.com/v1", + api_key="sk-test", + provider="deepseek", + thinking=True, + breaker=breaker or CircuitBreaker(fail_threshold=3, cooldown_s=60.0), + cache=cache, + telemetry=telemetry or _FakeTelemetry(), + timeout_s=30.0, + ttft_timeout_s=10.0, + inter_token_timeout_s=5.0, + max_retries=2, + retry_base_delay_s=0.01, + retry_max_delay_s=0.05, + ) + + +@pytest.mark.asyncio +async def test_circuit_open_raises() -> None: + """熔断器开启时应直接抛出 CircuitOpenError。""" + breaker = CircuitBreaker(fail_threshold=1, cooldown_s=999.0) + breaker.force_open(now=time.monotonic()) + client = _build_client(breaker=breaker) + with pytest.raises(CircuitOpenError): + await client.chat([{"role": "user", "content": "hi"}]) + + +@pytest.mark.asyncio +async def test_cache_hit_returns_cached_and_records_telemetry() -> None: + """缓存命中应返回缓存内容并记录遥测。""" + cache = _FakeRedisCache() + telemetry = _FakeTelemetry() + messages = [{"role": "user", "content": "cached question"}] + cached_resp = LLMResponse( + content="cached answer", + thinking="", + model="test-model", + provider="deepseek", + prompt_tokens=10, + completion_tokens=5, + latency_ms=100, + ttft_ms=50.0, + max_inter_token_ms=20.0, + cache_hit=False, + call_id="old-id", + ) + await cache.set("test-model", messages, cached_resp) + client = _build_client(cache=cache, telemetry=telemetry) + result = await client.chat(messages) + assert result.content == "cached answer" + assert result.cache_hit is True + assert result.call_id != "old-id" # 新 call_id + assert len(telemetry.calls) == 1 + assert telemetry.calls[0]["cache_hit"] is True + + +@pytest.mark.asyncio +async def test_successful_streaming_call() -> None: + """正常流式调用应返回完整内容和指标。""" + telemetry = _FakeTelemetry() + client = _build_client(telemetry=telemetry) + sse_lines = _make_sse_lines( + reasoning_chunks=["think1", "think2"], + content_chunks=["hello", " world"], + usage={"prompt_tokens": 10, "completion_tokens": 5}, + ) + with patch.object(client, "_stream_request") as mock_stream: + mock_stream.return_value = _async_line_iter(sse_lines) + result = await client.chat([{"role": "user", "content": "hi"}]) + + assert result.content == "hello world" + assert result.thinking == "think1think2" + assert result.cache_hit is False + assert result.ttft_ms is not None + assert result.prompt_tokens == 10 + assert result.completion_tokens == 5 + assert len(telemetry.calls) == 1 + assert telemetry.calls[0]["error"] is None + + +async def _async_line_iter(lines: list[str]): + """模拟 httpx 流式响应的异步行迭代器。""" + for line in lines: + yield line + + +@pytest.mark.asyncio +async def test_transient_error_retries_and_records_telemetry() -> None: + """瞬态错误应重试并最终记录遥测。""" + telemetry = _FakeTelemetry() + client = _build_client(telemetry=telemetry) + call_count = 0 + + async def failing_stream(body): + nonlocal call_count + call_count += 1 + if call_count <= 2: + raise httpx.ConnectError("connection refused") + return _async_line_iter( + _make_sse_lines(content_chunks=["ok"], usage={"prompt_tokens": 1, "completion_tokens": 1}) + ) + + with patch.object(client, "_stream_request", side_effect=failing_stream): + result = await client.chat([{"role": "user", "content": "hi"}]) + assert result.content == "ok" + assert call_count == 3 # 2 failures + 1 success + + +@pytest.mark.asyncio +async def test_fatal_error_force_opens_breaker() -> None: + """401/403 应 force_open 熔断器。""" + breaker = CircuitBreaker(fail_threshold=5, cooldown_s=60.0) + telemetry = _FakeTelemetry() + client = _build_client(breaker=breaker, telemetry=telemetry) + + async def auth_failure(body): + resp = httpx.Response(401, request=httpx.Request("POST", "https://test.com")) + raise httpx.HTTPStatusError("Unauthorized", request=resp.request, response=resp) + + with patch.object(client, "_stream_request", side_effect=auth_failure): + with pytest.raises(httpx.HTTPStatusError): + await client.chat([{"role": "user", "content": "hi"}]) + assert breaker.is_open(time.monotonic()) is True + assert len(telemetry.calls) == 1 + assert telemetry.calls[0]["error"] is not None + + +@pytest.mark.asyncio +async def test_qwen_thinking_stripped() -> None: + """Qwen provider 的 标签应被剥离。""" + telemetry = _FakeTelemetry() + client = _build_client(telemetry=telemetry) + client._provider = "qwen" + content_with_think = "我在想最终答案" + sse_lines = _make_sse_lines( + content_chunks=[content_with_think], + usage={"prompt_tokens": 1, "completion_tokens": 1}, + ) + with patch.object(client, "_stream_request") as mock_stream: + mock_stream.return_value = _async_line_iter(sse_lines) + result = await client.chat([{"role": "user", "content": "hi"}]) + assert result.content == "最终答案" + assert result.thinking == "我在想" + + +@pytest.mark.asyncio +async def test_parent_call_id_forwarded_to_telemetry() -> None: + """parent_call_id 应传递到遥测记录。""" + telemetry = _FakeTelemetry() + client = _build_client(telemetry=telemetry) + sse_lines = _make_sse_lines( + content_chunks=["ok"], + usage={"prompt_tokens": 1, "completion_tokens": 1}, + ) + with patch.object(client, "_stream_request") as mock_stream: + mock_stream.return_value = _async_line_iter(sse_lines) + await client.chat( + [{"role": "user", "content": "hi"}], + parent_call_id="parent-123", + session_id="sess-456", + ) + assert telemetry.calls[0]["parent_call_id"] == "parent-123" + assert telemetry.calls[0]["session_id"] == "sess-456" +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_governed_llm.py -v` +Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 实现 adapters/llm.py** + +```python +# adapters/llm.py +"""GovernedLLMClient — 实现 LLMProvider Protocol 的四层治理栈。 + +治理层序：熔断检查 → 缓存查询 → 流式调用（三层看门狗）→ 重试退避。 +所有 LLM 调用必须经过此入口，禁止裸调 OpenAI SDK。 +""" +from __future__ import annotations + +import asyncio +import json +import time +import uuid +from collections.abc import AsyncIterator +from typing import Any + +import httpx +from loguru import logger + +from adapters.breaker import CircuitBreaker +from adapters.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts +from core.protocols import TelemetryRecorder +from core.types import LLMResponse + + +class CircuitOpenError(Exception): + """熔断器已开启，拒绝调用。""" + + +class _SseAnomaly(Exception): + """SSE 协议异常。""" + + def __init__(self, kind: str) -> None: + self.kind = kind + super().__init__(f"SSE 协议异常: {kind}") + + +class GovernedLLMClient: + """四层治理栈 LLM 客户端。 + + 实现 core/protocols.py 的 LLMProvider Protocol。 + 内部流式消费 SSE，对外返回完整 LLMResponse。 + + Args: + model: 模型名。 + base_url: OpenAI 兼容 API 基地址。 + api_key: API 密钥。 + provider: provider 标识（deepseek/qwen/unknown）。 + thinking: 是否启用 thinking 模式。 + breaker: 熔断器实例。 + cache: Redis 缓存实例（None = 不缓存）。 + telemetry: 遥测记录器。 + timeout_s: 总超时秒数。 + ttft_timeout_s: 首 token 超时秒数（None = 不启用）。 + inter_token_timeout_s: token 间隔超时秒数（None = 不启用）。 + max_retries: 最大重试次数。 + retry_base_delay_s: 重试基础延迟秒数。 + retry_max_delay_s: 重试最大延迟秒数。 + """ + + def __init__( + self, + *, + model: str, + base_url: str, + api_key: str, + provider: str, + thinking: bool, + breaker: CircuitBreaker, + cache: Any | None, + telemetry: TelemetryRecorder, + timeout_s: float, + ttft_timeout_s: float | None, + inter_token_timeout_s: float | None, + max_retries: int, + retry_base_delay_s: float, + retry_max_delay_s: float, + ) -> None: + self._model = model + self._base_url = base_url.rstrip("/") + self._api_key = api_key + self._provider = provider + self._thinking = thinking + self._breaker = breaker + self._cache = cache + self._telemetry = telemetry + self._timeout_s = timeout_s + self._ttft_timeout_s = ttft_timeout_s or timeout_s + self._inter_token_timeout_s = inter_token_timeout_s or timeout_s + self._max_retries = max_retries + self._retry_base_delay_s = retry_base_delay_s + self._retry_max_delay_s = retry_max_delay_s + self._http = httpx.AsyncClient( + base_url=self._base_url, + headers={"Authorization": f"Bearer {api_key}"}, + timeout=httpx.Timeout( + connect=10.0, read=timeout_s, write=30.0, pool=10.0 + ), + ) + + async def chat( + self, + messages: list[dict[str, Any]], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + """执行 LLM 调用，经过四层治理栈。""" + now = time.monotonic() + + # 层1: 熔断检查 + if self._breaker.is_open(now): + raise CircuitOpenError( + f"熔断器已开启，模型 {self._model} 暂不可用" + ) + + call_id = str(uuid.uuid4()) + messages_json = json.dumps(messages, ensure_ascii=False) + + # 层4: 缓存查询 + if self._cache is not None: + cached = await self._cache.get(self._model, messages) + if cached is not None: + response = LLMResponse( + content=cached.content, + thinking=cached.thinking, + model=cached.model, + provider=cached.provider, + prompt_tokens=cached.prompt_tokens, + completion_tokens=cached.completion_tokens, + latency_ms=0, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=True, + call_id=call_id, + ) + await self._telemetry.record_llm_call( + call_id=call_id, + parent_call_id=parent_call_id, + session_id=session_id, + model_name=self._model, + provider=self._provider, + messages=messages_json, + response=response.content, + thinking=response.thinking, + prompt_tokens=response.prompt_tokens, + completion_tokens=response.completion_tokens, + latency_ms=0, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=True, + error=None, + ) + return response + + # 层2+3: 重试循环 + 流式调用 + last_error: Exception | None = None + for attempt in range(self._max_retries + 1): + started = time.monotonic() + try: + content, thinking, ttft_ms, max_itoken_ms, usage = ( + await self._call_streaming(messages) + ) + latency_ms = int((time.monotonic() - started) * 1000) + self._breaker.record_success() + + response = LLMResponse( + content=content, + thinking=thinking, + model=self._model, + provider=self._provider, + prompt_tokens=usage.get("prompt_tokens", 0), + completion_tokens=usage.get("completion_tokens", 0), + latency_ms=latency_ms, + ttft_ms=ttft_ms, + max_inter_token_ms=max_itoken_ms, + cache_hit=False, + call_id=call_id, + ) + + if self._cache is not None: + await self._cache.set(self._model, messages, response) + + await self._telemetry.record_llm_call( + call_id=call_id, + parent_call_id=parent_call_id, + session_id=session_id, + model_name=self._model, + provider=self._provider, + messages=messages_json, + response=content, + thinking=thinking, + prompt_tokens=response.prompt_tokens, + completion_tokens=response.completion_tokens, + latency_ms=latency_ms, + ttft_ms=ttft_ms, + max_inter_token_ms=max_itoken_ms, + cache_hit=False, + error=None, + ) + return response + + except (httpx.HTTPStatusError,) as exc: + status = exc.response.status_code + if status in (401, 403): + self._breaker.force_open(time.monotonic()) + await self._record_error( + call_id, parent_call_id, session_id, + messages_json, started, exc, + ) + raise + last_error = exc + self._breaker.record_failure(time.monotonic()) + + except ( + httpx.TimeoutException, + httpx.ConnectError, + StreamLivenessTimeout, + _SseAnomaly, + ) as exc: + last_error = exc + self._breaker.record_failure(time.monotonic()) + + if attempt < self._max_retries: + delay = min( + self._retry_base_delay_s * (2 ** attempt), + self._retry_max_delay_s, + ) + logger.warning( + "LLM 调用失败 (attempt {}/{}), {}s 后重试: {}", + attempt + 1, + self._max_retries + 1, + delay, + last_error, + ) + await asyncio.sleep(delay) + + await self._record_error( + call_id, parent_call_id, session_id, + messages_json, started, last_error, + ) + raise last_error # type: ignore[misc] + + async def _call_streaming( + self, messages: list[dict[str, Any]] + ) -> tuple[str, str, float | None, float | None, dict[str, int]]: + """执行流式 SSE 调用。返回 (content, thinking, ttft_ms, max_itoken_ms, usage)。""" + body = self._build_request_body(messages) + lines = await self._stream_request(body) + return await self._consume_stream(lines) + + def _build_request_body(self, messages: list[dict[str, Any]]) -> dict[str, Any]: + """构建 OpenAI 兼容请求体。""" + body: dict[str, Any] = { + "model": self._model, + "messages": messages, + "stream": True, + "stream_options": {"include_usage": True}, + } + body.update(self._build_thinking_body()) + return body + + def _build_thinking_body(self) -> dict[str, Any]: + """构建 provider 特定的 thinking 参数。""" + if self._provider == "deepseek": + return {"thinking": {"type": "enabled" if self._thinking else "disabled"}} + if self._provider == "qwen": + return {"enable_thinking": self._thinking} + return {} + + async def _stream_request( + self, body: dict[str, Any] + ) -> AsyncIterator[str]: + """发起流式 POST 请求，返回 SSE 行迭代器。""" + resp = await self._http.send( + self._http.build_request( + "POST", "/chat/completions", json=body + ), + stream=True, + ) + resp.raise_for_status() + return resp.aiter_lines() + + async def _consume_stream( + self, lines: AsyncIterator[str] + ) -> tuple[str, str, float | None, float | None, dict[str, int]]: + """消费 SSE 流，提取内容和指标。""" + usage_sink: dict[str, Any] = {} + sse_deltas = _iter_sse_deltas(lines, usage_sink) + guarded = stream_with_liveness_timeouts( + sse_deltas, + ttft_s=self._ttft_timeout_s, + inter_token_s=self._inter_token_timeout_s, + total_s=self._timeout_s, + ) + + content_parts: list[str] = [] + thinking_parts: list[str] = [] + ttft_ms: float | None = None + max_inter_token_ms: float | None = None + started = time.monotonic() + last_token_time = started + + async for is_content, text in guarded: + now = time.monotonic() + if ttft_ms is None: + ttft_ms = (now - started) * 1000 + else: + gap_ms = (now - last_token_time) * 1000 + if max_inter_token_ms is None or gap_ms > max_inter_token_ms: + max_inter_token_ms = gap_ms + last_token_time = now + + if is_content: + content_parts.append(text) + else: + thinking_parts.append(text) + + content = "".join(content_parts) + thinking = "".join(thinking_parts) + + if self._provider == "qwen" and not thinking_parts: + content, thinking = self._strip_qwen_thinking(content) + + usage = usage_sink.get("usage", {}) + prompt_tokens = usage.get("prompt_tokens", 0) if isinstance(usage, dict) else 0 + completion_tokens = ( + usage.get("completion_tokens", 0) if isinstance(usage, dict) else 0 + ) + + return ( + content, + thinking, + ttft_ms, + max_inter_token_ms, + {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens}, + ) + + @staticmethod + def _strip_qwen_thinking(content: str) -> tuple[str, str]: + """剥离 Qwen 模型的 ... 标签。""" + import re + + pattern = r"(.*?)" + match = re.search(pattern, content, re.DOTALL) + if match: + thinking = match.group(1).strip() + cleaned = re.sub(pattern, "", content, flags=re.DOTALL).strip() + return cleaned, thinking + return content, "" + + async def _record_error( + self, + call_id: str, + parent_call_id: str | None, + session_id: str | None, + messages_json: str, + started: float, + error: Exception | None, + ) -> None: + """记录失败调用的遥测。""" + latency_ms = int((time.monotonic() - started) * 1000) + await self._telemetry.record_llm_call( + call_id=call_id, + parent_call_id=parent_call_id, + session_id=session_id, + model_name=self._model, + provider=self._provider, + messages=messages_json, + response="", + thinking="", + prompt_tokens=0, + completion_tokens=0, + latency_ms=latency_ms, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + error=str(error) if error else None, + ) + + async def aclose(self) -> None: + """关闭 httpx 客户端。""" + await self._http.aclose() + + +def _sse_data_payload(raw: str) -> str | None: + """提取 SSE data: 行的 payload。""" + line = raw.strip() + if not line or line.startswith(":") or not line.startswith("data:"): + return None + return line[len("data:"):].strip() + + +def _sse_delta( + chunk: dict[str, Any], usage_sink: dict[str, Any] +) -> tuple[bool, str] | None: + """从 SSE JSON 帧提取 delta token。 + + 返回 (True, text) 表示 content，(False, text) 表示 reasoning。 + 无 delta 返回 None。副作用：写入 usage_sink。 + """ + if chunk.get("usage"): + usage_sink["usage"] = chunk["usage"] + choices = chunk.get("choices") or [] + if not choices: + return None + delta = choices[0].get("delta") or {} + content = delta.get("content") + if content: + return (True, content) + reasoning = delta.get("reasoning_content") + if reasoning: + return (False, reasoning) + return None + + +async def _iter_sse_deltas( + lines: AsyncIterator[str], usage_sink: dict[str, Any] +) -> AsyncIterator[tuple[bool, str]]: + """解析 SSE 行流为 (is_content, text) 元组流。""" + async for raw in lines: + data = _sse_data_payload(raw) + if data is None: + continue + if data == "[DONE]": + usage_sink["done"] = True + return + try: + chunk = json.loads(data) + except json.JSONDecodeError as exc: + raise _SseAnomaly("malformed_json") from exc + delta = _sse_delta(chunk, usage_sink) + if delta is not None: + yield delta +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_governed_llm.py -v` +Expected: 7 tests PASS + +- [ ] **Step 5: 提交** + +```bash +git add adapters/llm.py tests/unit/test_governed_llm.py +git commit -m "feat(adapters): 实现 GovernedLLMClient 四层治理栈 + +流式 SSE + 三层看门狗 + 重试退避 + 熔断 + Redis 缓存 + 遥测。 +provider 差异处理（DeepSeek reasoning_content vs Qwen think 标签）。" +``` + +--- + +### Task 12: 集成测试 — core/agent/ + adapters/ 端到端 + +**Files:** +- Create: `tests/integration/test_agent_governed_e2e.py` + +- [ ] **Step 1: 写集成测试** + +```python +# tests/integration/test_agent_governed_e2e.py +"""AgentLoop + GovernedLLMClient 集成测试。 + +验证 core/agent/ 通过 LLMProvider Protocol 与 adapters/llm.py 协作。 +使用 mock SSE 流，不依赖真实 LLM API。 +""" +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import pytest + +from adapters.breaker import CircuitBreaker +from adapters.llm import GovernedLLMClient +from adapters.telemetry import SQLiteTelemetryRecorder +from core.agent.loop import AgentLoop +from core.agent.protocols import hookimpl +from core.protocols import LLMProvider + + +class _StubDispatcher: + async def dispatch( + self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any] + ) -> str: + if tool_name == "submit_answer": + return "答案已提交" + if tool_name == "search_tree": + return "搜索结果: L2-3 节点" + raise ValueError(f"未知工具: {tool_name}") + + +def _sse_for_action(action: dict, reasoning: str = "") -> list[str]: + """构造一个完整 SSE 响应。""" + content = json.dumps( + {"reflect": {}, "plan": {}, "action": action}, ensure_ascii=False + ) + lines = [] + if reasoning: + lines.append( + f'data: {json.dumps({"choices": [{"delta": {"reasoning_content": reasoning}}], "model": "test"})}' + ) + lines.append( + f'data: {json.dumps({"choices": [{"delta": {"content": content}}], "model": "test"})}' + ) + lines.append( + f'data: {json.dumps({"choices": [], "model": "test", "usage": {"prompt_tokens": 10, "completion_tokens": 5}})}' + ) + lines.append("data: [DONE]") + return lines + + +@pytest.mark.asyncio +async def test_agent_loop_with_governed_client(tmp_path: Path) -> None: + """AgentLoop 通过 GovernedLLMClient 完成搜索+提交。""" + telemetry = SQLiteTelemetryRecorder(db_path=tmp_path / "telemetry.db") + client = GovernedLLMClient( + model="test-model", + base_url="https://api.test.com/v1", + api_key="sk-test", + provider="deepseek", + thinking=True, + breaker=CircuitBreaker(fail_threshold=5, cooldown_s=60.0), + cache=None, + telemetry=telemetry, + timeout_s=30.0, + ttft_timeout_s=10.0, + inter_token_timeout_s=5.0, + max_retries=0, + retry_base_delay_s=0.01, + retry_max_delay_s=0.05, + ) + + call_count = 0 + responses = [ + _sse_for_action( + {"tool": "search_tree", "args": {"query": "什么是人工智能"}}, + reasoning="让我思考一下", + ), + _sse_for_action( + {"tool": "submit_answer", "args": {"answer": "人工智能是..."}} + ), + ] + + original_stream = client._stream_request + + async def mock_stream(body): + nonlocal call_count + lines = responses[call_count] + call_count += 1 + + async def gen(): + for line in lines: + yield line + + return gen() + + assert isinstance(client, LLMProvider) + + loop = AgentLoop(llm=client, max_steps=10) + + with patch.object(client, "_stream_request", side_effect=mock_stream): + result = await loop.run( + system_prompt="你是一个搜索助手", + user_prompt="什么是人工智能？", + tool_dispatcher=_StubDispatcher(), + session_id="test-session", + ) + + assert result.stop_reason == "finished" + assert result.result == {"answer": "人工智能是..."} + assert result.steps_used == 2 + assert result.steps[0].thought == "让我思考一下" + assert result.steps[0].tool_call["tool"] == "search_tree" + assert result.steps[1].tool_call["tool"] == "submit_answer" + assert result.token_usage["prompt_tokens"] == 20 +``` + +- [ ] **Step 2: 运行集成测试** + +Run: `conda activate Video-Tree-TRM & pytest tests/integration/test_agent_governed_e2e.py -v` +Expected: PASS + +- [ ] **Step 3: 提交** + +```bash +git add tests/integration/test_agent_governed_e2e.py +git commit -m "test: AgentLoop + GovernedLLMClient 集成测试 + +验证 core/agent/ 通过 LLMProvider Protocol 与 adapters/ 端到端协作。" +``` + +--- + +### Task 13: 全量测试 + lint + +**Files:** 无新文件 + +- [ ] **Step 1: 运行全量测试** + +Run: `conda activate Video-Tree-TRM & pytest tests/ -v --tb=short` +Expected: 全部 PASS + +- [ ] **Step 2: 运行 lint** + +Run: `conda activate Video-Tree-TRM & ruff check app/ core/ adapters/ --fix && ruff format app/ core/ adapters/` +Expected: 无错误 + +- [ ] **Step 3: 修复任何问题后提交** + +```bash +git add -A +git commit -m "chore: lint 修复" +``` + +--- + +### Task 14: 文件整理 — 规范同步验证 + +**Files:** +- Verify: `research-wiki/ARCHITECTURE.md` +- Verify: `CLAUDE.md` +- Verify: `.env.example` + +最终核查实际实现与文档的一致性。 + +- [ ] **Step 1: 核查 ARCHITECTURE.md §2.3 目录结构** + +确认文档中列出的每个文件路径都存在： +```bash +ls core/protocols.py core/types.py core/agent/loop.py core/agent/types.py core/agent/protocols.py +ls adapters/llm.py adapters/streaming.py adapters/breaker.py adapters/redis_cache.py adapters/telemetry.py +``` + +- [ ] **Step 2: 核查 ARCHITECTURE.md §3.1 接缝清单** + +确认 Protocol 分类与实际代码一致： +- `core/protocols.py` 应包含 `LLMProvider`, `VLMProvider`, `TelemetryRecorder` +- `core/agent/protocols.py` 应包含 `ToolDispatcher`, `AgentLoopSpec` + +```bash +grep "class.*Protocol\|class.*Spec" core/protocols.py core/agent/protocols.py +``` + +- [ ] **Step 3: 核查 ARCHITECTURE.md §4 遥测字段** + +确认 `TelemetryRecorder.record_llm_call()` 签名包含所有文档列出的字段： + +```bash +grep -A 20 "async def record_llm_call" core/protocols.py +``` + +- [ ] **Step 4: 核查 ARCHITECTURE.md §5 治理栈** + +确认文档描述的四层（看门狗→重试→熔断→缓存）与 `GovernedLLMClient.chat()` 实现一致。 + +- [ ] **Step 5: 核查 CLAUDE.md §4.8/§4.9/§5** + +确认 CLAUDE.md 的遥测字段、韧性层数、项目结构树与 ARCHITECTURE.md 和实际代码一致。 + +- [ ] **Step 6: 核查 .env.example** + +确认新增的 `LLM_TTFT_TIMEOUT`、`LLM_INTER_TOKEN_TIMEOUT`、`REDIS_CACHE_TTL` 存在。 + +- [ ] **Step 7: 修复差异并提交** + +如有任何不一致，修复后提交： + +```bash +git add -A +git commit -m "docs: 最终文件整理 — 实现与规范同步验证" +``` + +--- + +## 核心算法保真校验 + +本计划涉及算法 #11（Agent Loop）的迁移。 + +**校验结果**：Task 6 的实现保真 TRM4 `core/loop.py` 的以下逻辑： +- `_parse_response`：`repair_json` → `json.loads` → 校验 `action`/`tool`/`args`（TRM4 lines 259-293） +- `_execute_tool`：`ValueError` 捕获 → `(output, False)`（TRM4 lines 295-315） +- `_build_feedback`：`[工具执行结果: {name}]` 格式（TRM4 lines 317-337） +- `run`：messages 格式、retry prompt 文案、`step_count` 只计有效调用、四个终止路径 + +**变更项**（有意为之，非简化）： +- 同步 → 异步：所有方法 async 化 +- `client: Any` → `llm: LLMProvider`：Protocol 类型化 +- `tool_fn: Callable` → `ToolDispatcher.dispatch()`：Protocol + context 参数 +- `Step` 新增 `call_id` 字段 +- thinking 从 `getattr(msg, "reasoning_content")` 改为 `response.thinking`（adapters 已统一剥离） + +本计划不涉及其余 12 项核心算法，保真校验对它们不适用。 diff --git a/research-wiki/plans/core-agent-adapters-llm.md b/research-wiki/plans/core-agent-adapters-llm.md new file mode 100644 index 0000000..8e431bf --- /dev/null +++ b/research-wiki/plans/core-agent-adapters-llm.md @@ -0,0 +1,9 @@ +--- +type: plan +node_id: plan:core-agent-adapters-llm +title: "core/agent/ + adapters/llm 基础设施实现计划" +date: 2026-07-07 +--- + +# core/agent/ + adapters/llm 基础设施实现计划 +