স্ট্রিমিং ও কাঠামোবদ্ধ ফলাফল

chunk  chunk  chunk  ...  done(stop_reason, usage)        a good stream
chunk  chunk                                            interrupted: no done
chunk  chunk  error                                     failed after partial text

# 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 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)

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     empty

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: True

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"}']
422

# 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 problems

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 id

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.

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: []

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: truncated

python -m pytest -q tests/test_streaming.py tests/test_results.py | sed 's/ in [0-9.]*s//'

...............................                                          [100%]
31 passed

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'

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: False

# 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 passed