Unified streaming LLM interface with provider-agnostic reasoning / tool-call abstraction.
yuullm normalises the streaming differences across LLM providers (OpenAI, Anthropic, and any OpenAI-compatible API) into a uniform AsyncIterator[Reasoning | ToolCall | Response]. It also collects Usage and Cost after the stream ends.
yuullm is stateless — no session, no history management. You own the message list.
Messages and output items are typed structs, tools and content blocks are plain dicts — minimal abstraction, maximum interop.
pip install yuullmimport yuullm
client = yuullm.YLLMClient(
provider=yuullm.providers.OpenAIProvider(api_key="sk-..."),
default_model="gpt-4o",
)
messages = [
yuullm.system("You are a helpful assistant."),
yuullm.user("What is 2+2?"),
]
stream, store = await client.stream(messages)
async for item in stream:
match item:
case yuullm.Reasoning(text=t):
print(f"[thinking] {t}", end="")
case yuullm.Response(text=t):
print(t, end="")
# After stream ends
usage = store["usage"]
cost = store["cost"] # Cost | NoneWhen you give tools to an LLM, the model may respond with ToolCall items instead of (or alongside) text. You need to execute those calls and feed results back. Here's the idiomatic pattern:
import json
import yuullm
messages = [yuullm.user("What's the weather in Paris?")]
while True:
stream, store = await client.stream(messages, tools=tools)
tool_calls: list[yuullm.ToolCall] = []
async for item in stream:
match item:
case yuullm.Reasoning(item=text) if isinstance(text, str):
print(f"[thinking] {text}", end="")
case yuullm.ToolCall() as tc:
tool_calls.append(tc)
case yuullm.Response(item=text) if isinstance(text, str):
print(text, end="")
case yuullm.Tick():
pass # heartbeat during tool-call streaming, safe to ignore
if not tool_calls:
break # model replied with text, done
# Append assistant message containing tool calls
messages.append(yuullm.assistant(
*[{"type": "tool_call", "id": tc.id,
"name": tc.name, "arguments": tc.arguments}
for tc in tool_calls]
))
# Execute each tool and append results
for tc in tool_calls:
result = execute_tool(tc.name, json.loads(tc.arguments))
messages.append(yuullm.tool(tc.id, json.dumps(result)))Key points:
- Use
match/caseto dispatch all four stream item types.Tickcarries no payload — ignore it unless you have a reason not to. - The
while Trueloop handles multi-round tool use (the model may chain multiple tool calls before producing a final text response). yuullm.assistant(...)andyuullm.tool(...)are helpers that buildMessagestructs with the right role and content.
yuullm abstracts away raw provider chunks into Reasoning | ToolCall | Response. But sometimes you need the raw chunks — for example, to forward SSE events to a frontend in real time, or to detect a specific tool call name before the full arguments finish streaming.
The on_raw_chunk hook gives you provider-level visibility without abandoning the streaming abstraction.
Pass a callback to client.stream(). It fires on every raw provider chunk before yuullm processes it:
def forward_to_frontend(chunk):
# chunk type depends on provider:
# OpenAI: openai.types.chat.ChatCompletionChunk
# Anthropic: event object with .type attribute
sse_queue.put(chunk)
stream, store = await client.stream(
messages,
on_raw_chunk=forward_to_frontend,
)Problem: During tool-call streaming, the provider accumulates argument deltas internally and yields nothing to the async-for loop. If your on_raw_chunk hook pushes SSE events into a queue, the consumer loop never gets a chance to flush them until the tool call finishes — SSE events arrive in a burst instead of in real time.
Solution: When on_raw_chunk is registered, yuullm yields Tick() items during tool-call argument accumulation. Tick carries no data; it just keeps your async-for loop spinning so side-channel work (like flushing an SSE queue) can proceed promptly.
If you don't use on_raw_chunk, no Tick is ever emitted — fully backward compatible.
Fires a callback when a specific tool call name is detected in the raw stream, useful for early UI feedback (e.g., showing a "Searching..." indicator before arguments finish streaming):
def on_search_start(index: int):
notify_ui("search_started")
stream, store = await client.stream(
messages,
on_raw_chunk=yuullm.on_tool_call_name("search", on_search_start),
)provider = yuullm.providers.OpenAIProvider(
api_key="sk-...",
base_url="https://api.openai.com/v1", # or any compatible endpoint
provider_name="openai", # used for price lookup
)Works with any OpenAI-compatible API (DeepSeek, OpenRouter, vLLM, etc.) by setting base_url and provider_name:
# DeepSeek
provider = yuullm.providers.OpenAIProvider(
api_key="sk-...",
base_url="https://api.deepseek.com/v1",
provider_name="deepseek",
)provider = yuullm.providers.AnthropicProvider(
api_key="sk-ant-...",
provider_name="anthropic",
)Cost is calculated using a three-level fallback:
| Priority | Source | Description |
|---|---|---|
| 1 (highest) | Provider-supplied | Aggregators like OpenRouter return cost in the API response |
| 2 | YAML config | User-supplied price table for custom / negotiated pricing |
| 3 (lowest) | genai-prices | Community-maintained database via pydantic/genai-prices |
If none match, store["cost"] is None — never blocks.
client = yuullm.YLLMClient(
provider=...,
default_model="gpt-4o",
price_calculator=yuullm.PriceCalculator(
yaml_path="./custom_prices.yaml", # optional
),
)YAML price file format
- provider: openai
models:
- id: gpt-4o
prices:
input_mtok: 2.5 # USD per million input tokens
output_mtok: 10
cache_read_mtok: 1.25 # optional
- provider: anthropic
models:
- id: claude-sonnet-4-20250514
prices:
input_mtok: 3
output_mtok: 15
cache_read_mtok: 0.3
cache_write_mtok: 3.75Matching is exact on (provider, model_id). No fuzzy matching.
ContentItem = TextItem | ImageItem | AudioItem | FileItem
ProtocolItem = ToolCallItem | ToolResultItem
MessageItem = ContentItem | ProtocolItem
Content = list[ContentItem]
MessageContent = list[MessageItem]
class Message:
role: Role
content: MessageContent
provider_extra: dict[str, Any]Helper functions: system(content), user(*items), assistant(*items), tool(tool_call_id, content).
Provider-specific message options can be passed through helper keyword arguments and are stored in provider_extra.
| Type | Fields | Description |
|---|---|---|
Reasoning |
item: ContentItem |
Chain-of-thought fragment |
ToolCall |
id, name, arguments |
Tool invocation (arguments is raw JSON string) |
Response |
item: ContentItem |
Final reply fragment |
Tick |
(none) | Heartbeat during tool-call streaming (only when on_raw_chunk is set) |
YLLMClient(
provider: Provider,
default_model: str,
tools: list[dict] | None = None,
price_calculator: PriceCalculator | None = None,
)Returns (AsyncIterator[StreamItem], store). model and tools override the defaults. After the iterator is exhausted, store["usage"] is a Usage and store["cost"] is Cost | None.
Usage(provider, model, request_id, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, total_tokens)
Cost(input_cost, output_cost, total_cost, cache_read_cost, cache_write_cost, source)./scripts/setup-dev.shInstalls git hooks (currently: pre-push tag/version validation).