From ee90fb8e88c37cb8c04c404d37862a9e5d93019f Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 6 Jul 2026 23:08:52 -0400 Subject: [PATCH] =?UTF-8?q?fix(adapters):=20=E4=BF=AE=E5=A4=8D=20GovernedL?= =?UTF-8?q?LMClient=20=E6=B5=81=E5=BC=8F=E6=B6=88=E8=B4=B9=E5=92=8C?= =?UTF-8?q?=E8=A7=84=E6=A0=BC=E5=81=8F=E5=B7=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _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) --- adapters/llm.py | 88 +++++++++++++---------- tests/unit/test_governed_llm.py | 119 ++++++++++++-------------------- 2 files changed, 96 insertions(+), 111 deletions(-) diff --git a/adapters/llm.py b/adapters/llm.py index f61e2be..5bb2793 100644 --- a/adapters/llm.py +++ b/adapters/llm.py @@ -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, diff --git a/tests/unit/test_governed_llm.py b/tests/unit/test_governed_llm.py index 2970654..0f3146a 100644 --- a/tests/unit/test_governed_llm.py +++ b/tests/unit/test_governed_llm.py @@ -23,9 +23,7 @@ class _FakeRedisCache: def __init__(self) -> None: self._store: dict[str, LLMResponse] = {} - async def get( - self, model: str, messages: list[dict[str, str]] - ) -> LLMResponse | None: + async def get(self, model: str, messages: list[dict[str, str]]) -> LLMResponse | None: key = f"{model}:{json.dumps(messages, sort_keys=True)}" return self._store.get(key) @@ -49,52 +47,6 @@ class _FakeTelemetry: self.calls.append(kwargs) -# ── 工具函数 ───────────────────────────────────────────────────────── - - -async def _async_line_iter(lines: list[str]): - """将字符串列表转为异步行迭代器。""" - for line in lines: - yield line - - -def _make_sse_lines( - content_chunks: list[str], - *, - reasoning_chunks: list[str] | None = None, - model: str = "test-model", - usage: dict[str, int] | None = None, -) -> list[str]: - """构造 SSE 文本行序列，模拟 OpenAI 兼容流式响应。""" - lines: list[str] = [] - - # 先产出 reasoning 帧 - for chunk in reasoning_chunks or []: - frame = { - "choices": [{"delta": {"reasoning_content": chunk}}], - "model": model, - } - lines.append(f"data: {json.dumps(frame)}\n") - - # 再产出 content 帧 - for chunk in content_chunks: - frame = { - "choices": [{"delta": {"content": chunk}}], - "model": model, - } - lines.append(f"data: {json.dumps(frame)}\n") - - # usage 帧 - if usage is None: - usage = {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} - frame_usage = {"choices": [], "usage": usage, "model": model} - lines.append(f"data: {json.dumps(frame_usage)}\n") - - # 终止标记 - lines.append("data: [DONE]\n") - return lines - - def _build_client( *, breaker: CircuitBreaker | None = None, @@ -180,16 +132,18 @@ async def test_successful_streaming_call(): client = _build_client(telemetry=telemetry) messages = [{"role": "user", "content": "hello"}] - sse_lines = _make_sse_lines( - ["Hello", " world"], - reasoning_chunks=["Let me", " think"], - usage={"prompt_tokens": 20, "completion_tokens": 10, "total_tokens": 30}, - ) + async def fake_call_streaming( + _messages: Any, + ) -> tuple[str, str, float | None, float | None, dict]: + return ( + "Hello world", + "Let me think", + 1.5, # ttft_ms + 0.8, # max_inter_token_ms + {"prompt_tokens": 20, "completion_tokens": 10, "total_tokens": 30}, + ) - async def fake_stream_request(_messages: Any) -> list[str]: - return sse_lines - - with patch.object(client, "_stream_request", side_effect=fake_stream_request): + with patch.object(client, "_call_streaming", side_effect=fake_call_streaming): result = await client.chat(messages) assert result.content == "Hello world" @@ -213,17 +167,24 @@ async def test_transient_error_retries_and_records_telemetry(): client = _build_client(telemetry=telemetry) messages = [{"role": "user", "content": "hello"}] - sse_lines = _make_sse_lines(["OK"]) call_count = 0 - async def flaky_stream(_messages: Any) -> list[str]: + async def flaky_call_streaming( + _messages: Any, + ) -> tuple[str, str, float | None, float | None, dict]: nonlocal call_count call_count += 1 if call_count <= 2: raise httpx.ConnectError("connection refused") - return sse_lines + return ( + "OK", + "", + 1.0, + None, + {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + ) - with patch.object(client, "_stream_request", side_effect=flaky_stream): + with patch.object(client, "_call_streaming", side_effect=flaky_call_streaming): result = await client.chat(messages) assert result.content == "OK" @@ -236,6 +197,10 @@ async def test_transient_error_retries_and_records_telemetry(): success_calls = [c for c in telemetry.calls if c.get("error") is None] assert len(success_calls) == 1 + # 所有遥测记录应使用同一个 call_id（Important 2 修复验证） + call_ids = {c["call_id"] for c in telemetry.calls} + assert len(call_ids) == 1 + @pytest.mark.asyncio async def test_fatal_error_force_opens_breaker(): @@ -250,13 +215,17 @@ async def test_fatal_error_force_opens_breaker(): request=httpx.Request("POST", "http://localhost:8000/v1/chat/completions"), ) - async def auth_fail(_messages: Any) -> list[str]: + async def auth_fail( + _messages: Any, + ) -> tuple[str, str, float | None, float | None, dict]: raise httpx.HTTPStatusError( "Unauthorized", request=mock_response.request, response=mock_response ) - with patch.object(client, "_stream_request", side_effect=auth_fail), \ - pytest.raises(httpx.HTTPStatusError): + with ( + patch.object(client, "_call_streaming", side_effect=auth_fail), + pytest.raises(httpx.HTTPStatusError), + ): await client.chat(messages) # 熔断器已被 force_open @@ -291,16 +260,20 @@ async def test_parent_call_id_forwarded_to_telemetry(): client = _build_client(telemetry=telemetry) messages = [{"role": "user", "content": "hello"}] - sse_lines = _make_sse_lines(["result"]) - - async def fake_stream(_messages: Any) -> list[str]: - return sse_lines - - with patch.object(client, "_stream_request", side_effect=fake_stream): - await client.chat( - messages, session_id="sess-42", parent_call_id="parent-99" + async def fake_call_streaming( + _messages: Any, + ) -> tuple[str, str, float | None, float | None, dict]: + return ( + "result", + "", + 1.0, + None, + {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, ) + with patch.object(client, "_call_streaming", side_effect=fake_call_streaming): + await client.chat(messages, session_id="sess-42", parent_call_id="parent-99") + assert len(telemetry.calls) == 1 assert telemetry.calls[0]["session_id"] == "sess-42" assert telemetry.calls[0]["parent_call_id"] == "parent-99"