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 事件
331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""对话域路由：创建会话、发送消息（幂等原子）与 Run SSE。"""
|
||
|
||
import json
|
||
from collections.abc import AsyncIterator
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, BackgroundTasks, Depends, Header, HTTPException, Request, status
|
||
from fastapi.responses import JSONResponse, StreamingResponse
|
||
from sqlalchemy import select
|
||
|
||
from app.auth.dependencies import get_current_user, require_state_change
|
||
from app.conversations import service
|
||
from app.conversations.schemas import MessageCreatedOut, MessageIn, MessageOut, RunOut
|
||
from app.db.session import DbSession
|
||
from app.models import PersonalDomain, Run, User
|
||
|
||
router = APIRouter(prefix="/v1/conversations", tags=["conversations"])
|
||
|
||
|
||
async def _domain_of(db: DbSession, user: User) -> PersonalDomain:
|
||
result = await db.execute(
|
||
select(PersonalDomain).where(PersonalDomain.user_id == user.id).limit(1)
|
||
)
|
||
domain = result.scalar_one_or_none()
|
||
if domain is None:
|
||
raise HTTPException(status.HTTP_409_CONFLICT, "用户缺少个人域")
|
||
return domain
|
||
|
||
|
||
@router.post("", status_code=201)
|
||
async def create_conversation(
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(require_state_change)],
|
||
request: Request,
|
||
body: dict[str, str],
|
||
) -> dict[str, str]:
|
||
domain = await _domain_of(db, user)
|
||
conversation = await service.create_conversation(
|
||
db, domain.id, body.get("title", "未命名对话"), getattr(request.state, "trace_id", None)
|
||
)
|
||
return {"id": conversation.id, "title": conversation.title}
|
||
|
||
|
||
@router.post(
|
||
"/{conversation_id}/messages",
|
||
status_code=201,
|
||
response_model=None,
|
||
)
|
||
async def post_message(
|
||
conversation_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(require_state_change)],
|
||
request: Request,
|
||
background_tasks: BackgroundTasks,
|
||
body: MessageIn,
|
||
idempotency_key: Annotated[str, Header(alias="Idempotency-Key")],
|
||
) -> MessageCreatedOut | JSONResponse:
|
||
if not idempotency_key or len(idempotency_key) > 128:
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "Idempotency-Key 不合法")
|
||
domain = await _domain_of(db, user)
|
||
message, run, replayed = await service.replay_or_execute_message(
|
||
db,
|
||
domain.id,
|
||
conversation_id,
|
||
idempotency_key,
|
||
body,
|
||
getattr(request.state, "trace_id", None),
|
||
)
|
||
payload = MessageCreatedOut(
|
||
message=MessageOut(
|
||
id=message.id,
|
||
conversation_id=message.conversation_id,
|
||
branch_id=message.branch_id,
|
||
role=message.role,
|
||
content=message.content,
|
||
parser_version=message.parser_version,
|
||
created_at=message.created_at.isoformat(),
|
||
),
|
||
run=(
|
||
RunOut(
|
||
id=run.id,
|
||
conversation_id=run.conversation_id,
|
||
status=run.status,
|
||
created_at=run.created_at.isoformat(),
|
||
completed_at=run.completed_at.isoformat() if run.completed_at else None,
|
||
)
|
||
if run is not None
|
||
else None
|
||
),
|
||
)
|
||
if replayed:
|
||
# 重放成功按契约返回 200 与原资源（避开装饰器固定 201）
|
||
return JSONResponse(status_code=200, content=json.loads(payload.model_dump_json()))
|
||
if run is not None:
|
||
# 先显式提交：确保 Run 在后台任务执行前已落库（与依赖清理顺序解耦）
|
||
await db.commit()
|
||
from app.gateway import execute_run
|
||
|
||
background_tasks.add_task(
|
||
execute_run,
|
||
request.app.state.session_factory,
|
||
run.id,
|
||
transport=getattr(request.app.state, "gw_transport", None),
|
||
)
|
||
return payload
|
||
|
||
|
||
run_router = APIRouter(prefix="/v1/runs", tags=["runs"])
|
||
|
||
|
||
@run_router.get("/{run_id}/events")
|
||
async def run_events(
|
||
run_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(get_current_user)],
|
||
) -> StreamingResponse:
|
||
"""Run SSE：重放持久化事件流（sequence 单调），客户端按序去重（契约第 5 节）。"""
|
||
result = await db.execute(select(Run).where(Run.id == run_id))
|
||
run = result.scalar_one_or_none()
|
||
if run is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在")
|
||
domain = await _domain_of(db, user)
|
||
if run.personal_domain_id != domain.id:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在")
|
||
|
||
import asyncio as _asyncio
|
||
|
||
from app.models import RunEvent
|
||
|
||
terminal = {"completed", "failed", "cancelled"}
|
||
|
||
async def event_stream() -> AsyncIterator[str]:
|
||
last_seq = -1
|
||
deadline = _asyncio.get_event_loop().time() + 30.0
|
||
while True:
|
||
rows = list(
|
||
(
|
||
await db.execute(
|
||
select(RunEvent)
|
||
.where(RunEvent.run_id == run.id, RunEvent.sequence > last_seq)
|
||
.order_by(RunEvent.sequence.asc())
|
||
)
|
||
).scalars().all()
|
||
)
|
||
for event in rows:
|
||
payload = json.dumps(
|
||
{"type": event.type, "data": event.data, "sequence": event.sequence},
|
||
ensure_ascii=False,
|
||
)
|
||
yield f"data: {payload}\n\n"
|
||
last_seq = event.sequence
|
||
if not rows:
|
||
await db.rollback() # 释放快照，下一轮可见新写入
|
||
await db.refresh(run)
|
||
if run.status in terminal:
|
||
break
|
||
if _asyncio.get_event_loop().time() > deadline:
|
||
break
|
||
await _asyncio.sleep(0.05)
|
||
|
||
return StreamingResponse(
|
||
event_stream(),
|
||
media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-store"},
|
||
)
|
||
|
||
|
||
@router.get("/{conversation_id}/messages")
|
||
async def list_messages(
|
||
conversation_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(get_current_user)],
|
||
limit: int = 50,
|
||
cursor: str | None = None,
|
||
branch_id: str | None = None,
|
||
) -> dict[str, object]:
|
||
"""消息列表：opaque cursor 分页（默认 50，契约第 4.1 节）。"""
|
||
if not 1 <= limit <= 200:
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "limit 取值范围 1-200")
|
||
domain = await _domain_of(db, user)
|
||
page, next_cursor, has_more = await service.list_messages(
|
||
db, domain.id, conversation_id, branch_id, limit, cursor
|
||
)
|
||
from app.conversations.schemas import MessageOut
|
||
|
||
return {
|
||
"data": [
|
||
MessageOut(
|
||
id=m.id,
|
||
conversation_id=m.conversation_id,
|
||
branch_id=m.branch_id,
|
||
role=m.role,
|
||
content=m.content,
|
||
parser_version=m.parser_version,
|
||
created_at=m.created_at.isoformat(),
|
||
).model_dump()
|
||
for m in page
|
||
],
|
||
"next_cursor": next_cursor,
|
||
"has_more": has_more,
|
||
}
|
||
|
||
|
||
@router.get("/{conversation_id}/branches")
|
||
async def list_branches(
|
||
conversation_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(get_current_user)],
|
||
) -> dict[str, object]:
|
||
domain = await _domain_of(db, user)
|
||
branches = await service.list_branches(db, domain.id, conversation_id)
|
||
from app.conversations.schemas import BranchOut
|
||
|
||
return {
|
||
"data": [
|
||
BranchOut(
|
||
id=b.id, conversation_id=b.conversation_id, title=b.title,
|
||
created_at=b.created_at.isoformat(),
|
||
).model_dump()
|
||
for b in branches
|
||
]
|
||
}
|
||
|
||
|
||
@router.post("/{conversation_id}/current-branch")
|
||
async def set_current_branch(
|
||
conversation_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(require_state_change)],
|
||
body: dict[str, str],
|
||
) -> dict[str, str]:
|
||
"""切换当前路径（主脉工作入口；采用≠归真）。"""
|
||
domain = await _domain_of(db, user)
|
||
conversation = await service.set_current_branch(
|
||
db, domain.id, conversation_id, body.get("branch_id", "")
|
||
)
|
||
return {"id": conversation.id, "current_branch_id": conversation.current_branch_id or ""}
|
||
|
||
|
||
@router.get("/{conversation_id}")
|
||
async def get_conversation(
|
||
conversation_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(get_current_user)],
|
||
) -> dict[str, object]:
|
||
domain = await _domain_of(db, user)
|
||
conversation = await service.require_conversation(db, domain.id, conversation_id)
|
||
from app.conversations.schemas import ConversationOut
|
||
|
||
return ConversationOut(
|
||
id=conversation.id,
|
||
title=conversation.title,
|
||
current_branch_id=conversation.current_branch_id,
|
||
created_at=conversation.created_at.isoformat(),
|
||
).model_dump() | {"realm": conversation.realm}
|
||
|
||
|
||
@router.post("/{conversation_id}/realm")
|
||
async def set_realm(
|
||
conversation_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(require_state_change)],
|
||
request: Request,
|
||
body: dict[str, str],
|
||
) -> dict[str, str]:
|
||
"""入境/出境：切换对话界域（显式、留审计）。"""
|
||
domain = await _domain_of(db, user)
|
||
conversation = await service.set_realm(
|
||
db,
|
||
domain.id,
|
||
conversation_id,
|
||
body.get("realm", ""),
|
||
getattr(request.state, "trace_id", None),
|
||
)
|
||
return {"id": conversation.id, "realm": conversation.realm}
|
||
|
||
|
||
message_router = APIRouter(prefix="/v1/messages", tags=["messages"])
|
||
|
||
|
||
@message_router.post("/{message_id}/canonizations", status_code=201, response_model=None)
|
||
async def canonize_message(
|
||
message_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(require_state_change)],
|
||
request: Request,
|
||
body: dict[str, str],
|
||
) -> JSONResponse | dict[str, str | None]:
|
||
"""归真：携带 canonical_scope（ADR-020）；重复归真幂等返回原状态（200）。"""
|
||
from fastapi.responses import JSONResponse
|
||
|
||
domain = await _domain_of(db, user)
|
||
message, replayed = await service.canonize_message(
|
||
db, domain.id, message_id, body.get("scope", ""), getattr(request.state, "trace_id", None)
|
||
)
|
||
payload = {
|
||
"message_id": message.id,
|
||
"canonical_scope": message.realm,
|
||
"canonized_at": message.canonized_at.isoformat() if message.canonized_at else None,
|
||
}
|
||
if replayed:
|
||
return JSONResponse(status_code=200, content=payload)
|
||
return payload
|
||
|
||
|
||
@message_router.post("/{message_id}/revisions", status_code=201)
|
||
async def revise_message(
|
||
message_id: str,
|
||
db: DbSession,
|
||
user: Annotated[User, Depends(require_state_change)],
|
||
request: Request,
|
||
body: MessageIn,
|
||
) -> dict[str, object]:
|
||
"""编辑版本：新消息引用原消息；原消息永不修改。"""
|
||
domain = await _domain_of(db, user)
|
||
revision = await service.revise_message(
|
||
db, domain.id, message_id, body, getattr(request.state, "trace_id", None)
|
||
)
|
||
from app.conversations.schemas import MessageOut
|
||
|
||
return MessageOut(
|
||
id=revision.id,
|
||
conversation_id=revision.conversation_id,
|
||
branch_id=revision.branch_id,
|
||
role=revision.role,
|
||
content=revision.content,
|
||
parser_version=revision.parser_version,
|
||
created_at=revision.created_at.isoformat(),
|
||
).model_dump() | {"parent_message_id": revision.parent_message_id,
|
||
"variant_kind": revision.variant_kind}
|