feat(adapters): 实现 GovernedLLMClient 四层治理栈
流式 SSE + 三层看门狗 + 重试退避 + 熔断 + Redis 缓存 + 遥测。 provider 差异处理（DeepSeek reasoning_content vs Qwen think 标签）。 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
+574
@@ -0,0 +1,574 @@
|
||||
"""GovernedLLMClient — 四层治理栈统一 LLM 调用网关。
|
||||
|
||||
组合熔断器、Redis 缓存、指数退避重试、流式 SSE 消费 + 三层活性看门狗，
|
||||
并在每次调用后自动写入遥测。provider 差异处理覆盖 DeepSeek（reasoning_content）
|
||||
和 Qwen（<think> 标签剥离）。
|
||||
|
||||
治理栈五层（ARCHITECTURE.md §5）:
|
||||
1. 硬超时 — asyncio.wait_for + StreamLivenessTimeout
|
||||
2. 指数退避重试 — max_retries + base_delay + max_delay
|
||||
3. 熔断器 — CircuitBreaker（连续 N 失败开路，冷却后半开探针）
|
||||
4. Redis 响应缓存 — content-addressed hash(model + messages)
|
||||
5. 遥测 — TelemetryRecorder（每次调用必录）
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from uuid import uuid4
|
||||
|
||||
import httpx
|
||||
from loguru import logger
|
||||
|
||||
from adapters.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts
|
||||
from core.types import LLMResponse
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
from adapters.breaker import CircuitBreaker
|
||||
|
||||
|
||||
class CircuitOpenError(Exception):
|
||||
"""熔断器已开启，拒绝调用。"""
|
||||
|
||||
|
||||
class _SseAnomaly(Exception): # noqa: N818 — 参考 CHSAnalyzer2 命名
|
||||
"""SSE 协议异常（malformed_json 等）。"""
|
||||
|
||||
def __init__(self, kind: str) -> None:
|
||||
self.kind = kind
|
||||
super().__init__(f"SSE 协议异常: {kind}")
|
||||
|
||||
|
||||
# ── SSE 解析（模块级纯函数） ────────────────────────────────────────
|
||||
|
||||
|
||||
def _sse_data_payload(raw: str) -> str | None:
|
||||
"""提取一行 SSE 的 data 段；ping/空行/非 data 行返回 None。
|
||||
|
||||
参数:
|
||||
raw: 原始 SSE 行文本。
|
||||
|
||||
返回:
|
||||
data 段 payload 字符串，或 None（应被跳过的行）。
|
||||
"""
|
||||
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, usage_sink: dict) -> tuple[bool, str] | None:
|
||||
"""旁路收集 usage 并提取一帧增量 token。
|
||||
|
||||
返回 (True, content) 或 (False, reasoning_content)，无增量则 None。
|
||||
优先 content（正式输出）；否则取 reasoning_content（thinking 模型的思考 token）。
|
||||
|
||||
参数:
|
||||
chunk: 一帧解析后的 SSE JSON。
|
||||
usage_sink: 旁路字典；含 usage 的帧写入 usage_sink["usage"]。
|
||||
|
||||
返回:
|
||||
(is_content, text) 或 None。
|
||||
"""
|
||||
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
|
||||
) -> AsyncIterator[tuple[bool, str]]:
|
||||
"""OpenAI 兼容 SSE 行流 → 逐帧产出 (是否 content, 文本)。
|
||||
|
||||
usage 旁路写入 usage_sink。内部消化 ping/空行/[DONE]/usage-only 帧。
|
||||
|
||||
参数:
|
||||
lines: 底层 SSE 文本行异步迭代器。
|
||||
usage_sink: 旁路字典。
|
||||
|
||||
产出:
|
||||
(is_content, text)。
|
||||
|
||||
异常:
|
||||
_SseAnomaly: data 段 JSON 解析失败。
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
# ── Provider 差异处理 ────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _build_thinking_body(provider: str) -> dict:
|
||||
"""构造 thinking/reasoning 启用参数（provider 差异处理）。
|
||||
|
||||
参数:
|
||||
provider: provider 标识（如 "deepseek"、"qwen" 等）。
|
||||
|
||||
返回:
|
||||
合并进请求体的字典片段。
|
||||
"""
|
||||
provider_lower = provider.lower()
|
||||
if "deepseek" in provider_lower:
|
||||
return {"thinking": {"type": "enabled"}}
|
||||
if "qwen" in provider_lower:
|
||||
return {"enable_thinking": True}
|
||||
return {}
|
||||
|
||||
|
||||
_THINK_PATTERN = re.compile(r"<think>(.*?)</think>", re.DOTALL)
|
||||
|
||||
|
||||
def _strip_qwen_thinking(content: str) -> tuple[str, str]:
|
||||
"""剥离 Qwen 模型 <think>...</think> 标签，拆分为 (正文, 思考内容)。
|
||||
|
||||
参数:
|
||||
content: 含有可能的 <think> 标签的模型原始输出。
|
||||
|
||||
返回:
|
||||
(stripped_content, thinking_text)。无 think 标签时 thinking 为空字符串。
|
||||
"""
|
||||
match = _THINK_PATTERN.search(content)
|
||||
if match is None:
|
||||
return content, ""
|
||||
thinking = match.group(1).strip()
|
||||
stripped = _THINK_PATTERN.sub("", content).strip()
|
||||
return stripped, thinking
|
||||
|
||||
|
||||
# ── 瞬时错误判定 ─────────────────────────────────────────────────────
|
||||
|
||||
_TRANSIENT_STATUS_CODES = frozenset({429, 500, 502, 503, 504})
|
||||
_FATAL_STATUS_CODES = frozenset({401, 403})
|
||||
|
||||
|
||||
def _is_transient_error(exc: Exception) -> bool:
|
||||
"""判断异常是否为瞬时可重试错误。
|
||||
|
||||
参数:
|
||||
exc: 捕获的异常。
|
||||
|
||||
返回:
|
||||
True 表示可重试，False 表示不可重试。
|
||||
"""
|
||||
if isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout, httpx.WriteTimeout)):
|
||||
return True
|
||||
if isinstance(exc, httpx.HTTPStatusError):
|
||||
return exc.response.status_code in _TRANSIENT_STATUS_CODES
|
||||
return isinstance(exc, (StreamLivenessTimeout, _SseAnomaly))
|
||||
|
||||
|
||||
def _is_fatal_auth_error(exc: Exception) -> bool:
|
||||
"""判断异常是否为不可恢复的认证/授权错误。
|
||||
|
||||
参数:
|
||||
exc: 捕获的异常。
|
||||
|
||||
返回:
|
||||
True 表示应立即 force_open 熔断器。
|
||||
"""
|
||||
if isinstance(exc, httpx.HTTPStatusError):
|
||||
return exc.response.status_code in _FATAL_STATUS_CODES
|
||||
return False
|
||||
|
||||
|
||||
# ── GovernedLLMClient ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class GovernedLLMClient:
|
||||
"""四层治理栈统一 LLM 调用网关，实现 LLMProvider Protocol。
|
||||
|
||||
构造函数接收所有组件（依赖注入），不从全局状态获取任何配置。
|
||||
|
||||
参数:
|
||||
model: 模型名称。
|
||||
base_url: API 基础 URL（如 http://localhost:8000/v1）。
|
||||
api_key: API 密钥。
|
||||
provider: provider 标识（如 "deepseek"、"qwen"）。
|
||||
thinking: 是否启用 thinking/reasoning 模式。
|
||||
breaker: 熔断器实例。
|
||||
cache: Redis 响应缓存实例。
|
||||
telemetry: 遥测记录器实例。
|
||||
timeout_s: 总超时秒数。
|
||||
ttft_timeout_s: 首 token 超时秒数。
|
||||
inter_token_timeout_s: token 间超时秒数。
|
||||
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,
|
||||
telemetry: Any,
|
||||
timeout_s: float,
|
||||
ttft_timeout_s: float,
|
||||
inter_token_timeout_s: float,
|
||||
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
|
||||
self._inter_token_timeout_s = inter_token_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 {self._api_key}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=httpx.Timeout(timeout_s),
|
||||
)
|
||||
|
||||
async def chat(
|
||||
self,
|
||||
messages: list[dict[str, Any]],
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
parent_call_id: str | None = None,
|
||||
) -> LLMResponse:
|
||||
"""发起 LLM 调用，经四层治理栈：熔断 → 缓存 → 重试+流式 → 遥测。
|
||||
|
||||
参数:
|
||||
messages: OpenAI 格式消息列表。
|
||||
session_id: 会话 ID（传递到遥测）。
|
||||
parent_call_id: 父调用 ID（传递到遥测）。
|
||||
|
||||
返回:
|
||||
LLMResponse 统一响应。
|
||||
|
||||
异常:
|
||||
CircuitOpenError: 熔断器开路。
|
||||
httpx.HTTPStatusError: 不可恢复的 HTTP 错误（401/403）。
|
||||
"""
|
||||
started = time.monotonic()
|
||||
call_id = str(uuid4())
|
||||
|
||||
# ① 熔断器检查
|
||||
if self._breaker.is_open(self._provider, time.monotonic()):
|
||||
raise CircuitOpenError(
|
||||
f"熔断器已开启，拒绝调用 provider={self._provider}"
|
||||
)
|
||||
|
||||
# ② 缓存查询
|
||||
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=json.dumps(messages, ensure_ascii=False),
|
||||
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
|
||||
|
||||
# ③ 重试循环 + 流式消费
|
||||
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)
|
||||
)
|
||||
|
||||
# 熔断器记成功
|
||||
self._breaker.record_success(self._provider)
|
||||
|
||||
# provider 差异：Qwen think 标签剥离
|
||||
if "qwen" in self._provider.lower() and "<think>" in content:
|
||||
content, qwen_thinking = _strip_qwen_thinking(content)
|
||||
if qwen_thinking:
|
||||
thinking_text = qwen_thinking
|
||||
|
||||
latency_ms = int((time.monotonic() - started) * 1000)
|
||||
prompt_tokens = usage.get("prompt_tokens", 0)
|
||||
completion_tokens = usage.get("completion_tokens", 0)
|
||||
|
||||
response = LLMResponse(
|
||||
content=content,
|
||||
thinking=thinking_text,
|
||||
model=self._model,
|
||||
provider=self._provider,
|
||||
prompt_tokens=prompt_tokens,
|
||||
completion_tokens=completion_tokens,
|
||||
latency_ms=latency_ms,
|
||||
ttft_ms=ttft_ms,
|
||||
max_inter_token_ms=max_itoken_ms,
|
||||
cache_hit=False,
|
||||
call_id=call_id,
|
||||
)
|
||||
|
||||
# ④ 写缓存
|
||||
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=json.dumps(messages, ensure_ascii=False),
|
||||
response=content,
|
||||
thinking=thinking_text,
|
||||
prompt_tokens=prompt_tokens,
|
||||
completion_tokens=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 Exception as exc:
|
||||
last_exc = exc
|
||||
attempt_latency = int((time.monotonic() - attempt_start) * 1000)
|
||||
|
||||
# 致命错误（401/403）→ force_open + 遥测 + 立即抛出
|
||||
if _is_fatal_auth_error(exc):
|
||||
self._breaker.force_open(self._provider, time.monotonic())
|
||||
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=json.dumps(messages, ensure_ascii=False),
|
||||
response="",
|
||||
thinking="",
|
||||
prompt_tokens=0,
|
||||
completion_tokens=0,
|
||||
latency_ms=attempt_latency,
|
||||
ttft_ms=None,
|
||||
max_inter_token_ms=None,
|
||||
cache_hit=False,
|
||||
error=str(exc),
|
||||
)
|
||||
raise
|
||||
|
||||
# 瞬时错误 → 记失败 + 遥测 + 退避重试
|
||||
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
|
||||
parent_call_id=parent_call_id,
|
||||
session_id=session_id,
|
||||
model_name=self._model,
|
||||
provider=self._provider,
|
||||
messages=json.dumps(messages, ensure_ascii=False),
|
||||
response="",
|
||||
thinking="",
|
||||
prompt_tokens=0,
|
||||
completion_tokens=0,
|
||||
latency_ms=attempt_latency,
|
||||
ttft_ms=None,
|
||||
max_inter_token_ms=None,
|
||||
cache_hit=False,
|
||||
error=str(exc),
|
||||
)
|
||||
logger.warning(
|
||||
"llm.call.transient_error",
|
||||
extra={
|
||||
"provider": self._provider,
|
||||
"attempt": attempt + 1,
|
||||
"max_retries": self._max_retries,
|
||||
"error": str(exc),
|
||||
},
|
||||
)
|
||||
if attempt < self._max_retries - 1:
|
||||
delay = min(
|
||||
self._retry_base_delay_s * (2 ** attempt),
|
||||
self._retry_max_delay_s,
|
||||
)
|
||||
if delay > 0:
|
||||
await asyncio.sleep(delay)
|
||||
continue
|
||||
|
||||
# 非瞬时、非致命 → 记遥测后直接抛出
|
||||
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=json.dumps(messages, ensure_ascii=False),
|
||||
response="",
|
||||
thinking="",
|
||||
prompt_tokens=0,
|
||||
completion_tokens=0,
|
||||
latency_ms=attempt_latency,
|
||||
ttft_ms=None,
|
||||
max_inter_token_ms=None,
|
||||
cache_hit=False,
|
||||
error=str(exc),
|
||||
)
|
||||
raise
|
||||
|
||||
# 所有重试耗尽
|
||||
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 行。
|
||||
|
||||
参数:
|
||||
messages: OpenAI 格式消息列表。
|
||||
|
||||
返回:
|
||||
SSE 文本行列表。
|
||||
|
||||
异常:
|
||||
httpx.HTTPStatusError: HTTP 状态码错误。
|
||||
httpx.ConnectError: 连接错误。
|
||||
"""
|
||||
payload: dict[str, Any] = {
|
||||
"model": self._model,
|
||||
"messages": messages,
|
||||
"stream": True,
|
||||
"stream_options": {"include_usage": True},
|
||||
}
|
||||
if self._thinking:
|
||||
payload.update(_build_thinking_body(self._provider))
|
||||
|
||||
collected_lines: list[str] = []
|
||||
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
|
||||
|
||||
async def _consume_stream(
|
||||
self, sse_lines: list[str]
|
||||
) -> tuple[str, str, float | None, float | None, dict]:
|
||||
"""消费 SSE 行序列，累积 content/thinking，测量 TTFT 和 max_inter_token_ms。
|
||||
|
||||
参数:
|
||||
sse_lines: SSE 文本行列表。
|
||||
|
||||
返回:
|
||||
(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)
|
||||
guarded = stream_with_liveness_timeouts(
|
||||
raw_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
|
||||
stream_start = time.monotonic()
|
||||
last_token_time = stream_start
|
||||
|
||||
async for is_content, text in guarded:
|
||||
now = time.monotonic()
|
||||
if ttft_ms is None:
|
||||
ttft_ms = (now - stream_start) * 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)
|
||||
usage = usage_sink.get("usage", {})
|
||||
return content, thinking, ttft_ms, max_inter_token_ms, usage
|
||||
|
||||
async def close(self) -> None:
|
||||
"""关闭底层 HTTP 客户端。"""
|
||||
await self._http.aclose()
|
||||
Reference in New Issue
Block a user