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 事件
251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""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
|