A streaming agent turn¶
This tutorial models a single streaming agent turn — a user sends a message, an
LLM-backed procedure streams its reply token by token, and the completed reply is recorded
as a durable fact. Along the way it introduces two context inputs beyond the earlier
tutorials: an environment (injected configuration, in place of os.environ) and
telemetry sinks (host-supplied observation channels).
send_message ─▶ run_agent_turn ─▶ user_message_sent / agent_replied (reactivity)
│
├─(env: llm)──────────▶ the LLM client config
└─(telemetry: stream_chunk, usage)──▶ live tokens + token counts
The arc here is the one you'd actually follow: start from the API, model the feature around what you learned, then let the generated contract replace your scratch script.
Partly validated
The generation pipeline on this page (through uv sync) is executed and checked by
just tutorials-check. The live run — example.py and demo.py — calls a real
LLM, so it needs an API key and its output varies; those blocks are shown for
illustration and are not executed.
Step 1 — Start from the API¶
Before modelling anything, you learn the provider's streaming API by hand. This throwaway script streams a completion, prints each token as it arrives, and reads the aggregate token usage from the final chunk:
example.py
import os
import openai
from dotenv import load_dotenv
load_dotenv()
client = openai.OpenAI(
api_key = os.getenv("OPENAI_API_KEY"),
base_url = os.getenv("OPENAI_API_BASE"),
)
stream = client.chat.completions.create(
model="google/gemma-4-26b-a4b-it",
messages=[
{
"role": "user",
"content": "What is the meaning of life?"
}
],
stream=True,
stream_options={"include_usage": True},
)
usage = None
for chunk in stream:
if chunk.choices:
content = chunk.choices[0].delta.content
if content:
print(content, end="", flush=True)
# The final chunk carries the aggregate usage when include_usage is set.
if getattr(chunk, "usage", None):
usage = chunk.usage
print()
if usage:
# Pricing in USD per 1M tokens.
PRICE_PER_1M_PROMPT = 0.123
PRICE_PER_1M_COMPLETION = 0.454
prompt_tokens = usage.prompt_tokens
completion_tokens = usage.completion_tokens
total_tokens = usage.total_tokens
prompt_cost = prompt_tokens * PRICE_PER_1M_PROMPT / 1_000_000
completion_cost = completion_tokens * PRICE_PER_1M_COMPLETION / 1_000_000
total_cost = prompt_cost + completion_cost
print()
print("──── telemetry ────")
print(f"prompt : {prompt_tokens:>8} tokens")
print(f"completion : {completion_tokens:>8} tokens")
print(f"total : {total_tokens:>8} tokens")
print(f"cost : ${total_cost:.6f}")
else:
print("[telemetry] no usage returned by the endpoint")
Three things in here become the shape of the feature:
os.getenv(...)for the key and base URL → an environment input (the host injects config; the procedure never readsos.environ).print(content, ...)per token → a telemetry sink (stream_chunk): a transport concern, streamed live, never recorded.- the final
usage→ a second telemetry sink (usage): a turn-level observation.
Running example.py needs an API key, so we won't run it here — it's the sketch we model
against.
Step 2 — Model the feature¶
Translate that sketch into a feature-file. Note the two new sections, environment: and
telemetry:, and how run_agent_turn declares what it consumes (environment: [llm],
telemetry: [stream_chunk, usage]) alongside the events it emits:
agent.feat.yaml
description: |
Agent Chat — a minimal streaming agent turn.
A user sends a message; an LLM-backed procedure streams its reply token by
token to a telemetry sink, then records the completed reply as a durable
event. This slimmed example focuses on the reactivity loop plus the
environment + telemetry context inputs:
send_message → run_agent_turn → user_message_sent / agent_replied (reactivity)
Streaming note: live token streaming is a TRANSPORT concern handled inside
the `run_agent_turn` procedure — it forwards LLM chunks to the `stream_chunk`
telemetry sink, and reports aggregate token counts once via the `usage` sink.
The event stream records only completed facts, never individual tokens.
# ── Write intents ────────────────────────────────────────────────────────
commands:
send_message: |
A user wants to send a message to a conversation.
Carries session_id (the shared conversation key), role (user), and content.
# ── Immutable facts ──────────────────────────────────────────────────────
events:
user_message_sent: |
A user message was appended to a conversation.
Carries session_id, content, and sent_at.
agent_replied: |
The agent completed a reply in a conversation.
Carries session_id, content, replied_at, and token/cost telemetry
(prompt_tokens, completion_tokens, total_tokens, cost_usd). Telemetry is
observed provenance from the model provider, not a derivation.
# ── Business logic ───────────────────────────────────────────────────────
procedures:
run_agent_turn:
description: |
Handle a user message by producing an agent reply.
1. Emit `user_message_sent` to durably record the user's message.
2. Call the LLM with stream=True, forwarding each token chunk to the
`stream_chunk` telemetry sink.
3. On stream completion, report aggregate token usage via the `usage`
telemetry sink, then emit `agent_replied` with the full reply text.
```mermaid
flowchart TD
cmd[send_message] --> umsg[user_message_sent]
umsg --> llm[LLM stream]
llm -- chunks --> stream[stream_chunk]
llm -- done --> usage[usage]
usage --> reply[agent_replied]
```
command: send_message
emits:
- user_message_sent
- agent_replied
environment:
- llm
telemetry:
- stream_chunk
- usage
# ── Injected constants (in place of os env) ──────────────────────────────
environment:
llm: |
The LLM client configuration for this turn — model, API key, and base URL.
Injected by the host in place of reading os env.
# ── Host-injected observation sinks ──────────────────────────────────────
telemetry:
stream_chunk: |
Per-token text delta, forwarded live to the SSE transport — called once per
chunk during the stream. A transport concern, never recorded as an event.
usage: |
Turn-level token usage (prompt/completion/total) reported by the provider
once at stream completion. An observation, not provenance — never an event,
and not available per-chunk. Kept separate from `stream_chunk` because the
two have different cadence and shape.
The event stream records only completed facts (user_message_sent, agent_replied) —
never individual tokens. Streaming is a transport concern, which is exactly why it's
telemetry, not events.
Tip
New to the pipeline? The guestbook tutorial covers definitions → static → libraries in more detail; this one moves quickly.
Step 3 — Scaffold and fill the schemas¶
$ dizzy generate definitions agent.feat.yaml .
Generated def/ stubs and libconfig.yaml. Next steps:
<...>
$ ls -1 def
commands.yaml
environment.yaml
events.yaml
telemetry.yaml
environment.yaml and telemetry.yaml are scaffolded just like commands and events —
with attributes: {} for you to fill. Give the LLM config its fields:
--- a/def/environment.yaml
+++ b/def/environment.yaml
@@ -10,4 +10,16 @@ classes:
description: |-
The LLM client configuration for this turn — model, API key, and base URL.
Injected by the host in place of reading os env.
- attributes: {}
+ attributes:
+ model:
+ description: Provider model identifier (e.g. claude-opus-4-8).
+ range: string
+ required: true
+ api_key:
+ description: Secret API key for authenticating with the provider.
+ range: string
+ required: true
+ base_url:
+ description: Base URL of the provider's API endpoint.
+ range: string
+ required: true
Give each telemetry sink its shape — a single text delta for stream_chunk, and the
token counts for usage:
--- a/def/telemetry.yaml
+++ b/def/telemetry.yaml
@@ -8,13 +8,26 @@ imports:
classes:
stream_chunk:
description: |-
- Per-token text delta, forwarded live to the SSE transport — called once per
- chunk during the stream. A transport concern, never recorded as an event.
- attributes: {}
+ A single live token delta. `run_agent_turn` calls this once per chunk to
+ forward streamed output to the SSE transport. A transport concern — never
+ recorded as an event.
+ attributes:
+ text:
+ description: The incremental text delta for this chunk.
+ range: string
+ required: true
usage:
description: |-
- Turn-level token usage (prompt/completion/total) reported by the provider
- once at stream completion. An observation, not provenance — never an event,
- and not available per-chunk. Kept separate from `stream_chunk` because the
- two have different cadence and shape.
- attributes: {}
+ Turn-level token usage reported by the provider once, at stream
+ completion. An observation of provider provenance — never an event, and
+ not available per-chunk.
+ attributes:
+ prompt_tokens:
+ range: integer
+ required: true
+ completion_tokens:
+ range: integer
+ required: true
+ total_tokens:
+ range: integer
+ required: true
And the command and events:
--- a/def/commands.yaml
+++ b/def/commands.yaml
@@ -10,4 +10,16 @@ classes:
description: |-
A user wants to send a message to a conversation.
Carries session_id (the shared conversation key), role (user), and content.
- attributes: {}
+ attributes:
+ session_id:
+ description: Shared conversation key this message belongs to.
+ range: string
+ required: true
+ role:
+ description: Author of the message — "user" for a sent message.
+ range: string
+ required: true
+ content:
+ description: The message text.
+ range: string
+ required: true
--- a/def/events.yaml
+++ b/def/events.yaml
@@ -10,11 +10,35 @@ classes:
description: |-
A user message was appended to a conversation.
Carries session_id, content, and sent_at.
- attributes: {}
+ attributes:
+ session_id:
+ description: Shared conversation key this message belongs to.
+ range: string
+ required: true
+ content:
+ description: The user's message text.
+ range: string
+ required: true
+ sent_at:
+ description: When the user message was recorded.
+ range: datetime
+ required: true
agent_replied:
description: |-
The agent completed a reply in a conversation.
Carries session_id, content, replied_at, and token/cost telemetry
(prompt_tokens, completion_tokens, total_tokens, cost_usd). Telemetry is
observed provenance from the model provider, not a derivation.
- attributes: {}
+ attributes:
+ session_id:
+ description: Shared conversation key this reply belongs to.
+ range: string
+ required: true
+ content:
+ description: The agent's full reply text.
+ range: string
+ required: true
+ replied_at:
+ description: When the agent reply was recorded.
+ range: datetime
+ required: true
Apply all four:
$ git apply edits/commands.yaml.diff edits/events.yaml.diff edits/environment.yaml.diff edits/telemetry.yaml.diff
Step 4 — Compile and package¶
$ dizzy generate static agent.feat.yaml .
<...>
$ dizzy generate libraries agent.feat.yaml .
<...>
$ ls -1 lib/python-uv
gen_def
gen_int
procedure
pyproject.toml
There's one element here — the run_agent_turn procedure. Its generated context now
carries context.env (the environment) and context.telemetry (the sinks) in addition
to context.emit.
Step 5 — Implement the procedure¶
This is where example.py's logic moves into the procedure — but now reading the LLM
config from context.env.llm instead of os.environ, and forwarding tokens to
context.telemetry.stream_chunk instead of print. It emits user_message_sent first,
streams the reply, reports usage, then emits agent_replied with the full text:
--- a/lib/python-uv/procedure/run_agent_turn/src/run_agent_turn.py
+++ b/lib/python-uv/procedure/run_agent_turn/src/run_agent_turn.py
@@ -1,11 +1,68 @@
-# Implementation stub — fill in your logic here
-from gen_int.python.procedure.run_agent_turn_protocol import run_agent_turn_protocol
+# Implementation of the run_agent_turn procedure against its generated context.
+from datetime import datetime, timezone
+
+import openai
+
from gen_int.python.procedure.run_agent_turn_context import run_agent_turn_context
from gen_def.pydantic.commands import SendMessage
+from gen_def.pydantic.events import UserMessageSent, AgentReplied
+from gen_def.pydantic.telemetry import StreamChunk, Usage # two distinct sinks
def run_agent_turn(
context: run_agent_turn_context,
command: SendMessage,
) -> None:
- raise NotImplementedError
+ # 1. Durably record the user's message as a fact.
+ context.emit.user_message_sent(
+ UserMessageSent(
+ session_id=command.session_id,
+ content=command.content,
+ sent_at=datetime.now(timezone.utc),
+ )
+ )
+
+ # 2. Stream the reply, forwarding each token chunk to the telemetry sink.
+ # The LLM client config is injected via the environment, not os.environ.
+ client = openai.OpenAI(
+ api_key=context.env.llm.api_key,
+ base_url=context.env.llm.base_url,
+ )
+ stream = client.chat.completions.create(
+ model=context.env.llm.model,
+ messages=[{"role": command.role, "content": command.content}],
+ stream=True,
+ stream_options={"include_usage": True},
+ )
+
+ reply_parts: list[str] = []
+ usage = None
+ for chunk in stream:
+ if chunk.choices:
+ delta = chunk.choices[0].delta.content
+ if delta:
+ reply_parts.append(delta)
+ # Per-token transport observation.
+ context.telemetry.stream_chunk(StreamChunk(text=delta))
+ # The final chunk carries aggregate usage when include_usage is set.
+ if getattr(chunk, "usage", None):
+ usage = chunk.usage
+
+ # 3. Turn-level usage observation — reported once, separately from the stream.
+ if usage is not None:
+ context.telemetry.usage(
+ Usage(
+ prompt_tokens=usage.prompt_tokens,
+ completion_tokens=usage.completion_tokens,
+ total_tokens=usage.total_tokens,
+ )
+ )
+
+ # 4. Record the completed reply as a durable fact.
+ context.emit.agent_replied(
+ AgentReplied(
+ session_id=command.session_id,
+ content="".join(reply_parts),
+ replied_at=datetime.now(timezone.utc),
+ )
+ )
Confirm the generated workspace and your implementation form a coherent, installable package:
$ uv sync --project lib/python-uv
<...>
Step 6 — Run it¶
demo.py is the host that replaces example.py: it owns an in-memory event log, supplies
the emit closures, injects the llm environment, and provides the telemetry sinks — then
invokes run_agent_turn exactly as a real runtime would:
demo.py
"""Run the generated agent lib for a single turn, streaming to the CLI.
This is the *host*: it supplies the interstitial infrastructure DIZZY leaves
open — an in-memory event log, the emit/query closures, the injected LLM config
(environment), and the telemetry sink — then invokes the `run_agent_turn`
procedure exactly as a real runtime would.
python demo.py # uses the pre-made message below
python demo.py "hi there" # or pass your own
"""
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
# Make the generated workspace packages importable without `uv sync`.
HERE = Path(__file__).parent
LIB = HERE / "lib" / "python-uv"
for sub in ("gen_def", "gen_int", "procedure/run_agent_turn/src"):
sys.path.insert(0, str(LIB / sub))
load_dotenv(HERE / ".env")
from gen_def.pydantic.commands import SendMessage
from gen_def.pydantic.environment import Llm
from gen_int.python.procedure.run_agent_turn_context import (
run_agent_turn_context,
run_agent_turn_emitters,
run_agent_turn_env,
run_agent_turn_telemetry,
)
from run_agent_turn import run_agent_turn
MODEL = os.getenv("OPENAI_MODEL", "google/gemma-4-26b-a4b-it")
SESSION_ID = "demo-session"
# ── The event log: the source of truth the host owns ──────────────────────
EVENTS: list = []
# ── emit closures: append immutable facts to the log ──────────────────────
def emit_event(event) -> None:
EVENTS.append(event)
# ── telemetry sinks: stream text live, capture usage separately ───────────
captured = {}
def stream_chunk(chunk) -> None:
print(chunk.text, end="", flush=True)
def usage(report) -> None:
captured["usage"] = report
def build_context() -> run_agent_turn_context:
return run_agent_turn_context(
emit=run_agent_turn_emitters(
user_message_sent=emit_event,
agent_replied=emit_event,
),
env=run_agent_turn_env(
llm=Llm(
model=MODEL,
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ["OPENAI_API_BASE"],
)
),
telemetry=run_agent_turn_telemetry(stream_chunk=stream_chunk, usage=usage),
)
def main() -> None:
message = sys.argv[1] if len(sys.argv) > 1 else "What is the meaning of life?"
print(f">>> {message}\n")
run_agent_turn(
build_context(),
SendMessage(session_id=SESSION_ID, role="user", content=message),
)
print()
usage = captured.get("usage")
if usage:
print("\n──── telemetry ────")
print(f"prompt : {usage.prompt_tokens:>8} tokens")
print(f"completion : {usage.completion_tokens:>8} tokens")
print(f"total : {usage.total_tokens:>8} tokens")
# The durable facts the turn recorded (the event stream is the truth).
print(f"\n[events] recorded {len(EVENTS)}: {[type(e).__name__ for e in EVENTS]}")
if __name__ == "__main__":
main()
The live run needs provider credentials. Put them in a .env beside demo.py:
.env
OPENAI_API_KEY=sk-...
OPENAI_API_BASE=https://api.your-provider.example/v1
OPENAI_MODEL=claude-opus-4-8
Then run it (this calls the live provider, so the reply and token counts will differ each time). The text streams in token by token, the telemetry sink reports usage, and the durable events are what remain:
uv run --project lib/python-uv python demo.py "Say hi in five words"
>>> Say hi in five words
Hello there, friendly human greetings!
──── telemetry ────
prompt : 14 tokens
completion : 7 tokens
total : 21 tokens
[events] recorded 2: ['UserMessageSent', 'AgentReplied']
The tokens streamed live through telemetry; the event log recorded only the two completed
facts. That separation — transport vs. provenance — is the point of giving run_agent_turn
both telemetry sinks and events.
The full source for every step on this page lives in the tutorial's own asset folder
(docs/tutorials/agent/ — example.py, demo.py, and the patches under edits/).