ai.streaming

Page actions AI-ready formats and sharing
Open LLM text
Share with AI
Ask Claude Ask ChatGPT Ask Gemini Ask Copilot
module

AI streaming helpers for SSE + Fragment pattern.

Provides ergonomic wrappers around the core pattern of streaming LLM tokens as re-rendered HTML fragments via Server-Sent Events.

The fundamental pattern::

async def generate():…

AI streaming helpers for SSE + Fragment pattern.

Provides ergonomic wrappers around the core pattern of streaming LLM tokens as re-rendered HTML fragments via Server-Sent Events.

The fundamental pattern::

async def generate():
    text = ""
    async for token in llm.stream(prompt):
        text += token
        yield Fragment("chat.html", "response", text=text)
return EventStream(generate())

This module wraps that pattern into reusable helpers so common cases are one-liners while keeping the underlying primitives accessible.

ai.streaming

Name Type Default Description
type —
qualified_name —
element_type —
description —
source_file —
line_number —
is_autodoc —
autodoc_element —
_autodoc_template —
_autodoc_url_path —
_autodoc_page_type —
title —
doc_content_hash —

Symbols on this page

function extract_markdown_units

Extract complete markdown units from a buffer. Returns (complete, in_progress).

Complete units are full lines (or full code blocks). In-progress is the last incomplete…

Jump to symbol function stream_to_fragments

Wrap an LLM token stream as a Fragment-yielding async generator.

Accumulates tokens and yields aFragmentwith the accumulated text after each token. The…

Jump to symbol async function _stream_fragments

Internal generator that accumulates tokens and yields Fragments.

Jump to symbol function stream_events_to_fragments

Wrapstream_events() as progressive Fragmentrenders.

Jump to symbol function stream_with_sources

Stream LLM tokens as fragments, optionally prefixed with a sources block.

Common RAG pattern: send retrieved sources first (immediate), then stream the AI response…

Jump to symbol async function _stream_with_sources_impl

Internal: yield sources fragment, then stream response fragments.

Jump to symbol
extract_markdown_units
function 
def extract_markdown_units(buffer: str) -> tuple[str, str]

Extract complete markdown units from a buffer. Returns (complete, in_progress).

Complete units are full lines (or full code blocks). In-progress is the last incomplete part (mid-line or inside an unclosed code block). Use this for server-side chunked markdown rendering: render only complete units to avoid partial-syntax artifacts.

Parameters

Name Type Default Description
buffer str —
stream_to_fragments
function 
def stream_to_fragments(tokens: AsyncIterator[str], template_name: str, block_name: str, /, *, context_key: str = 'text', extra_context: dict[str, Any] | None = None) -> AsyncIterator[Any]

Wrap an LLM token stream as a Fragment-yielding async generator.

Accumulates tokens and yields aFragmentwith the accumulated text after each token. The Fragment re-renders the named block with the current text, which htmx swaps into the DOM.

Usage::

from chirp import EventStream
from chirp.ai import LLM
from chirp.ai.streaming import stream_to_fragments

llm = LLM("anthropic:claude-sonnet-4-20250514")

@app.route("/chat", methods=["POST"])
async def chat(request: Request):
    prompt = (await request.form())["prompt"]
    fragments = stream_to_fragments(
        llm.stream(prompt),
        "chat.html", "response",
    )
    return EventStream(fragments)

Parameters

Name Type Default Description
tokens AsyncIterator[str] — Async iterator of string tokens (from ``llm.stream()``).
template_name str — Kida template containing the target block.
block_name str — Name of the block to re-render with each update.
context_key str 'text' Template variable name for the accumulated text. Defaults to ``"text"``.
extra_context dict[str, Any] | None None Additional template variables passed to every Fragment render (e.g., user info, metadata).
_stream_fragments
function async 
async def _stream_fragments(tokens: AsyncIterator[str], template_name: str, block_name: str, *, context_key: str, extra_context: dict[str, Any], fragment_cls: type) -> AsyncIterator[Any]

Internal generator that accumulates tokens and yields Fragments.

Parameters

Name Type Default Description
tokens AsyncIterator[str] —
template_name str —
block_name str —
context_key str —
extra_context dict[str, Any] —
fragment_cls type —
stream_events_to_fragments
function 
def stream_events_to_fragments(events: AsyncIterator[Any], template_name: str, block_name: str, /, *, context_key: str = 'text', extra_context: dict[str, Any] | None = None) -> AsyncIterator[Any]

Wrapstream_events() as progressive Fragmentrenders.

Parameters

Name Type Default Description
events AsyncIterator[Any] —
template_name str —
block_name str —
context_key str 'text'
extra_context dict[str, Any] | None None
stream_with_sources
function 
def stream_with_sources(tokens: AsyncIterator[str], template_name: str, *, response_block: str = 'response', sources_block: str | None = None, sources: Any = None, context_key: str = 'text', extra_context: dict[str, Any] | None = None, share_link_block: str | None = None, on_complete: Callable[[str, Any, dict[str, Any]], Any] | None = None, chunk_renderer: Callable[[str], str] | None = None) -> AsyncIterator[Any]

Stream LLM tokens as fragments, optionally prefixed with a sources block.

Common RAG pattern: send retrieved sources first (immediate), then stream the AI response progressively.

Usage::

@app.route("/ask", methods=["POST"])
async def ask(request: Request):
    question = (await request.form())["question"]
    docs = await db.fetch(Document, "SELECT ... WHERE match(?)", question)
    return EventStream(stream_with_sources(
        llm.stream(f"Context: {docs}\nQ: {question}"),
        "ask.html",
        sources_block="sources",
        sources=docs,
        response_block="answer",
    ))

Parameters

Name Type Default Description
tokens AsyncIterator[str] — Async iterator of string tokens (from ``llm.stream()``).
template_name str — Kida template containing the target blocks.
response_block str 'response' Block name for the streaming response.
sources_block str | None None Block name for the sources (rendered once, first).
sources Any None Context value passed to the sources block.
context_key str 'text' Template variable name for accumulated text.
extra_context dict[str, Any] | None None Additional context for all Fragment renders.
share_link_block str | None None
on_complete Callable[[str, Any, dict[str, Any]], Any] | None None
chunk_renderer Callable[[str], str] | None None
_stream_with_sources_impl
function async 
async def _stream_with_sources_impl(tokens: AsyncIterator[str], template_name: str, *, response_block: str, sources_block: str | None, sources: Any, context_key: str, extra_context: dict[str, Any], share_link_block: str | None, on_complete: Callable[[str, Any, dict[str, Any]], Any] | None, chunk_renderer: Callable[[str], str] | None, fragment_cls: type) -> AsyncIterator[Any]

Internal: yield sources fragment, then stream response fragments.

Parameters

Name Type Default Description
tokens AsyncIterator[str] —
template_name str —
response_block str —
sources_block str | None —
sources Any —
context_key str —
extra_context dict[str, Any] —
share_link_block str | None —
on_complete Callable[[str, Any, dict[str, Any]], Any] | None —
chunk_renderer Callable[[str], str] | None —
fragment_cls type —

View source · /home/runner/work/chirp/chirp/site/../src/chirp/ai/streaming.py:1