Asynchronous calls and streaming

AsyncBroker provides the asynchronous API. Broker uses the same underlying behavior through synchronous methods. Call AsyncBroker methods with await:

async with llmbroker.AsyncBroker() as broker:
    reply = await broker.ask("Hello")
    print(reply.text)

Use await llmbroker.arun_tool_loop(...) for an asynchronous tool loop. See Tools & agents.

Streaming

stream() yields text chunks as they arrive. Model selection and fallback behave as they do for a regular call:

async with llmbroker.AsyncBroker() as broker:
    stream = broker.stream("Write a haiku about brokers", operation="write")
    async for delta in stream:
        print(delta, end="", flush=True)

    print(stream.llm_name, stream.usage)  # model name and usage information
    await stream.record_quality(0.9)  # rate the completed reply

stream(...) returns an asynchronous iterator. After the reply is complete, its fields contain the model name and usage information. The broker owns its streams: leaving its context closes unfinished work and waits for journal writes before closing the HTTP client and storage. No separate stream context is required. The synchronous Broker streams the same way with a plain for loop; see Streaming from synchronous code.

Before the first text chunk arrives, the broker can try another model. This happens if the provider returns an error, rate-limits the request, or completes a reply with neither text nor tool calls. Once a chunk has been returned, switching models would require replacing visible text. If the reply is interrupted at that point, StreamInterruptedError is raised and the application retains the chunks it already received. Behavior with fastest_of > 1 is described under Concurrent streaming.

Concurrent streaming

With fastest_of > 1, the broker calls multiple models concurrently. Every model continues until one produces a complete reply. The model that returns the first text chunk is not necessarily the one that finishes first, so the broker does not cancel the other requests when text first appears.

Chunks received before one reply completes are provisional. If another model finishes first, the iterator raises StreamReplacementError. Its replacement field contains the complete final reply. Replace all previously displayed text with that reply; chunks from different models are never combined.

stream = broker.stream("Write a haiku", fastest_of=2, stream_selection_window=1.0)
parts = []
try:
    async for delta in stream:
        parts.append(delta)
        show(delta)
except llmbroker.StreamReplacementError as exc:
    replace_everything_with(exc.replacement.text)  # replace the provisional text
else:
    text = "".join(parts)

print(stream.llm_name)  # model whose reply became the final result
await stream.record_quality(0.9)  # rate the final reply

Rate a concurrent call through the stream object or exc.replacement. With fastest_of, the journal can contain several successful attempts with the same trace_id, although the user received only one reply. record_quality(..., trace_id=...) cannot determine which reply was shown. Use the returned object or save its call_id.

The iterator is finished after StreamReplacementError and yields no more data. The same handle remains open for another(). Catch this exception before the broader LLMRequestError, or a complete reply will be handled as a failure.

stream_selection_window affects only the text shown first. For this many seconds, one second by default, the broker waits for data from the model currently first in the selection order. If it does not start responding, data already received from another model is shown. If the first model fails, the wait ends immediately. A value of 0 displays the first text received, regardless of model order.

Expiration of stream_selection_window is not an error or a timeout and does not change later model selection. The complete reply from the model that finishes first is still the final result.

Each concurrent request consumes provider quota, and completed replies remain in memory until requested or the handle closes. This is independent of how quickly the application reads chunks. If one model stops generating or does not complete within wait, the broker can use another model's complete reply.

When fastest_of is unset or equals 1, the broker stays with the model that returns the first chunk, and stream_selection_window has no effect.

Requesting another complete answer

Call another() while the broker is open when the application rejects a complete answer. The broker first uses retained lanes and then tries pool models that this routed request has not attempted:

stream = broker.stream(prompt, fastest_of=3, wait=25)
try:
    text = "".join([delta async for delta in stream])
    rated = stream
except llmbroker.StreamReplacementError as exc:
    text, rated = exc.replacement.text, exc.replacement

while not acceptable(text):
    await rated.record_quality(0.0)
    answer = await stream.another()
    if answer is None:
        break
    text, rated = answer.text, answer

Each returned result has its own llm_name, call_id, usage, and rating target. The original stream keeps the identity of its first authoritative answer. None means no further complete answer can be obtained from this call and is returned again on later another() calls.

Keeping the handle open starts no new provider request after the first answer; only another() may do that. Lanes already started may continue and retain their answers. Closing cancels unfinished lanes and waits for their journal rows and capacity releases.

Time limit for the complete reply

