✨ feat: WP-05 模型网关核心——逻辑模型、流式适配与 Run 真实执行
- 迁移 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 事件
This commit is contained in:
chenxin committed
1 parent
40221170ae
commit
b4242f3844
10 files changed
+789
-56
No files matched your search
@@ -0,0 +1,6 @@
|
||||
"""模型网关：路由与 Run 执行（WP-05）。"""
|
||||
|
||||
from .adapters import AdapterRequest, GatewayError, stream_chat
|
||||
from .runner import execute_run
|
||||
|
||||
__all__ = ["AdapterRequest", "GatewayError", "stream_chat", "execute_run"]
|
||||
@@ -0,0 +1,107 @@
|
||||
"""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)
|
||||
@@ -0,0 +1,250 @@
|
||||
"""Run 执行器：调用模型、产生真实流事件与 assistant 消息（WP-05）。
|
||||
|
||||
- 后台任务执行：独立数据库会话（不复用请求会话）。
|
||||
- 事件序列（契约第 5 节）：run.status.changed(running) → content.block.started
|
||||
→ content.block.delta* → content.block.completed → message.completed →
|
||||
run.status.changed(completed/failed)。
|
||||
- 首个用户可见 delta 前的失败允许显式 fail；无可用端点时 Run 失败并留 error 事件。
|
||||
"""
|
||||
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from app.core.ids import new_id
|
||||
from app.gateway.adapters import AdapterRequest, GatewayError, stream_chat
|
||||
from app.models import (
|
||||
AuditEvent,
|
||||
Message,
|
||||
ModelCall,
|
||||
OutboxEvent,
|
||||
ProviderEndpoint,
|
||||
Run,
|
||||
RunEvent,
|
||||
Turn,
|
||||
)
|
||||
|
||||
|
||||
async def _next_sequence(db: AsyncSession, run_id: str) -> int:
|
||||
last = await db.execute(
|
||||
select(RunEvent.sequence)
|
||||
.where(RunEvent.run_id == run_id)
|
||||
.order_by(RunEvent.sequence.desc())
|
||||
.limit(1)
|
||||
)
|
||||
row = last.scalar_one_or_none()
|
||||
return (row + 1) if row is not None else 0
|
||||
|
||||
|
||||
def _emit(db: AsyncSession, run_id: str, seq: int, event_type: str, data: dict[str, Any]) -> int:
|
||||
db.add(RunEvent(run_id=run_id, sequence=seq, type=event_type, data=data))
|
||||
return seq + 1
|
||||
|
||||
|
||||
async def execute_run(
|
||||
factory: async_sessionmaker[AsyncSession],
|
||||
run_id: str,
|
||||
transport: Any | None = None,
|
||||
) -> None:
|
||||
"""执行一个 Run 至终态。transport 注入仅供测试。"""
|
||||
async with factory() as db:
|
||||
try:
|
||||
run = await db.get(Run, run_id)
|
||||
if run is None:
|
||||
return
|
||||
seq = await _next_sequence(db, run_id)
|
||||
# 状态事件：running
|
||||
run.status = "running"
|
||||
seq = _emit(
|
||||
db, run_id, seq, "run.status.changed",
|
||||
{"run_id": run_id, "status": "running"},
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
# 路由：按逻辑模型选择启用端点（优先级升序）
|
||||
endpoint = (
|
||||
await db.execute(
|
||||
select(ProviderEndpoint)
|
||||
.where(
|
||||
ProviderEndpoint.enabled.is_(True),
|
||||
ProviderEndpoint.logical_model_id == "default-chat",
|
||||
)
|
||||
.order_by(ProviderEndpoint.priority.asc())
|
||||
.limit(1)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
turn = Turn(run_id=run_id, index=0)
|
||||
db.add(turn)
|
||||
await db.flush()
|
||||
|
||||
if endpoint is None:
|
||||
# fail closed：无可用端点时 Run 显式失败，不静默降级
|
||||
call = ModelCall(
|
||||
turn_id=turn.id, status="failed",
|
||||
request_summary={}, error="无可用 Provider 端点（default-chat）",
|
||||
started_at=datetime.now(UTC),
|
||||
completed_at=datetime.now(UTC),
|
||||
)
|
||||
db.add(call)
|
||||
run.status = "failed"
|
||||
run.completed_at = datetime.now(UTC)
|
||||
seq = _emit(db, run_id, seq, "content.block.failed", {
|
||||
"run_id": run_id, "message_id": None, "index": 0,
|
||||
"block": {"id": new_id("block_"), "type": "text", "status": "failed"},
|
||||
"error": "no_endpoint_available",
|
||||
})
|
||||
seq = _emit(db, run_id, seq, "message.completed",
|
||||
{"run_id": run_id, "message_ids": [], "final_status": "failed"})
|
||||
seq = _emit(db, run_id, seq, "run.status.changed",
|
||||
{"run_id": run_id, "status": "failed"})
|
||||
await db.commit()
|
||||
return
|
||||
|
||||
# 对话历史（最小：当前分支文本消息）
|
||||
messages = (
|
||||
await db.execute(
|
||||
select(Message)
|
||||
.where(
|
||||
Message.conversation_id == run.conversation_id,
|
||||
Message.role.in_(["user", "assistant"]),
|
||||
)
|
||||
.order_by(Message.created_at.asc(), Message.id.asc())
|
||||
.limit(50)
|
||||
)
|
||||
).scalars().all()
|
||||
prompt = [
|
||||
{
|
||||
"role": m.role,
|
||||
"content": "".join(
|
||||
str(b.get("text", b.get("source", "")))
|
||||
for b in m.content
|
||||
if isinstance(b, dict)
|
||||
),
|
||||
}
|
||||
for m in messages
|
||||
]
|
||||
|
||||
call = ModelCall(
|
||||
turn_id=turn.id, endpoint_id=endpoint.id, status="streaming",
|
||||
request_summary={"messages": len(prompt), "model_id": endpoint.model_id},
|
||||
started_at=datetime.now(UTC),
|
||||
)
|
||||
db.add(call)
|
||||
await db.commit()
|
||||
|
||||
block_id = new_id("block_")
|
||||
seq = _emit(db, run_id, seq, "content.block.started", {
|
||||
"run_id": run_id, "index": 0,
|
||||
"block": {"id": block_id, "type": "text", "status": "streaming"},
|
||||
})
|
||||
await db.commit()
|
||||
|
||||
# 流式消费：真实 delta 事件
|
||||
started = time.monotonic()
|
||||
ttft_ms: int | None = None
|
||||
full_text: list[str] = []
|
||||
usage: dict[str, int] | None = None
|
||||
finish_reason: str | None = None
|
||||
delta_index = 0
|
||||
try:
|
||||
adapter_req = AdapterRequest(messages=prompt, model=endpoint.model_id)
|
||||
async for chunk in stream_chat(
|
||||
endpoint.base_url, endpoint.model_id, endpoint.api_key_env,
|
||||
adapter_req, transport=transport,
|
||||
):
|
||||
if chunk.kind == "delta" and chunk.text:
|
||||
if ttft_ms is None:
|
||||
ttft_ms = int((time.monotonic() - started) * 1000)
|
||||
full_text.append(chunk.text)
|
||||
seq = _emit(db, run_id, seq, "content.block.delta", {
|
||||
"run_id": run_id, "block_id": block_id,
|
||||
"delta_index": delta_index,
|
||||
"delta": {"type": "text_append", "text": chunk.text},
|
||||
})
|
||||
delta_index += 1
|
||||
await db.commit()
|
||||
elif chunk.kind == "usage" and chunk.usage:
|
||||
usage = chunk.usage
|
||||
elif chunk.kind == "done" and chunk.finish_reason:
|
||||
finish_reason = chunk.finish_reason
|
||||
except GatewayError as exc:
|
||||
failed_call = await db.get(ModelCall, call.id)
|
||||
assert failed_call is not None
|
||||
call.status = "failed"
|
||||
call.error = str(exc)[:500]
|
||||
call.completed_at = datetime.now(UTC)
|
||||
run.status = "failed"
|
||||
run.completed_at = datetime.now(UTC)
|
||||
seq = _emit(db, run_id, seq, "content.block.failed", {
|
||||
"run_id": run_id, "block_id": block_id, "index": 0,
|
||||
"error": "upstream_error",
|
||||
})
|
||||
seq = _emit(db, run_id, seq, "message.completed",
|
||||
{"run_id": run_id, "message_ids": [], "final_status": "failed"})
|
||||
seq = _emit(db, run_id, seq, "run.status.changed",
|
||||
{"run_id": run_id, "status": "failed"})
|
||||
await db.commit()
|
||||
return
|
||||
|
||||
text = "".join(full_text)
|
||||
seq = _emit(db, run_id, seq, "content.block.completed", {
|
||||
"run_id": run_id, "index": 0,
|
||||
"block": {"id": block_id, "type": "text", "status": "completed", "text": text},
|
||||
})
|
||||
|
||||
# assistant 消息（权威快照）
|
||||
assistant = Message(
|
||||
conversation_id=run.conversation_id,
|
||||
branch_id=(
|
||||
await db.execute(
|
||||
select(Message.branch_id).where(Message.run_id == run_id).limit(1)
|
||||
)
|
||||
).scalar_one_or_none() or new_id("branch_"),
|
||||
role="assistant",
|
||||
realm="reality",
|
||||
content=[{"id": block_id, "type": "text", "text": text, "status": "completed"}],
|
||||
run_id=run_id,
|
||||
)
|
||||
db.add(assistant)
|
||||
await db.flush()
|
||||
seq = _emit(db, run_id, seq, "message.completed", {
|
||||
"run_id": run_id, "message_ids": [assistant.id], "final_status": "completed",
|
||||
})
|
||||
run.status = "completed"
|
||||
run.completed_at = datetime.now(UTC)
|
||||
seq = _emit(db, run_id, seq, "run.status.changed",
|
||||
{"run_id": run_id, "status": "completed"})
|
||||
|
||||
success_call = await db.get(ModelCall, call.id)
|
||||
assert success_call is not None
|
||||
success_call.status = "succeeded"
|
||||
# Provider 未返回的 usage 保持 None：不估算伪装（ADR-003）
|
||||
success_call.usage = usage
|
||||
success_call.ttft_ms = ttft_ms
|
||||
success_call.completed_at = datetime.now(UTC)
|
||||
if finish_reason:
|
||||
success_call.request_summary = {
|
||||
**success_call.request_summary, "finish_reason": finish_reason
|
||||
}
|
||||
db.add(
|
||||
AuditEvent(
|
||||
actor_type="system", actor_id="gateway", action="model_call.completed",
|
||||
aggregate_type="run", aggregate_id=run_id,
|
||||
payload={"endpoint": endpoint.name, "usage": usage, "ttft_ms": ttft_ms},
|
||||
)
|
||||
)
|
||||
db.add(
|
||||
OutboxEvent(
|
||||
aggregate_type="message", aggregate_id=assistant.id,
|
||||
event_type="message.created",
|
||||
payload={"message_id": assistant.id, "run_id": run_id, "role": "assistant"},
|
||||
)
|
||||
)
|
||||
await db.commit()
|
||||
except Exception:
|
||||
await db.rollback()
|
||||
raise
|
||||
Reference in new issue
Block a user