Stream¶

Stream(program, args, kwargs)

One call of an AI function or a module, watched while it is made.

Made by fn.stream(...); the call starts at once, in the background. It is the same call as fn(...): the same retries, tools and call log line, and the same value in the end.

Iterate it (for piece in s, or async for) for the answer's text as the model writes it; s.events() for everything that happens, in this call and every call inside it; s.show() to print it as it is written. s.result waits for the call and returns (or raises) what fn(...) would; await s does the same in async code. Iterating twice replays from the start.

When the model is asked again (an unreadable reply, a provider error, an escalation), the pieces already shown cannot be taken back: the next ones are the new answer, and a Retry event says so. So "".join(s) is the text as shown; the answer is s.result, and s.text is always the answer so far.

Closing the stream (s.close(), or the end of a with block) cancels the call if it is still running: s.result then raises Cancelled. A stream never closed runs to its end.

Attributes¶

Examples¶

import functai
from functai import *

@ai
def haiku(topic: str) -> str:
    """A haiku about the topic."""
    ...

for piece in haiku.stream("autumn rain"):
    print(piece, end="", flush=True)

functai: no model chosen, so using gpt-4.1-mini (environment ($OPENAI_API_KEY)). Choose one with functai.configure(lm=...).
Soft autumn rain falls,  
Whispering through amber leaves,  
Nature’s gentle breath.

Methods¶

aclose¶

Stream.aclose()

close(), for async with and async code.

approve¶

Stream.approve(approval=None, *, by=None)

Say yes to a tool call this stream's call waits for (approve is a rule and no function answers: the call waits here). approval: an Approval event, its invocation number, or None for the only one.

close¶

Stream.close()

Stop: cancel the call if it is still running (at the model's next piece of text; the provider may bill what it already generated).

deny¶

Stream.deny(approval=None, reason=None, *, by=None)

Say no: the model is told the person did not allow it (and why).

events¶

Stream.events(view=None)

Every event of the call and of the calls inside it, in order: Started, Request, Text, Thinking, ToolCall, ToolResult, Retry, Done, Failed, Approval, Approved (functai.streaming). Works with for and async for; each event has .kind, .position (its place in its tree's log) and .to_dict() (the contract's JSON, format 2).

view: the events as one kind of reader may see them, as the contract's JSON (what a server sends): "kept" (what log_content lets be kept) or "outside" (a caller who sees only the program's boundary: its answer's text, approvals addressed to it, its end). See functai.views.

show¶

Stream.show(file=None)

Print the call as it is written, and wait for its end.

The answer's text as it arrives; when the function writes several outputs (a reasoning, then the answer), each is labelled; tool calls and their results get a line each; a retry says why. For a module, each AI function it calls, with its answer. In a notebook, the text appears as it is written. Raises what the call raised.

text_of¶

Stream.text_of(fn)

The answer's text of every call of fn inside this stream, as it is written (for a module's stream: s.text_of(summarize)).

wait¶

Stream.wait(timeout=None)

Wait for the call to end (at most timeout seconds); returns the stream.