ElebobaandClaude Opus 5.5
d1f92407f9
CI / ruff · pytest (pull_request) Successful in 2m43s
- POST /dialog/{token}/events/stream: Server-Sent Events with the agent's steps («Смотрю вашу карту ЕМИАС…»,
«Ищу в клинических рекомендациях…»), the answer as the model writes it (OpenRouter stream=true), then the
final replies (same as /events, incl. the forbidden-output filter and sources). X-Accel-Buffering: no.
- Web: free-text messages use the stream — a live bubble with steps, typing dots and a cursor; falls back to
the plain request if the stream breaks.
- Phones: the bottom tab bar gets «Чат» (it had no way back to the chat).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
"""Live agent output: OpenRouter stream parsing and the SSE endpoint (status → delta → done)."""
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.config import Settings
|
|
from app.db import session
|
|
from app.db.bootstrap import init_db
|
|
from app.llm import agent
|
|
from app.llm.agent import stream
|
|
from app.llm.agent.providers import OpenRouterProvider
|
|
from app.main import app
|
|
|
|
MEM = "sqlite+aiosqlite:///:memory:"
|
|
|
|
|
|
def _sse(*chunks: dict) -> bytes:
|
|
lines = [": OPENROUTER PROCESSING", ""]
|
|
lines += [f"data: {json.dumps(c, ensure_ascii=False)}" for c in chunks] + ["data: [DONE]", ""]
|
|
return "\n\n".join(lines).encode()
|
|
|
|
|
|
async def test_openrouter_stream_deltas_and_tool_calls():
|
|
body = _sse(
|
|
{"choices": [{"delta": {"content": "Сейчас "}}]},
|
|
{
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"tool_calls": [{"index": 0, "id": "c1", "function": {"name": "get_patient_", "arguments": ""}}]
|
|
}
|
|
}
|
|
]
|
|
},
|
|
{"choices": [{"delta": {"tool_calls": [{"index": 0, "function": {"name": "record", "arguments": "{}"}}]}}]},
|
|
)
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200, content=body)))
|
|
provider = OpenRouterProvider(Settings(_env_file=None, openrouter_api_key="k"), client=client)
|
|
q: asyncio.Queue = asyncio.Queue()
|
|
token = stream.STREAM.set(q)
|
|
try:
|
|
turn = await provider.chat([{"role": "user", "content": "x"}], [])
|
|
finally:
|
|
stream.STREAM.reset(token)
|
|
assert [c.name for c in turn.tool_calls] == ["get_patient_record"]
|
|
events = [q.get_nowait() for _ in range(q.qsize())]
|
|
assert ("delta", "Сейчас ") in events and ("reset", None) in events
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
await session.configure(identity_url=MEM, clinical_url=MEM)
|
|
await init_db()
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as c:
|
|
yield c
|
|
await session.configure()
|
|
|
|
|
|
async def test_sse_endpoint_streams_steps_then_final(client, monkeypatch):
|
|
async def fake(token, text, **kw):
|
|
stream.emit("status", stream.TOOL_STATUS["get_patient_record"])
|
|
stream.emit("delta", "Ответ ")
|
|
stream.emit("delta", "агента")
|
|
return agent.AgentAnswer(text="Ответ агента", sources=[], tools=["get_patient_record"], provider="fake")
|
|
|
|
monkeypatch.setattr(agent, "answer_chat", fake)
|
|
token = (await client.post("/api/v1/auth/demo/fresh", json={})).json()["token"]
|
|
kinds = []
|
|
async with client.stream(
|
|
"POST", f"/api/v1/dialog/{token}/events/stream", json={"type": "text", "text": "Можно ибупрофен?"}
|
|
) as resp:
|
|
assert resp.headers["content-type"].startswith("text/event-stream")
|
|
async for line in resp.aiter_lines():
|
|
if line.startswith("event:"):
|
|
kinds.append(line[6:].strip())
|
|
if line.startswith("data:") and kinds and kinds[-1] == "done":
|
|
final = json.loads(line[5:])
|
|
assert kinds[0] == "status" and "delta" in kinds and kinds[-1] == "done"
|
|
assert any("Ответ агента" in r["text"] for r in final)
|