iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(adapters): 修复 GovernedLLMClient 流式消费和规格偏差

Browse Source 
- _stream_request + _consume_stream 合并为 _call_streaming，
  在 async with self._http.stream() 内直接消费流，
  确保看门狗作用于真实 HTTP 流而非内存列表（Critical 1）
- breaker 检查移到 call_id 生成之前（Critical 2）
- cache/ttft_timeout_s/inter_token_timeout_s 支持 None（Important 1）
- 重试失败路径遥测使用统一 call_id（Important 2）

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-06 23:08:52 -04:00
parent cfd3277f80
commit ee90fb8e88
2 changed files with 96 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
adapters/llm.py
+50 -38
View File
@@ -234,11 +234,11 @@ class GovernedLLMClient:
provider: str,
thinking: bool,
breaker: CircuitBreaker,
cache: Any,
cache: Any | None,
telemetry: Any,
timeout_s: float,
ttft_timeout_s: float,
inter_token_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,
@@ -252,8 +252,10 @@ class GovernedLLMClient:
self._cache = cache
self._telemetry = telemetry
self._timeout_s = timeout_s
self._ttft_timeout_s = ttft_timeout_s
self._inter_token_timeout_s = inter_token_timeout_s
self._ttft_timeout_s = ttft_timeout_s if ttft_timeout_s is not None else timeout_s
self._inter_token_timeout_s = (
inter_token_timeout_s if inter_token_timeout_s is not None else 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
@@ -288,14 +290,16 @@ class GovernedLLMClient:
httpx.HTTPStatusError: 不可恢复的 HTTP 错误（401/403）。
"""
started = time.monotonic()
call_id = str(uuid4())
# ① 熔断器检查
# ① 熔断器检查（规格要求在 call_id 生成之前）
if self._breaker.is_open(self._provider, time.monotonic()):
raise CircuitOpenError(f"熔断器已开启，拒绝调用 provider={self._provider}")
# ② 缓存查询
cached = await self._cache.get(self._model, messages)
# ② call_id 生成
call_id = str(uuid4())
# ③ 缓存查询（cache 为 None 时跳过）
cached = await self._cache.get(self._model, messages) if self._cache is not None else None
if cached is not None:
response = LLMResponse(
content=cached.content,
@@ -329,14 +333,13 @@ class GovernedLLMClient:
)
return response
# ③ 重试循环 + 流式消费
# ④ 重试循环 + 流式消费
last_exc: Exception | None = None
for attempt in range(self._max_retries):
attempt_start = time.monotonic()
try:
sse_lines = await self._stream_request(messages)
content, thinking_text, ttft_ms, max_itoken_ms, usage = await self._consume_stream(
sse_lines
content, thinking_text, ttft_ms, max_itoken_ms, usage = await self._call_streaming(
messages
)
# 熔断器记成功
@@ -366,8 +369,9 @@ class GovernedLLMClient:
call_id=call_id,
)
# ④ 写缓存
await self._cache.set(self._model, messages, response)
# ④ 写缓存（cache 为 None 时跳过）
if self._cache is not None:
await self._cache.set(self._model, messages, response)
# ⑤ 遥测
await self._telemetry.record_llm_call(
@@ -420,7 +424,7 @@ class GovernedLLMClient:
if _is_transient_error(exc):
self._breaker.record_failure(self._provider, time.monotonic())
await self._telemetry.record_llm_call(
call_id=str(uuid4()), # 每次重试用新 call_id
call_id=call_id,
parent_call_id=parent_call_id,
session_id=session_id,
model_name=self._model,
@@ -478,20 +482,14 @@ class GovernedLLMClient:
assert last_exc is not None
raise last_exc
async def _stream_request(self, messages: list[dict[str, Any]]) -> list[str]:
"""发起流式 HTTP 请求并收集全部 SSE 行。
此方法为主要的 HTTP 交互点，测试通过 patch 替换以注入假 SSE 行。
def _build_request_body(self, messages: list[dict[str, Any]]) -> dict[str, Any]:
"""构造流式 chat 请求体。
参数:
messages: OpenAI 格式消息列表。
返回:
SSE 文本行列表。
异常:
httpx.HTTPStatusError: HTTP 状态码错误。
httpx.ConnectError: 连接错误。
请求体字典。
"""
payload: dict[str, Any] = {
"model": self._model,
@@ -501,34 +499,48 @@ class GovernedLLMClient:
}
if self._thinking:
payload.update(_build_thinking_body(self._provider))
return payload
collected_lines: list[str] = []
async def _call_streaming(
self, messages: list[dict[str, Any]]
) -> tuple[str, str, float | None, float | None, dict]:
"""发起流式 HTTP 请求并在 context manager 内直接消费 SSE 流。
将 HTTP 请求与流消费合并，确保三层活性看门狗作用于真实 HTTP 流而非内存列表。
测试通过 patch 此方法注入假结果。
参数:
messages: OpenAI 格式消息列表。
返回:
(content, thinking, ttft_ms, max_inter_token_ms, usage_dict)。
异常:
httpx.HTTPStatusError: HTTP 状态码错误。
httpx.ConnectError: 连接错误。
StreamLivenessTimeout: 活性超时。
"""
payload = self._build_request_body(messages)
async with self._http.stream("POST", "/chat/completions", json=payload) as resp:
if resp.status_code >= 400:
await resp.aread()
resp.raise_for_status()
async for line in resp.aiter_lines():
collected_lines.append(line)
return collected_lines
# 直接在 context manager 内消费流，看门狗作用于真实 HTTP 流
return await self._consume_stream(resp.aiter_lines())
async def _consume_stream(
self, sse_lines: list[str]
self, lines: AsyncIterator[str]
) -> tuple[str, str, float | None, float | None, dict]:
"""消费 SSE 行序列，累积 content/thinking，测量 TTFT 和 max_inter_token_ms。
"""消费 SSE 行异步迭代器，累积 content/thinking，测量 TTFT 和 max_inter_token_ms。
参数:
sse_lines: SSE 文本行列表。
lines: SSE 文本行异步迭代器（直接来自 httpx resp.aiter_lines()）。
返回:
(content, thinking, ttft_ms, max_inter_token_ms, usage_dict)。
"""
usage_sink: dict = {}
async def _lines_iter() -> AsyncIterator[str]:
for line in sse_lines:
yield line
raw_deltas = _iter_sse_deltas(_lines_iter(), usage_sink)
raw_deltas = _iter_sse_deltas(lines, usage_sink)
guarded = stream_with_liveness_timeouts(
raw_deltas,
ttft_s=self._ttft_timeout_s,