Calling an LLM

Inside an agent, the model goes through ctx.llm. Every call, token and tool result is recorded, so a replay never calls the provider again.

@agent
async def support(ctx: AgentContext, message: str) -> str:
    return await ctx.llm.run(
        [{"role": "user", "content": message}],
        tools=[lookup_order],
        system="You are a helpful support assistant.",
    )

Call	Tools	Returns
ctx.llm.run(messages, tools=, system=)	runs the tool loop	the final assistant message’s text
ctx.llm.stream(messages, tools=, system=)	runs the tool loop	async iterator of events (tokens, tool calls, tool results)
ctx.llm.complete(messages, system=)	none	the reply text

tools are @tool functions. The loop runs until the model stops asking for them. run returns only the last message’s text; narration before a tool call (“let me look that up”) is in the event stream, not the answer.

Use stream when you want to forward tokens as they arrive:

async for event in ctx.llm.stream(messages, tools=[search]):
    ...

Providers

Pick a client and pass it on the @agent or at run time. Each provider extra installs its own SDK.

from pyrula.agents.llm import OpenAILLM, AnthropicLLM, LiteLLM

OpenAILLM(model="gpt-4o-mini")
AnthropicLLM(model="claude-sonnet-4-6")
LiteLLM(model="...")   # routes through LiteLLM to anything it supports

The import is lazy: a provider only shows up if its SDK is installed, so install the one you use.

Streaming mode

Each client takes streaming_mode:

from pyrula.agents.stream.kinds import StreamingMode

OpenAILLM(model="gpt-4o-mini", streaming_mode=StreamingMode.LIVE)

Mode	Blocks reach the run stream	A provider error mid-call
SAFE (default)	once the call finishes, in one write	retried, up to 2 more attempts
LIVE	as they stream	not retried; the call fails

SAFE buffers up to max_buffer_bytes (5 MiB) per call; more ends the turn with BUFFER_OVERFLOW. Use LIVE when time-to-first-token matters more than retries.

Long conversations

Before each model call, the loop estimates the prompt’s tokens. Over a budget, it compacts the history; still over the model’s window, it ends the turn with context_window_exceeded instead of sending a request the provider would reject.

• Window: known for common Claude and GPT models. For anything else set context_window, or management is off (a warning says so once per model).
• Budget: (context_window - reserved_output_tokens) × compaction_fraction; defaults 4096 and 0.75.
• Default compactor: drop_oldest keeps the first message and the 6 most recent, drops the oldest in between, and never splits a tool call from its result. It makes no LLM call, so it replays identically. Each compaction writes a context:compacted event.

Set these per agent with AgentLimits, or on the LLM client:

from pyrula.agents import AgentLimits, agent
from pyrula.agents.llm.context import make_drop_oldest

@agent(limits=AgentLimits(
    context_window=128_000,
    compaction_fraction=0.6,
    history_compactor=make_drop_oldest(keep_recent=10),   # None disables compaction
))
async def support(ctx, message: str) -> str:
    ...

A custom compactor is async (CompactionContext) -> list[dict]. It gets history, token_count, budget, window and model, and can call await ctx.summarize(messages, instructions=...), a journaled LLM call, so summarizing stays replay-safe. Keep tool calls and their results together, as drop_oldest does.

Replay safety

Raw SDK calls inside an agent body are not recorded and will fire again on replay. Always go through ctx.llm.stream so the model interaction lands in the run stream. Same rule as any other side effect, covered in Determinism.