Reference

PyWrapAI

Single responsibility: call an LLM, embed text, return a normalized response. No conversation history, no RAG, no agents, no databases — those live in PyWrapAI-Graph and PyWrapAI-Monitor.

Constructor

llm = LLM(provider="anthropic", model="claude-haiku-4-5-20251001")
ParameterTypeDefaultWhat it does
providerstr"openai"One of "openai", "anthropic", "gemini", "ollama".
modelstr | NoneNoneModel name. Uses the provider's default if not set.
api_keystr | NoneNoneReads from the environment variable if not set.
base_urlstr | NoneNoneOverride the provider's API base URL.
temperaturefloat0.70.0 = deterministic, 1.0 = creative.
max_tokensint | NoneNoneCap on output tokens (provider default if unset).
system_promptstr | NoneNonePrepended to every call automatically.
track_tokensboolTrueSet False to disable the TokenTracker.
pricingstr | NoneNonePath to a JSON pricing config file.
max_retriesint0Retry attempts on failure (0 = no retries).
retry_delayfloat1.0Base seconds between retries — exponential backoff + jitter.
fallbackLLM | NoneNoneAnother LLM to try if all retries fail.
cachebool | BaseCacheFalseTrue or a BaseCache instance to enable caching.
timeoutfloat | NoneNoneSeconds before raising a timeout error (SDK default if unset).

Provider defaults

ProviderDefault model
openaigpt-4o-mini
anthropicclaude-haiku-4-5-20251001
geminigemini-2.0-flash-lite
ollamallama3.2

Core methods

llm.chat(input, tools=None, tool_choice=None)

Send a message and return the full response. Checks the cache first, applies retry + fallback, records token usage.

inputstr | list[Message]A plain string, or a list of Message objects for multi-turn conversations.
toolslist[dict] | NoneProvider-neutral tool definitions from ToolRegistry.to_schema().
tool_choicestr | None"auto", "required", "none", or a specific tool name.

Returns: LLMResponse — content is "" and tool_calls is non-empty when the model calls a tool instead of replying.

response = llm.chat("What is machine learning?")

response = llm.chat([
    Message(Role.USER,      "My name is Alice."),
    Message(Role.ASSISTANT, "Hello Alice!"),
    Message(Role.USER,      "What is my name?"),
])
print(response.content)  # → Your name is Alice.
llm.stream(input)

Stream the response as text fragments. Token usage is still tracked — read from the final chunk once the generator is exhausted.

inputstr | list[Message]Same as chat(). Passing tools= raises ValueError — use chat() for tool calling.

Returns: Iterator[str] — small text fragments as they arrive.

for chunk in llm.stream("Write a poem about the sea."):
    print(chunk, end="", flush=True)
llm.chat_structured(input, schema)

Forces valid JSON matching a schema. Uses the provider's native JSON mode where available (OpenAI, Gemini, Ollama); falls back to schema-injection for Anthropic.

inputstr | list[Message]The question or prompt.
schematype | dictA Pydantic BaseModel class, or a plain dict.

Returns: A validated Pydantic model instance, or a parsed dict.

class ExtractedFact(BaseModel):
    subject: str
    predicate: str
    object: str

result = llm.chat_structured(
    "Marie Curie discovered polonium.",
    schema=ExtractedFact,
)
print(result.subject)  # → "Marie Curie"
await llm.achat(input, tools=None, tool_choice=None)

Async version of chat() — same parameters, uses the provider's native async client (AsyncOpenAI, AsyncAnthropic, client.aio, httpx.AsyncClient). await llm.astream(input) mirrors stream(), no tools=.

Returns: awaitable LLMResponse (achat) / AsyncIterator[str] (astream).

async def main():
    response = await llm.achat("What is Python?")
    async for chunk in llm.astream("Tell me a story"):
        print(chunk, end="", flush=True)

asyncio.run(main())

Core data types

Role

Enum: USER, ASSISTANT, SYSTEM, TOOL. .value is the lowercase string the provider APIs expect.

Message

role, content, tool_calls (Optional[list[ToolCall]]), tool_result (Optional[ToolResult]). Immutable dataclass.

TokenUsage

input_tokens, output_tokens, and a .total property. Counts come straight from the API — never estimated.

LLMResponse

content, usage, model, tool_calls, cost_usd, duration_ms, raw, retrieved_chunks (set by GraphLLM when RAG context was injected).

ToolCall

id, name, arguments (dict), thought_signature (Gemini thinking models only — must round-trip verbatim).

ToolResult

tool_call_id, name, content — wraps your function's return value to send back to the model.

Token tracking & pricing

Every call is recorded in a TokenTracker, accessed via llm.tokens. Cost comes from a visible, overridable price table — never estimated.

llm.tokens.total_input     # int
llm.tokens.total_cost      # float — USD
llm.tokens.calls           # list[CallRecord]
llm.tokens.summary()       # formatted string report

Chat models — USD per 1M tokens

ModelInputOutput
gpt-4o$2.50$10.00
gpt-4o-mini$0.15$0.60
claude-opus-4-8$15.00$75.00
claude-sonnet-4-6$3.00$15.00
claude-haiku-4-5-20251001$0.80$4.00
gemini-2.0-flash$0.10$0.40
gemini-1.5-flash$0.075$0.30
ollama (any local model)$0.00$0.00

