CIAgent
4c52d29f91
---ci--- phase: 2 milestone: v0.2 status: complete ---/ci--- REQ-2-004 complete. BaseAgent ABC, agent-scoped sessions (20-msg window, 500-cap LRU), registry, 4-layer structured output defense, 6-module prompt library, D-021-aligned learner corpus, chat session persistence. 61/61 tests, ruff clean.
97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""POST /v1/chat/stream — SSE chat with the D-016 envelope.
|
|
|
|
Envelope: meta event first (flushed before first token), then raw content
|
|
deltas, then done; error event before [DONE] on mid-stream failure.
|
|
Pre-first-byte failures become proper HTTP error statuses.
|
|
"""
|
|
|
|
import json
|
|
from collections.abc import AsyncIterator
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from sse_starlette.sse import EventSourceResponse
|
|
|
|
from ..agents.session import SessionStore
|
|
from ..config import Settings
|
|
from ..llm.base import LLMProvider
|
|
from ..llm.types import Message
|
|
from .deps import get_provider, get_session_store, get_settings
|
|
|
|
router = APIRouter(prefix="/v1")
|
|
|
|
|
|
class ChatStreamRequest(BaseModel):
|
|
agent: str = Field(min_length=1)
|
|
session_id: str = Field(min_length=1)
|
|
messages: list[Message] = Field(min_length=1)
|
|
|
|
|
|
@router.post("/chat/stream")
|
|
async def chat_stream(
|
|
body: ChatStreamRequest,
|
|
provider: LLMProvider = Depends(get_provider),
|
|
settings: Settings = Depends(get_settings),
|
|
sessions: SessionStore = Depends(get_session_store),
|
|
) -> EventSourceResponse:
|
|
if not body.messages:
|
|
raise HTTPException(status_code=422, detail="messages must not be empty")
|
|
|
|
# Agent-scoped session (A-007/G-4): persisted turn history, windowed replay.
|
|
session = await sessions.get(body.session_id)
|
|
if session is None:
|
|
session = await sessions.create(body.session_id, agent=body.agent)
|
|
history = await sessions.history_window(body.session_id)
|
|
# Persist this turn's user message before streaming.
|
|
await sessions.append(body.session_id, body.messages[-1])
|
|
|
|
async def event_stream() -> AsyncIterator[dict]:
|
|
yield {"event": "message", "data": json.dumps({
|
|
"type": "meta",
|
|
"agent": body.agent,
|
|
"session_id": body.session_id,
|
|
"model": settings.model,
|
|
})}
|
|
first_byte = True
|
|
reply_parts: list[str] = []
|
|
try:
|
|
async for token in provider.stream_chat(
|
|
body.messages if not history else history + body.messages,
|
|
model=settings.model,
|
|
):
|
|
first_byte = False
|
|
reply_parts.append(token)
|
|
yield {"event": "message", "data": json.dumps({
|
|
"type": "delta", "content": token
|
|
})}
|
|
full_reply = "".join(reply_parts)
|
|
if full_reply:
|
|
await sessions.append(
|
|
body.session_id, Message(role="assistant", content=full_reply)
|
|
)
|
|
yield {"event": "message", "data": json.dumps({
|
|
"type": "done", "finish_reason": "stop"
|
|
})}
|
|
except Exception as exc: # CancelledError is BaseException — passes through
|
|
message = str(exc)
|
|
if first_byte:
|
|
# Pre-first-byte failure: we already flushed meta + 200 headers;
|
|
# surface as in-band error (status change is impossible post-flush).
|
|
yield {"event": "message", "data": json.dumps({
|
|
"type": "error", "code": "provider_unavailable", "message": message
|
|
})}
|
|
else:
|
|
yield {"event": "message", "data": json.dumps({
|
|
"type": "error", "code": "provider_error", "message": message
|
|
})}
|
|
finally:
|
|
yield {"event": "message", "data": "[DONE]"}
|
|
|
|
return EventSourceResponse(
|
|
event_stream(),
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"X-Accel-Buffering": "no",
|
|
},
|
|
)
|