Chapter 3 · Streaming, Structured Outputs, and Reliable Responses
Streaming and Structured Results
- Page 5 of 22
- 21 min read
Mateo in Bogotá adds an answer box to a help page. If the page waits for the whole answer, the customer stares at a blank box. If it shows text as it arrives, the page feels fast, but now the code has to cope with half an answer, a connection that drops and an error that comes after some text. Later the same box must give structured data to other code, and that data must not be trusted just because it parses. This page builds both halves of EvidenceDesk.
You know Pydantic and FastAPI from Git, APIs and Backend for AI: Beginner, and the client from Provider-Neutral LLM Clients.
What you will learn
- How a stream differs from a final result, and what a fake stream built on recorded fixtures looks like.
- How to assemble chunks safely: empty, interrupted and failed streams, and why Bangla text must be split by characters, never by bytes.
- How to expose a stream through a small FastAPI endpoint and test it with no server.
- How to define a Pydantic answer schema, validate model output, and recover with a bounded retry or a safe failure.
- Why valid JSON is not a correct answer, and why you check the stop reason before parsing.
Streaming and final results
A final result is one complete response. A stream is a series of events: many partial chunks of text, then a completion event that carries the final metadata (stop reason, token counts). In our code every event has a kind:
chunk chunk chunk ... done(stop_reason, usage) a good stream
chunk chunk interrupted: no done
chunk chunk error failed after partial text| Final result | Stream | |
|---|---|---|
| You receive | one complete response | pieces, then a finish signal |
| Stop reason and usage | in the same response | at the end, if at all |
| A failure | the whole call fails | can come after partial text, when the HTTP status is already 200 |
Provider note (as of 2026-10-12): OpenAI:
responses.create(..., stream=True); the text piece isevent.deltawhenevent.type == "response.output_text.delta", the end is aresponse.completedevent, and anerrorevent has.message. Anthropic:with client.messages.stream(...) as stream, thenstream.text_streamandstream.get_final_message(); the lastmessage_deltacarriesstop_reasonand usage; a mid-stream error arrives as anerrorevent although the HTTP status was already 200. Gemini: events such asstep.delta(text inevent.delta.text),interaction.completedanderror. Ollama:chat(..., stream=True)returns an iterator overchunk['message']['content']; the REST stream is NDJSON and the last chunk hasdone: true.VERIFY: where usage appears in a stream (OpenAI, Gemini, Ollama) and how to cancel one early are not confirmed. Anthropic documents that leaving the
withblock closes the response, but not what the server still bills. Check the current documentation.
A fake streaming client
The fake stream replays a recorded answer from the same fixtures as FakeClient (Provider-Neutral LLM Clients), cut into chunks by a fixed size, so every run is identical. It can simulate a dropped connection (stop_after) and an error (fail_after). The file holds the whole module; later sections use the rest of it.
# file: evidencedesk/streaming.py
"""Streaming: events, a fake stream built on FakeClient, safe assembly, and a small SSE-style endpoint."""
from __future__ import annotations
import json
from collections.abc import Iterable, Iterator
from dataclasses import asdict, dataclass
from typing import Literal
from evidencedesk.llm import FakeClient, LLMRequest, Message, Usage
Kind = Literal["chunk", "done", "error"]
@dataclass(frozen=True)
class StreamEvent:
kind: Kind
text: str = "" # chunk: a piece of the answer; error: a short message; done: empty
usage: Usage | None = None # only on "done"
stop_reason: str | None = None # only on "done"
def chunk_text(text: str, size: int) -> list[str]:
"""Split by characters (code points), never by bytes."""
return [text[i:i + size] for i in range(0, len(text), size)]
class FakeStreamClient:
"""Replays a recorded answer as chunks. A stand-in: real streams arrive in sizes you do not control."""
def __init__(self, fixtures: dict[str, dict] | None = None, chunk_size: int = 12,
stop_after: int | None = None, fail_after: int | None = None) -> None:
self.inner = FakeClient(fixtures)
self.chunk_size = chunk_size
self.stop_after = stop_after # simulate a dropped connection: no "done" event
self.fail_after = fail_after # simulate an error after some chunks
def stream(self, request: LLMRequest) -> Iterator[StreamEvent]:
response = self.inner.complete(request) # raises FixtureMissing on the first next()
for n, piece in enumerate(chunk_text(response.text, self.chunk_size)):
if self.fail_after is not None and n == self.fail_after:
yield StreamEvent("error", "upstream failed")
return
if self.stop_after is not None and n == self.stop_after:
return
yield StreamEvent("chunk", piece)
yield StreamEvent("done", usage=response.usage, stop_reason=response.stop_reason)
@dataclass(frozen=True)
class Assembled:
text: str
status: Literal["complete", "empty", "interrupted", "error"]
usage: Usage | None = None
stop_reason: str | None = None
error: str | None = None
@property
def ok(self) -> bool:
return self.status == "complete"
def assemble(events: Iterable[StreamEvent]) -> Assembled:
"""Rebuild the answer. Only a stream that ends with `done` is complete."""
parts: list[str] = []
for event in events:
if event.kind == "chunk":
parts.append(event.text)
elif event.kind == "error":
return Assembled("".join(parts), "error", error=event.text)
elif event.kind == "done":
text = "".join(parts)
return Assembled(text, "complete" if text else "empty", event.usage, event.stop_reason)
else:
return Assembled("".join(parts), "error", error=f"unknown event kind {event.kind!r}")
text = "".join(parts)
return Assembled(text, "interrupted" if text else "empty")
def sse_lines(events: Iterable[StreamEvent]) -> Iterator[str]:
"""Server-sent-event style text: `event:` and `data:` lines, a blank line after each event."""
for event in events:
data = {"text": event.text}
if event.kind == "done":
data = {"stop_reason": event.stop_reason, "usage": asdict(event.usage or Usage())}
yield f"event: {event.kind}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
def safe_events(client: FakeStreamClient, request: LLMRequest) -> Iterator[StreamEvent]:
"""Turn any failure inside the stream into one generic error event (the HTTP status is already 200)."""
try:
yield from client.stream(request)
except Exception:
yield StreamEvent("error", "upstream failed")
def make_app(client: FakeStreamClient):
from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get("/ask/stream")
def ask_stream(q: str = Query(min_length=1, max_length=500)) -> StreamingResponse:
request = LLMRequest((Message("user", q),))
return StreamingResponse(sse_lines(safe_events(client, request)), media_type="text/event-stream")
return app
def parse_sse(lines: Iterable[str]) -> list[StreamEvent]:
"""Client side: read `event:` / `data:` lines back into events."""
events, kind = [], None
for line in lines:
if line.startswith("event: "):
kind = line[7:]
elif line.startswith("data: ") and kind:
data = json.loads(line[6:])
usage = Usage(**data["usage"]) if "usage" in data else None
events.append(StreamEvent(kind, data.get("text", ""), usage, data.get("stop_reason"))) # type: ignore[arg-type]
return eventsRun the fake and look at the events:
from evidencedesk.llm import LLMRequest, Message
from evidencedesk.streaming import FakeStreamClient
request = LLMRequest((Message("user", "How long does standard delivery take?"),))
for event in FakeStreamClient(chunk_size=24).stream(request):
print(event.kind, repr(event.text), event.stop_reason, event.usage)chunk 'Standard delivery takes ' None None
chunk '4 to 7 business days. Ex' None None
chunk 'press delivery takes 2 b' None None
chunk 'usiness days.' None None
done '' end Usage(input_tokens=38, output_tokens=20)The chunks end in the middle of words. A real model also splits where it likes, so never assume a chunk is a word. The metadata arrives only with done.
The fake stream is a stand-in so tests are repeatable. It replays text that a person wrote. Real models are less predictable, which is exactly why you test.
Assemble chunks safely
assemble rebuilds the answer, and only a stream that ends with done counts as complete. Partial text is kept for logs, but ok is false.
from evidencedesk.streaming import FakeStreamClient, assemble
cases = {"normal": {}, "interrupted": {"stop_after": 2}, "error after partial": {"fail_after": 2}}
for name, options in cases.items():
result = assemble(FakeStreamClient(chunk_size=20, **options).stream(request))
print(f"{name:20} {result.status:12} ok={result.ok!s:5} {result.text!r}")
print(f"{'no events at all':20} {assemble([]).status}")normal complete ok=True 'Standard delivery takes 4 to 7 business days. Express delivery takes 2 business days.'
interrupted interrupted ok=False 'Standard delivery takes 4 to 7 business '
error after partial error ok=False 'Standard delivery takes 4 to 7 business '
no events at all emptyAn interrupted stream is the dangerous case: it ends without an error, and half an answer looks like a short answer.
Chunk boundaries and Bangla text
In UTF-8 a Bangla character takes three bytes. If a chunk ends inside a character, its bytes cannot be decoded alone. Split by characters, as chunk_text does, and the join is always exact:
from evidencedesk.streaming import chunk_text
text = "সাধারণ ডেলিভারিতে ৪ থেকে ৭ কার্যদিবস লাগে।"
raw = text.encode("utf-8")
print(len(text), "characters,", len(raw), "bytes")
by_bytes = [raw[i:i + 8] for i in range(0, len(raw), 8)]
try:
"".join(piece.decode("utf-8") for piece in by_bytes)
except UnicodeDecodeError as exc:
print("bytes:", exc)
damaged = "".join(piece.decode("utf-8", errors="replace") for piece in by_bytes)
print("bytes, errors=replace: damaged marks =", damaged.count("�"), "| same text:", damaged == text)
print("characters: same text:", "".join(chunk_text(text, 8)) == text)42 characters, 114 bytes
bytes: 'utf-8' codec can't decode bytes in position 6-7: unexpected end of data
bytes, errors=replace: damaged marks = 30 | same text: False
characters: same text: TrueIf you only get bytes, use an incremental decoder from codecs. A chunk may still end between a letter and its sign, which looks odd on screen, but the joined text is exact. Reliable Output Contracts returns to Unicode boundaries.
Expose the stream through an API
A browser reads a stream best as server-sent events: lines such as event: chunk and data: {...}, with a blank line after each event. sse_lines writes them, make_app wraps them in a StreamingResponse, and safe_events turns any failure inside the generator into one generic error event, because the status code is already sent by then. The test uses TestClient (no real server); client.stream reads the lines as they come:
from fastapi.testclient import TestClient
from evidencedesk.streaming import FakeStreamClient, make_app
client = TestClient(make_app(FakeStreamClient(chunk_size=30)))
with client.stream("GET", "/ask/stream", params={"q": "How long does standard delivery take?"}) as response:
print(response.status_code, response.headers["content-type"])
for line in response.iter_lines():
print(line)
with client.stream("GET", "/ask/stream", params={"q": "Can I pay in instalments?"}) as response:
print(response.status_code, [line for line in response.iter_lines() if line])
print(client.get("/ask/stream", params={"q": ""}).status_code)200 text/event-stream; charset=utf-8
event: chunk
data: {"text": "Standard delivery takes 4 to 7"}
event: chunk
data: {"text": " business days. Express delive"}
event: chunk
data: {"text": "ry takes 2 business days."}
event: done
data: {"stop_reason": "end", "usage": {"input_tokens": 38, "output_tokens": 20}}
200 ['event: error', 'data: {"text": "upstream failed"}']
422A normal stream; a failure that stays inside the stream (status 200, one error event, no internal detail); and an empty question rejected with 422. Only a check that runs before the stream starts can still change the status code.
Provider note (as of 2026-10-12): FastAPI 0.143.0 and Starlette 1.7.0 are the PyPI versions today.
StreamingResponsetakes an async generator or a normal generator. FastAPI documents that a task can be cancelled only at anawait, and that its "Stream Data" page (FastAPI 0.134.0 and later) handles cancellation for you. FastAPI also has built-in server-sent events (fastapi.sse.EventSourceResponse, 0.135.0 and later); we write the lines by hand to show the format. Starlette documentsawait request.is_disconnected()for streaming responses; whether every endpoint needs it is not settled. Starlette'sTestClientis built onhttpx2(the FastAPI testing page still sayshttpx), and it does not emulate concurrent request and response streaming, so a real disconnect must be tested against a running server. That is not done here. Check the current documentation.
The principle for a client that goes away: stop reading, close the stream, and let the provider SDK documentation tell you what stops on the server (VERIFY). The next page, Reliable Output Contracts, returns to cancellation and event design.
Define a Pydantic response schema
When other code consumes the answer, "some text" is not enough. AnswerResult lists required fields, their types and allowed values: confidence and status are Literals, unknown keys are forbidden, and a small validator ties status to cited_ids. The rest of the module validates, retries and checks business rules; the following sections use it piece by piece.
# file: evidencedesk/results.py
"""Structured results: a Pydantic schema, validation that never trusts the model, a bounded retry, a business-rule check."""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator
from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse
CORPUS = Path(__file__).resolve().parent.parent / "corpus"
SORRY = "Sorry, I could not produce a reliable answer. Please try again or ask a person."
class AnswerResult(BaseModel):
model_config = ConfigDict(extra="forbid") # unknown keys are an error, not ignored
answer: str = Field(min_length=1, max_length=1000)
cited_ids: list[str]
confidence: Literal["high", "medium", "low"]
status: Literal["answered", "abstained"]
@model_validator(mode="after")
def citations_match_status(self):
if self.status == "answered" and not self.cited_ids:
raise ValueError("an answered result needs at least one cited id")
if self.status == "abstained" and self.cited_ids:
raise ValueError("an abstained result must not cite anything")
return self
@dataclass(frozen=True)
class Outcome:
ok: bool
code: str # "ok", "truncated", "unusable_stop", "invalid_json", "invalid_schema"
value: AnswerResult | None = None
problems: tuple[dict, ...] = () # only `loc` and `msg`, never the input or the Pydantic url
attempts: int = 1
@property
def user_message(self) -> str:
return self.value.answer if self.value else SORRY
def parse_answer(response: LLMResponse) -> Outcome:
"""Stop reason first, then JSON syntax, then schema. Never raises on model output."""
if response.stop_reason == "length":
return Outcome(False, "truncated")
if response.stop_reason != "end":
return Outcome(False, "unusable_stop")
try:
value = AnswerResult.model_validate_json(response.text)
except ValidationError as exc:
problems = tuple({"loc": ".".join(str(p) for p in e["loc"]), "msg": e["msg"]} for e in exc.errors())
code = "invalid_json" if exc.errors()[0]["type"] == "json_invalid" else "invalid_schema"
return Outcome(False, code, problems=problems)
return Outcome(True, "ok", value)
RETRYABLE = {"invalid_json", "invalid_schema"}
def get_answer(client: LLMClient, request: LLMRequest, max_attempts: int = 2) -> Outcome:
"""At most `max_attempts` calls. A cut-off answer is not retried: the same request would be cut off again."""
if max_attempts < 1:
raise ValueError("max_attempts must be at least 1")
outcome = Outcome(False, "invalid_json")
for attempt in range(1, max_attempts + 1):
outcome = parse_answer(client.complete(request))
if outcome.ok or outcome.code not in RETRYABLE:
break
return Outcome(outcome.ok, outcome.code, outcome.value, outcome.problems, attempt)
class SequenceClient:
"""Test double: returns the given responses in order (a bad one first, a good one second)."""
def __init__(self, responses: list[LLMResponse]) -> None:
self.responses = list(responses)
self.calls = 0
def complete(self, request: LLMRequest) -> LLMResponse:
self.calls += 1
return self.responses[min(self.calls, len(self.responses)) - 1]
def load_docs(folder: Path = CORPUS) -> dict[str, dict]:
"""Corpus documents by id: {"status": ..., "text": ...}. Reads the small front matter only."""
docs = {}
for path in sorted(folder.glob("*.md")):
head, _, body = path.read_text(encoding="utf-8").removeprefix("---\n").partition("\n---\n")
meta = dict(line.split(": ", 1) for line in head.splitlines() if ": " in line)
docs[meta["id"]] = {"status": meta["status"], "text": body}
return docs
def business_problems(result: AnswerResult, docs: dict[str, dict]) -> list[str]:
"""Rules a schema cannot express. A weak check, not a proof: it catches only what it looks for."""
problems = []
for doc_id in result.cited_ids:
if doc_id not in docs:
problems.append(f"unknown id {doc_id}")
elif docs[doc_id]["status"] != "current":
problems.append(f"{doc_id} is {docs[doc_id]['status']}")
sources = " ".join(docs[i]["text"] for i in result.cited_ids if i in docs)
for number in sorted(set(re.findall(r"\d+", result.answer))):
if re.search(rf"\b{number}\b", sources) is None:
problems.append(f"number {number} not in the cited text")
return problemsValidate model output
Model output is untrusted input, like a form field. parse_answer never raises on bad output: it returns an Outcome with a code and, for schema errors, only loc and msg (not the input, not the Pydantic URL). Try six replies:
import json
from evidencedesk.llm import LLMResponse, Usage
from evidencedesk.results import parse_answer
def reply(text, stop="end"):
return LLMResponse(text, "fake-model", Usage(10, 10), stop)
good = {"answer": "You can return a new, unridden bike within 30 days of delivery.",
"cited_ids": ["kc-returns-v3"], "confidence": "high", "status": "answered"}
cases = {
"good": json.dumps(good),
"prose instead of JSON": "Sure! The answer is 30 days.",
"broken JSON": json.dumps(good)[:40],
"bad value, extra key": json.dumps({**good, "confidence": "certain", "note": "hi"}),
"missing field": json.dumps({k: v for k, v in good.items() if k != "status"}),
"answered, no citation": json.dumps({**good, "cited_ids": []}),
}
for name, text in cases.items():
outcome = parse_answer(reply(text))
print(f"{name}: {outcome.code}")
for problem in outcome.problems:
print(" ", problem["loc"] or "(whole)", "|", problem["msg"])good: ok
prose instead of JSON: invalid_json
(whole) | Invalid JSON: expected value at line 1 column 1
broken JSON: invalid_json
(whole) | Invalid JSON: EOF while parsing a string at line 1 column 40
bad value, extra key: invalid_schema
note | Extra inputs are not permitted
confidence | Input should be 'high', 'medium' or 'low'
missing field: invalid_schema
status | Field required
answered, no citation: invalid_schema
(whole) | Value error, an answered result needs at least one cited idEvery failure is rejected the same way, with a precise place and message. Unknown keys fail too (extra="forbid"), because a silently ignored key hides a model that drifted.
Recover from invalid output
A failed parse may be a one-off glitch, so get_answer retries, but at most max_attempts times, and then returns a safe failure: ok=False and a fixed apology for the user. Every attempt costs money and time. Here SequenceClient plays a model that answers badly first and well second:
from evidencedesk.results import SequenceClient, get_answer
bad = reply("Sure! Here is the answer you asked for.")
scripts = {"bad, then good": [bad, reply(json.dumps(good))], "always bad": [bad]}
for name, script in scripts.items():
client = SequenceClient(script)
outcome = get_answer(client, request, max_attempts=3)
print(f"{name}: ok={outcome.ok} code={outcome.code} attempts={outcome.attempts} calls={client.calls}")
print(" user sees:", outcome.user_message)bad, then good: ok=True code=ok attempts=2 calls=2
user sees: You can return a new, unridden bike within 30 days of delivery.
always bad: ok=False code=invalid_json attempts=3 calls=3
user sees: Sorry, I could not produce a reliable answer. Please try again or ask a person.Only invalid JSON and invalid schema are retried, and the user message never contains the broken text. A cut-off answer is not retried: the same request would be cut off again.
Valid is not the same as correct
Validation proves shape, not truth. A model can return perfect JSON that cites a real document and still states a wrong fact. business_problems adds three rules the schema cannot express: cited ids must exist, must not be superseded, and every number in the answer must appear in the cited text. Against the Kestrel Cycles corpus (the old returns policy says 14 days, the current one says 30 for bikes):
from evidencedesk.results import AnswerResult, business_problems, load_docs
docs = load_docs()
claims = {
"correct": ("A new bike can be returned within 30 days.", "kc-returns-v3"),
"invented number": ("A new bike can be returned within 45 days.", "kc-returns-v3"),
"superseded source": ("A new bike can be returned within 14 days.", "kc-returns-v2"),
"wrong, passes all checks": ("A new bike can be returned within 14 days.", "kc-returns-v3"),
}
for name, (answer, doc_id) in claims.items():
result = AnswerResult(answer=answer, cited_ids=[doc_id], confidence="high", status="answered")
print(f"{name:25} schema ok | business problems: {business_problems(result, docs)}")correct schema ok | business problems: []
invented number schema ok | business problems: ['number 45 not in the cited text']
superseded source schema ok | business problems: ['kc-returns-v2 is superseded']
wrong, passes all checks schema ok | business problems: []The first three are caught. The last one is not: "14 days" does appear in the current policy, for accessories, so every check passes and the claim is still wrong. This weak check is a floor, not a proof: real support needs the cited text compared with the claim, and an eval set measures how often answers are wrong.
Handle truncation
If the model hits its output limit, the answer just stops. Check the stop reason first: "length" means the response is not usable, whatever the text looks like. If you parse first, a cut-off JSON object shows up as a confusing syntax error, and your retry loop wastes calls on a problem it cannot fix.
from evidencedesk.results import AnswerResult
whole = json.dumps(good)
cut = reply(whole[:70], "length")
print("stop reason first:", parse_answer(cut).code)
try:
AnswerResult.model_validate_json(cut.text)
except ValueError as exc:
print("parse first :", exc.errors()[0]["type"])
print("complete text, stop=length:", parse_answer(reply(whole, "length")).code)stop reason first: truncated
parse first : json_invalid
complete text, stop=length: truncatedProvider note (as of 2026-10-12): Structured output is requested differently by each provider: OpenAI uses
text.formatwithtype: "json_schema"andstrict: true(orresponses.parsewith a Pydantic model); Anthropic usesoutput_config.formatwith a JSON schema (ormessages.parse); Gemini's Interactions API takes aresponse_formatwith a schema; Ollama takes aformatparameter. What to check before parsing: for OpenAI, a content part of typerefusalandstatus == "incomplete"with reasonmax_output_tokens; for Anthropic,stop_reasonofrefusalormax_tokens; for Gemini, afinishReasonother thanSTOP(the Interactions equivalent is not confirmed); for Ollama,done_reason(its values are not confirmed). Each provider supports only a subset of JSON Schema, and Gemini's docs say values still need your own validation. Check the current documentation.
Run the tests
Both modules come with tests in tests/test_streaming.py and tests/test_results.py (in the kit; they hold every case above as assertions, plus Bangla at many chunk sizes):
python -m pytest -q tests/test_streaming.py tests/test_results.py | sed 's/ in [0-9.]*s//'............................... [100%]
31 passedThat is the pass line; sed removed the seconds. These tests cannot show how a real stream splits text or how often a real model returns bad JSON; they show that your code survives both.
Where this shows up at work
Every chat-style product streams, and every integration that feeds a model's output to other code validates it. Omar in Cairo reviews such features and asks three things: what does the user see when the stream breaks half way, what happens after the second bad JSON, and which check catches a well-formed wrong answer?
Common mistakes
- Treating a stream that ended as a complete answer. Without
doneit is only a prefix. Check the status, and keep partial text out of the user's answer. - Splitting or decoding by bytes. Multi-byte characters break. Work with
str, or an incremental decoder. - Parsing before checking the stop reason. A truncated answer looks like a syntax problem and gets retried in vain.
- Unlimited retries, or retrying everything. Bound them; retry only what another try can fix. Never show raw validation errors to users.
- Believing "it validates" means "it is right". Add business rules, then measure with evals.
Try it yourself
- Easy. Predict, then run: what are the status and text of a stream with
chunk_size=20that stops after one chunk? - Medium. Revisit Provider-Neutral LLM Clients: its battery question has a fixture that stops early with stop reason
length. Stream it. Doesassemblesayok? Write the rule that says whether the answer is usable. - Hard. Exercise, stream and validate a task result: write a pytest file that streams a JSON answer in chunks, assembles it and validates it with
parse_answer, with tests for a good stream, an interrupted stream, an error after partial text, bad JSON and a truncated stream.
Answers
Easy. One chunk of 20 characters arrived, then nothing: status interrupted.
from evidencedesk.llm import LLMRequest, Message
from evidencedesk.streaming import FakeStreamClient, assemble
request = LLMRequest((Message("user", "How long does standard delivery take?"),))
result = assemble(FakeStreamClient(chunk_size=20, stop_after=1).stream(request))
print(result.status, repr(result.text))interrupted 'Standard delivery ta'Medium. The stream finished normally, so ok is true, but the stop reason says the model was cut off. "Stream complete" and "answer complete" are different questions.
battery = LLMRequest((Message("user", "How should I store the e-bike battery for a long time?"),))
result = assemble(FakeStreamClient(chunk_size=30).stream(battery))
print(result.status, result.ok, result.stop_reason)
print("usable:", result.ok and result.stop_reason == "end")complete True length
usable: FalseHard. The helper never parses a stream that is not ok, and passes the stop reason on to the parser.
# file: tests/test_p05_exercise.py
import json
from evidencedesk.llm import LLMRequest, LLMResponse, Message
from evidencedesk.results import parse_answer
from evidencedesk.streaming import FakeStreamClient, assemble
Q = "Which returns document is old?"
PAYLOAD = {"answer": "kc-returns-v2 is the old returns policy.", "cited_ids": ["kc-returns-v2"],
"confidence": "medium", "status": "answered"}
def task_result(text, stop="end", **options):
fixtures = {Q: {"text": text, "usage": {"input_tokens": 12, "output_tokens": 30}, "stop_reason": stop}}
streamed = assemble(FakeStreamClient(fixtures, chunk_size=9, **options).stream(LLMRequest((Message("user", Q),))))
if not streamed.ok:
return streamed.status # interrupted or error: never parse a partial stream
return parse_answer(LLMResponse(streamed.text, "fake-model", streamed.usage, streamed.stop_reason))
def test_complete_stream_gives_a_valid_result():
outcome = task_result(json.dumps(PAYLOAD))
assert outcome.ok and outcome.value.cited_ids == ["kc-returns-v2"]
def test_interrupted_stream_is_never_parsed():
assert task_result(json.dumps(PAYLOAD), stop_after=2) == "interrupted"
def test_error_after_partial_text_is_never_parsed():
assert task_result(json.dumps(PAYLOAD), fail_after=2) == "error"
def test_complete_stream_with_bad_json_is_rejected():
assert task_result("The old one is kc-returns-v2.").code == "invalid_json"
def test_cut_off_stream_is_truncated_not_invalid_json():
assert task_result(json.dumps(PAYLOAD)[:30], stop="length").code == "truncated"python -m pytest -q tests/test_p05_exercise.py | sed 's/ in [0-9.]*s//'..... [100%]
5 passedPass checklist:
- Easy: status
interruptedand the first 20 characters. - Medium:
okis true, the stop reason islength, and your rule printsusable: False. - Hard: five tests pass; no test needs a network; a partial stream is never passed to the parser; the truncated case reports
truncated, notinvalid_json.
Summary
- A stream is chunks plus a finish event with the metadata. Only a stream that ends with
doneis complete; interrupted and failed streams are separate states. - Test streaming offline with a fake that replays recorded chunks; split text by characters, never by bytes.
- Expose a stream as server-sent-event lines, test it with
TestClient, and turn failures inside it into anerrorevent. - A Pydantic schema states required fields, types and allowed values. Validate every model output, report only
locandmsg, and recover with a bounded retry or a safe failure. - Valid JSON is not a correct answer: add business rules, and check the stop reason before parsing.
Next: Reliable Output Contracts adds schema versions, predictable empty, partial and error states, cancellation and disconnects, size and URL checks, provider comparisons and regression fixtures.