From 58b92da8b0e2d915e7de1bcfceceae8a6db39ac5 Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 15 Oct 2026 17:37:56 +0200 Subject: [PATCH] Assistant streams its answer token by token ai/client.py streams the completion and puts the chunks back together (Assembler), covering OpenAI's piecewise tool-call arguments, Ollama's whole calls, OpenRouter's comment lines, unnumbered calls and in-stream errors. The turn emits a delta event per piece and a message per round as before, and the chat grows the last answer as deltas arrive. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx --- backend/fluksio/ai/agent.py | 16 ++- backend/fluksio/ai/client.py | 100 +++++++++++++++-- backend/fluksio/api/routes/ai.py | 5 +- backend/tests/test_ai.py | 124 ++++++++++++++++++++- docs/code/agents.md | 3 +- frontend/src/client/sdk.gen.ts | 5 +- frontend/src/components/Chat/ChatPanel.tsx | 21 +++- frontend/src/components/Chat/stream.ts | 1 + 8 files changed, 243 insertions(+), 32 deletions(-) diff --git a/backend/fluksio/ai/agent.py b/backend/fluksio/ai/agent.py index df12bc0..14ad091 100644 --- a/backend/fluksio/ai/agent.py +++ b/backend/fluksio/ai/agent.py @@ -119,8 +119,10 @@ async def run_turn( ) -> AsyncIterator[dict[str, Any]]: """One user turn, yielding progress events as it goes. - A round is appended to *messages* only once every tool in it has answered, - so a turn cut off part-way leaves a transcript the provider still accepts. + A round's text arrives as ``delta`` events while the model writes it, and + once more whole as ``message`` when the round is done. A round is appended + to *messages* only once every tool in it has answered, so a turn cut off + part-way leaves a transcript the provider still accepts. """ messages.append({"role": "user", "content": text}) tools = await tool_specs() @@ -128,18 +130,22 @@ async def run_turn( token = caller_authorization.set(authorization) try: for _ in range(config.max_rounds): + choice: dict[str, Any] = {} try: - response = await client.chat_completion( + async for part in client.stream_completion( api_base=config.api_base, api_key=config.api_key, model=config.model, messages=[system, *messages], tools=tools, - ) + ): + if isinstance(part, str): + yield {"type": "delta", "content": part} + else: + choice = part except (client.ProviderError, httpx.HTTPError) as exc: yield {"type": "error", "detail": f"The provider answered {exc}"} return - choice = response["choices"][0]["message"] content = choice.get("content") or "" calls = choice.get("tool_calls") or [] assistant: dict[str, Any] = {"role": "assistant", "content": content} diff --git a/backend/fluksio/ai/client.py b/backend/fluksio/ai/client.py index fd19e75..d9607a6 100644 --- a/backend/fluksio/ai/client.py +++ b/backend/fluksio/ai/client.py @@ -4,10 +4,14 @@ Chat completions with function tools is the dialect every hosted provider and every local server speaks — OpenAI, Anthropic's compatibility endpoint, OpenRouter, Ollama, vLLM — so one client covers them without a provider SDK. The admin types the URL; the model never chooses where a request goes. + +The call is streamed, so the answer reaches the browser as it is written. """ from __future__ import annotations +import json +from collections.abc import AsyncIterator from typing import Any from urllib.parse import urlparse @@ -39,7 +43,69 @@ def _detail(response: httpx.Response) -> str: return f"{response.status_code}: {text[:300]}" -async def chat_completion( +class Assembler: + """The streamed chunks put back together into the message a plain call returns. + + The providers stream the same ``data:`` lines and differ only in the + margins: OpenRouter interleaves ``:`` comment lines, a thinking model's + reasoning arrives in a field of its own and is left out, OpenAI and + Anthropic send a tool call's arguments in pieces where Ollama sends the call + whole, and some leave out the call's ``index``. + """ + + def __init__(self) -> None: + self.content = "" + self.calls: list[dict[str, Any]] = [] + + def feed(self, line: str) -> str: + """One line of the stream; returns the text it adds to the answer.""" + if not line.startswith("data:"): + return "" + data = line[5:].strip() + if not data or data == "[DONE]": + return "" + try: + chunk = json.loads(data) + except ValueError as exc: + raise ProviderError(f"sent {data[:300]}") from exc + error = chunk.get("error") + if error: + if isinstance(error, dict): + error = error.get("message", error) + raise ProviderError(str(error)[:300]) + choices = chunk.get("choices") or [{}] + delta = choices[0].get("delta") or {} + for piece in delta.get("tool_calls") or []: + self._merge(piece) + text: str = delta.get("content") or "" + self.content += text + return text + + def _merge(self, piece: dict[str, Any]) -> None: + index = piece.get("index", len(self.calls)) + while len(self.calls) <= index: + self.calls.append( + { + "id": "", + "type": "function", + "function": {"name": "", "arguments": ""}, + } + ) + call = self.calls[index] + call["id"] = piece.get("id") or call["id"] + function = piece.get("function") or {} + call["function"]["name"] = function.get("name") or call["function"]["name"] + call["function"]["arguments"] += function.get("arguments") or "" + + def message(self) -> dict[str, Any]: + return { + "role": "assistant", + "content": self.content or None, + "tool_calls": self.calls or None, + } + + +async def stream_completion( *, api_base: str, api_key: str, @@ -47,21 +113,33 @@ async def chat_completion( messages: list[dict[str, Any]], tools: list[dict[str, Any]], timeout: float = 120.0, -) -> dict[str, Any]: - """``POST {api_base}/chat/completions``, the whole answer at once.""" - body: dict[str, Any] = {"model": model, "messages": messages} +) -> AsyncIterator[str | dict[str, Any]]: + """``POST {api_base}/chat/completions`` streamed. + + Yields each piece of the answer's text as it arrives, then the whole + message. The timeout is per chunk, so a long answer is not cut off. + """ + body: dict[str, Any] = {"model": model, "messages": messages, "stream": True} if tools: body["tools"] = tools - async with httpx.AsyncClient(timeout=timeout) as http: - response = await http.post( + assembler = Assembler() + async with ( + httpx.AsyncClient(timeout=timeout) as http, + http.stream( + "POST", api_base.rstrip("/") + "/chat/completions", json=body, headers=_headers(api_key), - ) - if response.is_error: - raise ProviderError(_detail(response)) - result: dict[str, Any] = response.json() - return result + ) as response, + ): + if response.is_error: + await response.aread() + raise ProviderError(_detail(response)) + async for line in response.aiter_lines(): + text = assembler.feed(line) + if text: + yield text + yield assembler.message() async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> list[str]: diff --git a/backend/fluksio/api/routes/ai.py b/backend/fluksio/api/routes/ai.py index 8c93fd7..0cae07b 100644 --- a/backend/fluksio/api/routes/ai.py +++ b/backend/fluksio/api/routes/ai.py @@ -259,8 +259,9 @@ def chat( ) -> StreamingResponse: """One turn, streamed as server-sent events. - ``message``, ``tool_call`` and ``tool_result`` events as the model works, - ``error`` when something stops it, and ``done`` last either way. + ``delta`` as the model writes, ``message`` with a round's whole text, + ``tool_call`` and ``tool_result`` as it works, ``error`` when something + stops it, and ``done`` last either way. """ config = ai_config.load() if config is None or not config.ready: diff --git a/backend/tests/test_ai.py b/backend/tests/test_ai.py index 430df41..a13f63e 100644 --- a/backend/tests/test_ai.py +++ b/backend/tests/test_ai.py @@ -1,14 +1,14 @@ """The assistant: who may set it up, and that it acts as the person chatting.""" import json -from collections.abc import Iterator +from collections.abc import AsyncIterator, Iterator from pathlib import Path from typing import Any import pytest from fastapi.testclient import TestClient -from fluksio.ai import agent +from fluksio.ai import agent, client from fluksio.core.config import settings @@ -94,12 +94,16 @@ def test_a_turn_runs_the_tools_as_the_person_chatting( ] ) - async def fake_completion(**kwargs: Any) -> dict[str, Any]: + async def fake_completion(**kwargs: Any) -> AsyncIterator[str | dict[str, Any]]: seen.append(kwargs["messages"]) assert any(t["function"]["name"] == "save_flow" for t in kwargs["tools"]) - return {"choices": [{"message": next(answers)}]} + answer = next(answers) + if answer["content"]: + yield "Counted " + yield "them." + yield answer - monkeypatch.setattr(agent.client, "chat_completion", fake_completion) + monkeypatch.setattr(agent.client, "stream_completion", fake_completion) with client.stream( "POST", @@ -114,6 +118,8 @@ def test_a_turn_runs_the_tools_as_the_person_chatting( "start", "tool_call", "tool_result", + "delta", + "delta", "message", "done", ] @@ -136,3 +142,111 @@ def test_a_turn_runs_the_tools_as_the_person_chatting( } client.delete("/api/v1/ai/chat", headers=superuser_token_headers) assert client.get("/api/v1/ai/chat", headers=superuser_token_headers).json() == [] + + +# One answer per provider as it streams it. Ollama's text is recorded from a +# local server, the rest follows the shape each endpoint documents. +STREAMS = { + "ollama": """ +data: {"id":"chatcmpl-888","object":"chat.completion.chunk","model":"gemma3:4b-it-qat","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":"Hello"},"finish_reason":null}]} + +data: {"id":"chatcmpl-888","object":"chat.completion.chunk","model":"gemma3:4b-it-qat","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":" there"},"finish_reason":null}]} + +data: {"id":"chatcmpl-888","object":"chat.completion.chunk","model":"gemma3:4b-it-qat","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":"stop"}]} + +data: [DONE] +""", + "ollama-tools": """ +data: {"id":"chatcmpl-12","object":"chat.completion.chunk","model":"qwen3","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning":"The user wants"},"finish_reason":null}]} + +data: {"id":"chatcmpl-12","object":"chat.completion.chunk","model":"qwen3","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":"","tool_calls":[{"id":"call_k2","index":0,"type":"function","function":{"name":"get_flow","arguments":"{\\"name\\":\\"demo\\"}"}}]},"finish_reason":null}]} + +data: {"id":"chatcmpl-12","object":"chat.completion.chunk","model":"qwen3","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":"tool_calls"}]} + +data: [DONE] +""", + "openai": """ +data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"role":"assistant","content":null,"tool_calls":[{"index":0,"id":"call_a","type":"function","function":{"name":"get_flow","arguments":""}}],"refusal":null},"finish_reason":null}]} + +data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"name\\""}}]},"finish_reason":null}]} + +data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":1,"id":"call_b","type":"function","function":{"name":"list_flows","arguments":""}}]},"finish_reason":null}]} + +data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":":\\"demo\\"}"}}]},"finish_reason":null}]} + +data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":1,"function":{"arguments":"{}"}}]},"finish_reason":null}]} + +data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]} + +data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[],"usage":{"prompt_tokens":9,"completion_tokens":12,"total_tokens":21}} + +data: [DONE] +""", + "anthropic": """ +data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"role":"assistant","content":"Reading "},"finish_reason":null}]} + +data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"content":"it."},"finish_reason":null}]} + +data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"toolu_01","type":"function","function":{"name":"get_flow","arguments":""}}]},"finish_reason":null}]} + +data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"name\\": \\"de"}}]},"finish_reason":null}]} + +data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"mo\\"}"}}]},"finish_reason":null}]} + +data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]} + +data: [DONE] +""", + "openrouter": """ +: OPENROUTER PROCESSING + +data: {"id":"gen-1","provider":"Google","model":"google/gemini-2.5-flash","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning":"Short answer."},"finish_reason":null}]} + +: OPENROUTER PROCESSING + +data: {"id":"gen-1","provider":"Google","model":"google/gemini-2.5-flash","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"Hello"},"finish_reason":null}]} + +data: {"id":"gen-1","provider":"Google","model":"google/gemini-2.5-flash","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":" there"},"finish_reason":"stop"}],"usage":{"prompt_tokens":4,"completion_tokens":2,"total_tokens":6}} + +data: [DONE] +""", + # Gemini's compatibility endpoint sends each call whole and unnumbered. + "gemini": """ +data: {"choices":[{"delta":{"role":"assistant","tool_calls":[{"id":"c1","type":"function","function":{"name":"get_flow","arguments":"{\\"name\\":\\"demo\\"}"}},{"id":"c2","type":"function","function":{"name":"list_flows","arguments":"{}"}}]},"finish_reason":"tool_calls","index":0}],"object":"chat.completion.chunk"} + +data: [DONE] +""", +} + +GET_DEMO = {"name": "get_flow", "arguments": '{"name":"demo"}'} +LIST = {"name": "list_flows", "arguments": "{}"} + + +@pytest.mark.parametrize( + ("provider", "content", "calls"), + [ + ("ollama", "Hello there", []), + ("ollama-tools", None, [GET_DEMO]), + ("openai", None, [GET_DEMO, LIST]), + ("anthropic", "Reading it.", [{**GET_DEMO, "arguments": '{"name": "demo"}'}]), + ("openrouter", "Hello there", []), + ("gemini", None, [GET_DEMO, LIST]), + ], +) +def test_a_streamed_answer_comes_back_whole( + provider: str, content: str | None, calls: list[dict[str, str]] +) -> None: + assembler = client.Assembler() + pieces = [assembler.feed(line) for line in STREAMS[provider].splitlines()] + message = assembler.message() + assert "".join(pieces) == (content or "") + assert message["content"] == content + assert [c["function"] for c in message["tool_calls"] or []] == calls + assert all(c["id"] and c["type"] == "function" for c in message["tool_calls"] or []) + + +def test_an_error_inside_the_stream_stops_it() -> None: + assembler = client.Assembler() + line = 'data: {"error":{"code":502,"message":"Provider disconnected"},"choices":[{"index":0,"delta":{"content":""},"finish_reason":"error"}]}' + with pytest.raises(client.ProviderError, match="Provider disconnected"): + assembler.feed(line) diff --git a/docs/code/agents.md b/docs/code/agents.md index 0479858..363299c 100644 --- a/docs/code/agents.md +++ b/docs/code/agents.md @@ -54,7 +54,8 @@ A superuser configures an OpenAI-compatible provider under the dashboard's Assistant entry for signed-in users. The assistant uses the same tools with your own permissions. Draft edits -appear in the editor, and each answer renders as markdown with a tool trace. +appear in the editor. Each answer streams in as the model writes it and +renders as markdown with a tool trace. Conversations are stored per user on the instance; the bin button clears yours. `MCP_ENABLED` is only needed for external clients, so the built-in assistant diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index bbf17fa..8e727a1 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -115,8 +115,9 @@ export class AiService { * Chat * One turn, streamed as server-sent events. * - * ``message``, ``tool_call`` and ``tool_result`` events as the model works, - * ``error`` when something stops it, and ``done`` last either way. + * ``delta`` as the model writes, ``message`` with a round's whole text, + * ``tool_call`` and ``tool_result`` as it works, ``error`` when something + * stops it, and ``done`` last either way. * @param data The data for the request. * @param data.requestBody * @returns unknown Successful Response diff --git a/frontend/src/components/Chat/ChatPanel.tsx b/frontend/src/components/Chat/ChatPanel.tsx index b2b0821..63ef012 100644 --- a/frontend/src/components/Chat/ChatPanel.tsx +++ b/frontend/src/components/Chat/ChatPanel.tsx @@ -175,10 +175,12 @@ export function ChatPanel() { .finally(() => setLoaded(true)) }, [open, loaded, available]) - // biome-ignore lint/correctness/useExhaustiveDependencies: scroll when the list grows, not on every render. + const tail = shown[shown.length - 1] + const tailLength = tail && "text" in tail ? tail.text.length : 0 + // biome-ignore lint/correctness/useExhaustiveDependencies: scroll when the list or a streamed answer grows, not on every render. useEffect(() => { bottom.current?.scrollIntoView({ block: "end" }) - }, [shown.length, waiting]) + }, [shown.length, tailLength, waiting]) const send = async () => { const text = draft.trim() @@ -190,11 +192,18 @@ export function ChatPanel() { controller.current = abort const onEvent = (event: ChatEvent) => { switch (event.type) { + // A round's text grows in the last item and is settled by `message`; + // a tool call or the next turn always comes between two rounds. + case "delta": case "message": - setItems((prev) => [ - ...prev, - { kind: "assistant", text: event.content }, - ]) + setItems((prev) => { + const last = prev[prev.length - 1] + if (last?.kind !== "assistant") + return [...prev, { kind: "assistant", text: event.content }] + const text = + event.type === "delta" ? last.text + event.content : event.content + return [...prev.slice(0, -1), { kind: "assistant", text }] + }) break case "tool_call": setItems((prev) => [ diff --git a/frontend/src/components/Chat/stream.ts b/frontend/src/components/Chat/stream.ts index bc8b276..6a0eea1 100644 --- a/frontend/src/components/Chat/stream.ts +++ b/frontend/src/components/Chat/stream.ts @@ -3,6 +3,7 @@ import { apiToken } from "@/lib/portal" export type ChatEvent = | { type: "start" } + | { type: "delta"; content: string } | { type: "message"; content: string } | { type: "tool_call"; name: string; arguments: Record } | { type: "tool_result"; name: string; ok: boolean; error?: string }