# 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 项核心算法，保真校验对它们不适用。