iomgaa
bc78138d8f
Self-Review + Codex 审查通过。修复：依赖前置、治理栈顺序统一、 GovernedLLMClient 补充异常/provider/parent_call_id 测试、ARQ 残留清理。 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
96 KiB
96 KiB
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 四层治理栈（熔断→缓存→流式看门狗→重试），含流式三层看门狗和遥测集成。
前置条件（执行前完成）：
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：
├── 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
├── 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 韧性参数 ── 段落中新增：
LLM_TTFT_TIMEOUT=30
LLM_INTER_TOKEN_TIMEOUT=15
LLM_RETRY_MAX_DELAY=30.0
REDIS_CACHE_TTL=86400
- Step 8: 提交
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: 写失败测试
# 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
# 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: 提交
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: 写失败测试
# 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
# 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: 提交
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: 写失败测试
# 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
# 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: 提交
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: 写失败测试
# 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
# 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: 提交
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 终止
# 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计数逻辑、四个终止路径
# 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: 提交
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: 写失败测试
# 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
# 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: 提交
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: 写失败测试
# 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
# 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: 提交
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: 写失败测试
# 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
# 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: 提交
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: 写失败测试
# 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
# 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: 提交
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: 写失败测试
# 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 的 <think> 标签应被剥离。"""
telemetry = _FakeTelemetry()
client = _build_client(telemetry=telemetry)
client._provider = "qwen"
content_with_think = "<think>我在想</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
# 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 模型的 <think>...</think> 标签。"""
import re
pattern = r"<think>(.*?)</think>"
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: 提交
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: 写集成测试
# 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: 提交
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: 修复任何问题后提交
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 目录结构
确认文档中列出的每个文件路径都存在：
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,TelemetryRecordercore/agent/protocols.py应包含ToolDispatcher,AgentLoopSpec
grep "class.*Protocol\|class.*Spec" core/protocols.py core/agent/protocols.py
- Step 3: 核查 ARCHITECTURE.md §4 遥测字段
确认 TelemetryRecorder.record_llm_call() 签名包含所有文档列出的字段：
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: 修复差异并提交
如有任何不一致，修复后提交：
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 项核心算法，保真校验对它们不适用。