diff --git a/app/api/routes.py b/app/api/routes.py index b6fb171..44a7053 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -8,9 +8,14 @@ disclosed — the dashboard polls every 2–3 s), `ack_alert` (in `repo.ack_aler `app/api/fixtures.py` is kept for tests and as the demo cabinet reference (doctor id). """ +import asyncio +import json +import logging import secrets +from typing import Any from fastapi import APIRouter, HTTPException +from fastapi.responses import StreamingResponse from sqlalchemy import select from app.api import demo as console @@ -42,10 +47,12 @@ from app.db.identity import Patient from app.db.session import get_clinical_session, get_identity_session from app.dialog.contracts import Event, get_engine from app.integrations.base import integrations +from app.llm.agent import stream as agent_stream from app.scheduler import jobs as scheduler from app.services.anonymous_care import display_name as anonymous_name from app.services.monitoring import refresh_status +log = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1") router.include_router(max_notifications.router) router.include_router(questionnaire_router) @@ -219,6 +226,49 @@ async def dialog_event(token: str, body: DialogIn) -> DialogOut: return DialogOut(replies=await get_engine().handle(token, body)) +@router.post("/dialog/{token}/events/stream", tags=["dialog"]) +async def dialog_event_stream(token: str, body: DialogIn) -> StreamingResponse: + """Same as /events, as Server-Sent Events: `status` (agent steps), `delta` (answer as it is written), + `reset`, then `done` with the final replies (identical to /events). `error` if the engine failed.""" + await _require_patient(token) + queue: asyncio.Queue[tuple[str, Any]] = asyncio.Queue() + marker = agent_stream.STREAM.set(queue) # copied into the task's context below + try: + + async def run() -> None: + try: + replies = await get_engine().handle(token, body) + queue.put_nowait(("done", [r.model_dump() for r in replies])) + except Exception: # noqa: BLE001 — reported to the client, which falls back to /events + log.exception("dialog stream failed") + queue.put_nowait(("error", None)) + + task = asyncio.create_task(run()) + finally: + agent_stream.STREAM.reset(marker) + + async def events(): + try: + while True: + try: + kind, payload = await asyncio.wait_for(queue.get(), timeout=15) + except TimeoutError: + yield ": keepalive\n\n" + continue + yield f"event: {kind}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n" + if kind in ("done", "error"): + return + finally: + if not task.done(): + await task + + return StreamingResponse( + events(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, # nginx: do not buffer SSE + ) + + @router.get("/dialog/{token}/outbox", response_model=DialogOut, tags=["dialog"]) async def dialog_outbox(token: str) -> DialogOut: """System-initiated replies (reminders, check-ins) waiting for the web console.""" diff --git a/app/llm/agent/__init__.py b/app/llm/agent/__init__.py index 379b252..5f25795 100644 --- a/app/llm/agent/__init__.py +++ b/app/llm/agent/__init__.py @@ -21,6 +21,7 @@ from pydantic import BaseModel from app.config import Settings, get_settings from app.core.types import ScrubbedText +from app.llm.agent import stream from app.llm.agent.prompts import SYSTEM_PROMPT from app.llm.agent.providers import ( GigaChatToolProvider, @@ -127,6 +128,7 @@ async def answer_chat( return None if is_small_talk(question): return AgentAnswer(text=SMALL_TALK_REPLY, sources=[], tools=[], provider="small_talk") + stream.emit("status", "Думаю над вопросом…") ctx = ToolContext(token=token) messages: list[dict[str, Any]] = [ {"role": "system", "content": SYSTEM_PROMPT + await _patient_brief(ctx)}, @@ -159,6 +161,7 @@ async def answer_chat( } ) for c in turn.tool_calls[:MAX_TOOL_CALLS_PER_TURN]: + stream.emit("status", stream.TOOL_STATUS.get(c.name, "Проверяю данные…")) result = await run_tool(ctx, c.name, c.arguments) messages.append({"role": "tool", "tool_call_id": c.id, "name": c.name, "content": result}) log.warning("chat agent: step limit reached tools=%s", ",".join(ctx.used)) diff --git a/app/llm/agent/providers.py b/app/llm/agent/providers.py index 8bd5852..63d4f01 100644 --- a/app/llm/agent/providers.py +++ b/app/llm/agent/providers.py @@ -16,6 +16,7 @@ import httpx from pydantic import BaseModel, Field from app.config import Settings +from app.llm.agent import stream log = logging.getLogger(__name__) @@ -53,6 +54,8 @@ class OpenRouterProvider: self._client = client async def chat(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> AssistantTurn: + if stream.active(): + return await self._chat_stream(messages, tools) body: dict[str, Any] = {"model": self._model, "messages": messages, "tools": tools, "temperature": 0.1} body["max_tokens"] = 900 if not any(m["role"] == "tool" for m in messages): @@ -174,6 +177,78 @@ class GigaChatToolProvider: return AssistantTurn(content=msg.content or None, tool_calls=calls) +# ------------------------------------------------------------------------------------------------ streaming + + +async def _iter_sse(resp: httpx.Response): + async for line in resp.aiter_lines(): + if not line.startswith("data:"): + continue # comments («: OPENROUTER PROCESSING») and blank lines + data = line[5:].strip() + if data == "[DONE]": + return + try: + yield json.loads(data) + except json.JSONDecodeError: + continue + + +async def _openrouter_stream(self: "OpenRouterProvider", messages: list[dict[str, Any]], tools: list[dict[str, Any]]): + """Same request with `stream: true`: content deltas go to the live channel, tool calls are assembled.""" + body: dict[str, Any] = { + "model": self._model, + "messages": messages, + "tools": tools, + "temperature": 0.1, + "max_tokens": 900, + "stream": True, + } + if not any(m["role"] == "tool" for m in messages): + body["tool_choice"] = "required" + headers = {"Authorization": f"Bearer {self._key}", "X-Title": "PTSRaiAgent prototype (synthetic data)"} + content: list[str] = [] + calls: dict[int, dict[str, Any]] = {} + try: + client = self._client or httpx.AsyncClient(timeout=self._timeout, proxy=self._proxy) + try: + async with client.stream("POST", self._url, json=body, headers=headers, timeout=self._timeout) as resp: + if resp.status_code != 200: + await resp.aread() + raise ProviderError(f"http {resp.status_code}") + async for chunk in _iter_sse(resp): + choice = (chunk.get("choices") or [{}])[0] + delta = choice.get("delta") or {} + if delta.get("content"): + content.append(delta["content"]) + stream.emit("delta", delta["content"]) + for tc in delta.get("tool_calls") or []: + slot = calls.setdefault(tc.get("index", 0), {"id": "", "name": "", "args": ""}) + slot["id"] = tc.get("id") or slot["id"] + fn = tc.get("function") or {} + slot["name"] += fn.get("name") or "" + slot["args"] += fn.get("arguments") or "" + finally: + if self._client is None: + await client.aclose() + except httpx.HTTPError as exc: + raise ProviderError(type(exc).__name__) from exc + out = [] + for i in sorted(calls): + c = calls[i] + try: + args = json.loads(c["args"] or "{}") + except json.JSONDecodeError: + args = {} + out.append(ToolCall(id=c["id"] or c["name"] or f"call{i}", name=c["name"], arguments=args)) + if out and content: + stream.emit("reset", None) # a preface before tool calls is not the answer + log.info("openrouter stream ok calls=%d chars=%d", len(out), sum(map(len, content))) + return AssistantTurn(content="".join(content) or None, tool_calls=out) + + +OpenRouterProvider._chat_stream = _openrouter_stream # type: ignore[attr-defined] + + # ------------------------------------------------------------------------------------------------ offline mock _DRUG_WORDS = re.compile( diff --git a/app/llm/agent/stream.py b/app/llm/agent/stream.py new file mode 100644 index 0000000..bf1bbd2 --- /dev/null +++ b/app/llm/agent/stream.py @@ -0,0 +1,32 @@ +"""Live progress of the chat agent for streaming channels (web SSE). + +The SSE endpoint sets a queue in the context; the agent pushes events while it works: + status — a tool step in plain Russian («Смотрю вашу карту ЕМИАС…») + delta — a chunk of the answer as the model writes it + reset — the streamed text so far was a preface before tool calls: clear it +Channels without a queue (Telegram, tests) are unaffected. Deltas are the raw model text: the final reply +(forbidden-output filter, sources footer) always comes in the closing `done` event and replaces it. +""" + +import asyncio +from contextvars import ContextVar +from typing import Any + +STREAM: ContextVar[asyncio.Queue[tuple[str, Any]] | None] = ContextVar("agent_stream", default=None) + +TOOL_STATUS = { + "get_patient_record": "Смотрю вашу карту ЕМИАС…", + "check_drug_for_patient": "Проверяю препарат по вашей карте: болезни, непереносимости, другие лекарства…", + "get_drug_info": "Открываю инструкцию к препарату (ГРЛС)…", + "search_guidelines": "Ищу в клинических рекомендациях Минздрава…", +} + + +def active() -> bool: + return STREAM.get() is not None + + +def emit(kind: str, payload: Any) -> None: + q = STREAM.get() + if q is not None: + q.put_nowait((kind, payload)) diff --git a/tests/test_agent_stream.py b/tests/test_agent_stream.py new file mode 100644 index 0000000..8258873 --- /dev/null +++ b/tests/test_agent_stream.py @@ -0,0 +1,83 @@ +"""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) diff --git a/web/src/api/client.ts b/web/src/api/client.ts index a0cb402..ea6b5ad 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -59,3 +59,53 @@ export const api = { outbox: async (token: string) => (await request<{ replies: Reply[] }>(`/dialog/${encodeURIComponent(token)}/outbox`)).replies, }; + +export type StreamEvent = + | { kind: "status"; text: string } + | { kind: "delta"; text: string } + | { kind: "reset" } + | { kind: "done"; replies: Reply[] }; + +/** POST /dialog/{token}/events/stream (Server-Sent Events): agent steps and the answer as it is written, + * then the final replies. Resolves with the final replies; throws if the stream breaks (caller falls back). */ +export async function sendEventStream( + token: string, + event: DialogEvent, + onEvent: (e: StreamEvent) => void, +): Promise { + const res = await fetch(`${BASE}/dialog/${token}/events/stream`, { + method: "POST", + headers: { "Content-Type": "application/json", Accept: "text/event-stream" }, + body: JSON.stringify(event), + }); + if (!res.ok || !res.body) throw new ApiError(res.status, res.statusText); + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + let cut: number; + while ((cut = buf.indexOf("\n\n")) >= 0) { + const block = buf.slice(0, cut); + buf = buf.slice(cut + 2); + let kind = "message"; + let data = ""; + for (const line of block.split("\n")) { + if (line.startsWith("event:")) kind = line.slice(6).trim(); + else if (line.startsWith("data:")) data += line.slice(5).trim(); + } + if (!data) continue; + const payload = JSON.parse(data); + if (kind === "status") onEvent({ kind, text: String(payload) }); + else if (kind === "delta") onEvent({ kind, text: String(payload) }); + else if (kind === "reset") onEvent({ kind: "reset" }); + else if (kind === "done") { + onEvent({ kind: "done", replies: payload as Reply[] }); + return payload as Reply[]; + } else if (kind === "error") throw new ApiError(500, "stream error"); + } + } + throw new ApiError(0, "stream closed"); +} diff --git a/web/src/components/AppShell.tsx b/web/src/components/AppShell.tsx index a93786e..2b89534 100644 --- a/web/src/components/AppShell.tsx +++ b/web/src/components/AppShell.tsx @@ -16,6 +16,7 @@ import { Icon, LogoMark, type IconName } from "./Icon"; const TABS: { to: string; label: string; icon: IconName }[] = [ { to: "/today", label: ru.nav.today, icon: "home" }, + { to: "/messages", label: ru.nav.chat, icon: "message" }, { to: "/meds", label: ru.nav.meds, icon: "pill" }, { to: "/doctor", label: ru.nav.doctor, icon: "doctor" }, { to: "/profile", label: ru.nav.profile, icon: "user" }, diff --git a/web/src/content/ru.ts b/web/src/content/ru.ts index f5288dc..b6f2aff 100644 --- a/web/src/content/ru.ts +++ b/web/src/content/ru.ts @@ -16,6 +16,7 @@ export const ru = { doctor: "Врач", profile: "Профиль", messages: "Сообщения", + chat: "Чат", signIn: "Вход", help: "Помощь", }, @@ -251,6 +252,7 @@ export const ru = { noticeTitle: "Опрос приостановлен", }, messages: { + typing: "Помощник печатает", title: "Сообщения", inputLabel: "Сообщение", inputPlaceholder: "Напишите сообщение", diff --git a/web/src/dialog/DialogContext.tsx b/web/src/dialog/DialogContext.tsx index 13d4793..0503642 100644 --- a/web/src/dialog/DialogContext.tsx +++ b/web/src/dialog/DialogContext.tsx @@ -1,6 +1,6 @@ import { createContext, useCallback, useContext, useEffect, useMemo, useRef, useState, type ReactNode } from "react"; -import { api } from "../api/client"; +import { api, sendEventStream } from "../api/client"; import type { DialogEvent, Reply } from "../api/types"; import { useSession } from "../session/SessionContext"; import { sessionStore, type ChatEntry, type DialogSnapshot } from "../session/store"; @@ -17,8 +17,15 @@ let seq = 0; const newId = () => `${Date.now().toString(36)}-${(seq++).toString(36)}`; const asEntries = (replies: Reply[]): ChatEntry[] => replies.map((reply) => ({ id: newId(), from: "assistant", reply })); +/** The agent at work: its steps and the answer as it is being written (cleared when the final reply lands). */ +export interface LiveReply { + status: string[]; + text: string; +} + interface DialogApi extends DialogSnapshot { busy: boolean; + live: LiveReply | null; error: boolean; started: boolean; /** Sends an event; `echo` is what the patient «said» (button label / text) for the history. */ @@ -36,6 +43,7 @@ export function DialogProvider({ children }: { children: ReactNode }) { const [owner, setOwner] = useState(token); const [snap, setSnap] = useState(() => (token ? sessionStore.loadDialog(token) : EMPTY)); const [busy, setBusy] = useState(false); + const [live, setLive] = useState(null); const [error, setError] = useState(false); const tokenRef = useRef(token); tokenRef.current = token; @@ -58,7 +66,35 @@ export function DialogProvider({ children }: { children: ReactNode }) { setBusy(true); setError(false); try { - const replies = await api.sendEvent(t, event); + let replies: Reply[]; + if (event.type === "text") { + // Free text may go to the agent: stream its steps and the answer so the viewer sees it working. + const said = echo; + if (said) { + // Show the patient's message right away, while the agent is still working. + setSnap((prev) => ({ + ...prev, + entries: [...prev.entries, { id: newId(), from: "patient" as const, text: said }], + })); + echo = undefined; + } + setLive({ status: [], text: "" }); + try { + replies = await sendEventStream(t, event, (e) => + setLive((prev) => { + const cur = prev ?? { status: [], text: "" }; + if (e.kind === "status") return { ...cur, status: [...cur.status, e.text] }; + if (e.kind === "delta") return { ...cur, text: cur.text + e.text }; + if (e.kind === "reset") return { ...cur, text: "" }; + return cur; + }), + ); + } catch { + replies = await api.sendEvent(t, event); // proxies without SSE support: plain request + } + } else { + replies = await api.sendEvent(t, event); + } setSnap((prev) => ({ entries: [ ...prev.entries, @@ -73,6 +109,7 @@ export function DialogProvider({ children }: { children: ReactNode }) { setError(true); return false; } finally { + setLive(null); setBusy(false); } }, @@ -102,8 +139,8 @@ export function DialogProvider({ children }: { children: ReactNode }) { }, [token]); const value = useMemo( - () => ({ ...snap, busy, error, started: snap.entries.length > 0, send, start, markRead }), - [snap, busy, error, send, start, markRead], + () => ({ ...snap, busy, live, error, started: snap.entries.length > 0, send, start, markRead }), + [snap, busy, live, error, send, start, markRead], ); return {children}; } diff --git a/web/src/screens/Messages.tsx b/web/src/screens/Messages.tsx index 3d9f73f..8691f41 100644 --- a/web/src/screens/Messages.tsx +++ b/web/src/screens/Messages.tsx @@ -18,7 +18,7 @@ export function Messages() { useEffect(() => { dialog.markRead(); endRef.current?.scrollIntoView({ block: "end" }); - }, [dialog.entries.length]); + }, [dialog.entries.length, dialog.live?.text, dialog.live?.status.length]); const send = async (e?: FormEvent) => { e?.preventDefault(); @@ -56,6 +56,32 @@ export function Messages() { ), )} + {dialog.live && ( +
  • + + + +
    + {dialog.live.status.map((st, i) => ( +

    + {st} +

    + ))} + {dialog.live.text ? ( +

    + {dialog.live.text} +

    + ) : ( +

    + + + +

    + )} +
    +
    • + )} {quick.length > 0 && (
    diff --git a/web/src/theme/global.css b/web/src/theme/global.css index 672710b..30fce66 100644 --- a/web/src/theme/global.css +++ b/web/src/theme/global.css @@ -103,7 +103,7 @@ button:disabled { inset: auto 0 0 0; z-index: 20; display: grid; - grid-template-columns: repeat(4, 1fr); + grid-template-columns: repeat(5, 1fr); height: calc(var(--nav-h) + var(--safe-bottom)); padding-bottom: var(--safe-bottom); background: var(--c-surface); @@ -2937,3 +2937,19 @@ button.row-card:hover:not(:disabled) { animation: none; } } + + +/* Live agent reply: steps + the answer as it is written (SSE) */ +.bubble--live .live { display: grid; gap: 4px; } +.live__step { margin: 0; font-size: 13px; color: var(--muted, #6b7280); } +.live__step.is-active::after { content: ""; display: inline-block; width: 1em; animation: live-ellipsis 1.2s steps(4) infinite; overflow: hidden; vertical-align: bottom; } +.live__text { margin: 2px 0 0; white-space: pre-wrap; } +.live__cursor { display: inline-block; width: 7px; height: 1em; margin-left: 2px; background: currentColor; vertical-align: text-bottom; animation: live-blink 1s steps(2) infinite; } +.live__dots { display: inline-flex; gap: 4px; margin: 4px 0 0; } +.live__dots span { width: 6px; height: 6px; border-radius: 50%; background: var(--muted, #9ca3af); animation: live-dot 1.2s infinite ease-in-out; } +.live__dots span:nth-child(2) { animation-delay: .15s; } +.live__dots span:nth-child(3) { animation-delay: .3s; } +@keyframes live-blink { 50% { opacity: 0; } } +@keyframes live-dot { 0%, 80%, 100% { opacity: .3; transform: translateY(0); } 40% { opacity: 1; transform: translateY(-3px); } } +@keyframes live-ellipsis { from { width: 0; } to { width: 1em; } } +.tabbar__item span { font-size: 11px; }