Embedding models — USD per 1M input tokens

ModelInput
text-embedding-3-small$0.02
text-embedding-3-large$0.13
text-embedding-ada-002$0.10
models/text-embedding-004 (Gemini)$0.00
nomic-embed-text (Ollama)$0.00

Unknown models return $0.00 with a logging.warning() — never a hard error. Override anything with LLM(pricing="./my_pricing.json") or globally via load_user_pricing().

Response caching

llm = LLM(provider="openai", cache=True)          # default 1000 entries
llm = LLM(provider="openai", cache=ResponseCache(max_size=500))

llm.cache.size    # entries currently cached
llm.cache.clear() # wipe

Key: SHA-256 over all seven call parameters (model, messages, temperature, max_tokens, system_prompt, tools, tool_choice) — a tool-enabled call never collides with a plain one. Eviction: LRU via an OrderedDict. Thread safety: guarded by a threading.Lock.

Retry & fallback

backup = LLM(provider="anthropic")
llm = LLM(provider="openai", max_retries=3, retry_delay=1.0, fallback=backup)
Attempt 1 fails (retryable) → wait ~0.5–1.0s (jittered)
Attempt 2 fails → wait ~1.0–2.0s
Attempt 3 fails → wait ~2.0–4.0s
Final attempt fails → falls through to fallback

Jitter: delay = base * (0.5 + random() * 0.5) — avoids thundering-herd retries. Retryable: RateLimitError, ServerError, ProviderTimeoutError. Non-retryable errors (AuthError, InvalidRequestError) skip straight to the fallback rather than wasting retry attempts.

Providers

OpenAI

LLM(provider="openai", model="gpt-4o-mini")

Reads OPENAI_API_KEY.

Anthropic

LLM(provider="anthropic", model="claude-haiku-4-5-20251001")

Reads ANTHROPIC_API_KEY. System prompt sent as a separate system= parameter automatically.

Google Gemini

LLM(provider="gemini", model="gemini-2.0-flash")

Reads GOOGLE_API_KEY. Uses the google-genai SDK.

Ollama (local)

LLM(provider="ollama", model="llama3.2")

No API key. Requires `ollama serve` running locally on :11434.

All provider SDK imports are lazy — inside method bodies, not at module load. import pywrapai never fails just because a provider SDK isn't installed.

Exception hierarchy

PyWrapAIErrorbase class for all library errors
ProviderErrorbase class for provider API errors
AuthError401/403 — invalid or missing API keygoes to fallback
RateLimitError429 — quota or rate limit exceededretried automatically
InvalidRequestError400 — bad model name, invalid parametersgoes to fallback
ServerError5xx — provider-side failureretried automatically
ProviderTimeoutErrorconnection timeout or deadline exceededretried automatically
AgentErrorbase class for agent loop errors (raised in PyWrapAI-Graph)
AgentBudgetExceededErrorAgentConfig.max_cost_usd exceeded
AgentContextLimitErrorAgentConfig.max_input_tokens exceeded
ToolNotFoundErrormodel requested an unregistered tool
ToolArgumentErrortool arguments failed schema validation

AgentError and its subclasses are raised by Agent.run()/arun() in PyWrapAI-Graph, but defined here and re-exported — all three libraries share one exception hierarchy with no circular imports.

Embedder

embedder = Embedder(provider="openai")

vectors = embedder.embed(["Paris is nice", "London is rainy"])
vector  = embedder.embed_one("What is the capital of France?")

# async
vectors = await embedder.aembed([...])
ProviderDefault modelToken counting
openaitext-embedding-3-smallFrom API response (exact)
geminimodels/text-embedding-0044-chars-per-token heuristic
ollamanomic-embed-text4-chars-per-token heuristic

Anthropic is not supported for embeddings — it offers no public embedding API. Passing provider="anthropic" raises InvalidRequestError immediately. embedder.tokens is a separate TokenTracker from llm.tokens, with kind="embedding" records.

Structured logging

Every call emits through the standard logging module — the library never calls basicConfig() or attaches its own handler. Attaching a handler and choosing a level is entirely up to your application.

pywrapairoot for all three libraries — attach your handler here
pywrapai.llmLLM.chat() / _do_chat() log points
pywrapai.agentAgent.run() / arun() (PyWrapAI-Graph)
pywrapai.monitorMonitoredGraph / MonitoredAgent (PyWrapAI-Monitor)
LoggerLevelWhen emitted
pywrapai.llmDEBUGBefore every chat() / achat() — provider, model, message count, tool names
pywrapai.llmDEBUGAfter a successful call — tokens, cost, duration, tool_calls count
pywrapai.llmDEBUGOn cache hit — model name
pywrapai.llmWARNINGOn each retryable failure — attempt number, exception type, delay
pywrapai.llmWARNINGWhen retries are exhausted and fallback is invoked

What PyWrapAI does not do

FeatureLives in
Conversation history managementPyWrapAI-Graph
RAG pipeline / document retrieval / vector storesPyWrapAI-Graph
Document chunkingPyWrapAI-Graph
Tool registration (@tool, ToolRegistry)PyWrapAI-Graph
Tool execution / agent ReAct loopPyWrapAI-Graph
System prompt versioning (file-based)PyWrapAI-Graph
Persistent analyticsPyWrapAI-Monitor
Web dashboardPyWrapAI-Monitor