chenxin
b4242f3844
- 迁移 7+8：logical_model（能力目录）、provider_endpoint（密钥只存环境变量名，永不落库）、turn、model_call（每 attempt 单独持久化：usage/TTFT/错误，Provider 未返回保持 null 不估算——ADR-003）；message.run_id 关联 - OpenAI-compatible 流式适配器：httpx SSE 解析、标准化 StreamChunk（delta/usage/done）、transport 可注入（测试与故障注入）；密钥经环境变量读取不进日志 - Run 执行器：后台执行完整生命周期——running → content.block.started → 真实 content.block.delta*（流式产生，delta_index 单调）→ content.block.completed → message.completed → 终态；assistant 消息为权威快照；ModelCall 记录 usage 与 TTFT；无可用端点时 Run 显式失败并留 error 事件（fail-closed） - post_message：响应提交后由框架 BackgroundTasks 执行 Run（显式预提交保证 Run 落库）；SSE 改为增量轮询（重放 + 等待新事件至终态，30s 上限） - 测试 54 项全绿（+2）：端到端 mock 流验证 ModelCall usage/TTFT/assistant 权威消息、无端点 Run failed + content.block.failed 事件
108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
"""OpenAI-compatible 流式适配器（WP-05）。
|
|
|
|
- 只消费逻辑请求（messages + 逻辑模型），输出标准化的流事件与 usage。
|
|
- 密钥从环境变量读取（endpoint.api_key_env），不落库、不进日志。
|
|
- Provider 未返回的 usage 字段保持 None（ADR-003：不估算伪装）。
|
|
- transport 可注入，供测试与故障注入。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from collections.abc import AsyncIterator
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class GatewayError(RuntimeError):
|
|
"""网关错误：稳定分类，供 ModelCall 失败记录与上层重试决策。"""
|
|
|
|
|
|
@dataclass
|
|
class StreamChunk:
|
|
kind: str # "delta" | "usage" | "done"
|
|
text: str | None = None
|
|
usage: dict[str, int] | None = None
|
|
finish_reason: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class AdapterRequest:
|
|
messages: list[dict[str, str]] # {"role", "content"}：纯文本最小集
|
|
model: str # 上游模型 ID
|
|
|
|
|
|
def _parse_sse_line(line: str) -> dict[str, Any] | None:
|
|
if not line.startswith("data:"):
|
|
return None
|
|
payload = line.removeprefix("data:").strip()
|
|
if not payload or payload == "[DONE]":
|
|
return None
|
|
try:
|
|
parsed: dict[str, Any] = json.loads(payload)
|
|
return parsed
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
|
|
def _usage_from_final(chunk: dict[str, Any]) -> dict[str, int] | None:
|
|
usage = chunk.get("usage")
|
|
if not isinstance(usage, dict):
|
|
return None
|
|
result: dict[str, int] = {}
|
|
keys = (
|
|
"prompt_tokens",
|
|
"completion_tokens",
|
|
"total_tokens",
|
|
"cache_creation_input_tokens",
|
|
"cache_read_input_tokens",
|
|
)
|
|
for key in keys:
|
|
value = usage.get(key)
|
|
if isinstance(value, int):
|
|
result[key] = value
|
|
return result or None
|
|
|
|
|
|
async def stream_chat(
|
|
base_url: str,
|
|
model_id: str,
|
|
api_key_env: str,
|
|
request: AdapterRequest,
|
|
*,
|
|
request_timeout_s: float = 120.0,
|
|
transport: httpx.AsyncBaseTransport | None = None,
|
|
) -> AsyncIterator[StreamChunk]:
|
|
"""流式调用 OpenAI-compatible /chat/completions，产出标准化块。"""
|
|
api_key = os.environ.get(api_key_env, "")
|
|
if not api_key:
|
|
raise GatewayError(f"环境变量 {api_key_env} 未配置 Provider 密钥")
|
|
url = base_url.rstrip("/") + "/chat/completions"
|
|
body = {
|
|
"model": model_id,
|
|
"messages": request.messages,
|
|
"stream": True,
|
|
"stream_options": {"include_usage": True},
|
|
}
|
|
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
|
|
async with httpx.AsyncClient(timeout=request_timeout_s, transport=transport) as client:
|
|
async with client.stream("POST", url, json=body, headers=headers) as resp:
|
|
if resp.status_code != 200:
|
|
detail = (await resp.aread()).decode("utf-8", "replace")[:500]
|
|
raise GatewayError(f"上游返回 {resp.status_code}: {detail}")
|
|
async for line in resp.aiter_lines():
|
|
chunk = _parse_sse_line(line)
|
|
if chunk is None:
|
|
continue
|
|
for choice in chunk.get("choices", []):
|
|
text = choice.get("delta", {}).get("content")
|
|
if text:
|
|
yield StreamChunk(kind="delta", text=text)
|
|
finish = choice.get("finish_reason")
|
|
if finish:
|
|
yield StreamChunk(kind="done", finish_reason=finish)
|
|
usage = _usage_from_final(chunk)
|
|
if usage is not None:
|
|
yield StreamChunk(kind="usage", usage=usage)
|