wait limits the time used to receive the complete reply, not only the first chunk. It counts only time spent waiting for provider data:

stream = broker.stream("Write a long answer", wait=20.0)
try:
    async for delta in stream:
        print(delta, end="", flush=True)
except llmbroker.LLMTimeoutError as exc:
    print(f"\nresponse timed out: {exc}")

Time spent by the application processing a chunk does not count. Timing pauses when data are handed to the application and resumes when the next chunk is requested. The limit still applies when a model starts quickly but generates the rest of the reply too slowly.

The same positive wait covers the initial answer, application validation, and all calls to another(). After the first complete answer the clock runs continuously. A reply completed before expiry remains available from memory, but the broker starts no new candidate after expiry.

When wait is unset, no user-defined time limit applies. If it expires before the first chunk, the call raises NoLLMAvailableError. After the first chunk, it raises LLMTimeoutError, and the application retains previously delivered data.

Before the first chunk a stream never tries the same model twice, so it ends as soon as every model it could open has been tried, instead of waiting for one of them to come back as ask does. When those models are only cooling down, the error is NoLLMAvailableError with reason="timeout" and retry_at set to the moment the first of them returns, so the application can retry later or fall back to another model; reason="excluded" still means no model can serve the request at all. See When nobody can answer.

A model that returned no data within wait is temporarily excluded from selection. If the model had already started responding, that temporary exclusion does not apply. In both cases, the broker remembers which wait value was too short and prefers another model for a similarly short call. An incomplete reply cannot be rated; calling record_quality on its object raises an exception.

Before the first chunk, llm_name and call_id are None because the broker can still try another model. usage is populated after the reply is complete.

A rating can also be recorded only after the reply is complete and the call is in the journal. Calling record_quality earlier raises ValueError.

You can stop reading at any point, but break does not close the stream. The broker closes it on exit. In a long-lived broker, use the stream's own context to release its resources as soon as this request is finished, including on errors. Closing also allows you to rate a partial reply while the broker is still open:

async with broker.stream("Write a haiku about brokers") as stream:
    async for delta in stream:
        if looks_wrong(delta):
            break

await stream.record_quality(0.0)

When using this optional context, keep validation and another() inside it. await stream.aclose() performs the same early release if needed without a context.

Streaming is also available for one specific model through direct(), without pool selection or fallback. See Direct model calls.

Streaming from synchronous code

The synchronous Broker, and the caller returned by its for_scope(...), provide the same stream() with the same arguments. Iterate it with a plain for loop. Everything above applies unchanged: chunks, fallback before the first chunk, StreamInterruptedError, StreamReplacementError, another(), wait, and rating:

with llmbroker.Broker() as broker:
    with broker.stream("Write a haiku about brokers") as stream:
        for delta in stream:
            print(delta, end="", flush=True)
    stream.record_quality(0.9)

Each exception is raised at the same chunk as in asynchronous code. Nothing is read ahead: a chunk is fetched when the loop asks for the next one. Many threads can stream through one broker at the same time. A stream, like a caller from for_scope(...), keeps its broker running until you finish with it, even when nothing else refers to the broker.

Closing the stream cancels the provider request. Leaving the with block, calling stream.close(), and dropping the object all close it. In a synchronous web server, this is what stops generation when a client disconnects. Django, for example, closes the body of a StreamingHttpResponse when the client goes away, and a generator that reads the stream inside with closes it at that moment:

def answer_body(llms, prompt):
    with llms.stream(prompt) as stream:
        for delta in stream:
            yield delta

After StreamReplacementError, read the complete reply from exc.replacement as in asynchronous code: .text, .llm_name, and .call_id. Its record_quality is a coroutine, so rate the final reply from synchronous code through stream.record_quality(...), which names the replacement once it has replaced the provisional text. another() returns a synchronous result with its own record_quality.

A local database for one process

SQLite stores models, keys, and the journal in one file. This is sufficient for a single-process application. The maintained model list is written before the first call:

async with llmbroker.AsyncBroker("broker.db") as broker:
    print((await broker.ask("Hello")).text)

An empty database is populated automatically before the pool is created, so no separate initialization step is required. The list is then updated automatically. If the remote catalog is unavailable, the broker logs a warning and uses the data in the database. If the database is empty, it uses the copy bundled with the installed package.

If the application uses the same file, account for WAL mode and SQLite file locking. See SQLite: sharing and WAL. For multiple processes or hosts, update the database once during deployment instead. See Servers & clusters.