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")| Parameter | Type | Default | What it does |
|---|---|---|---|
provider | str | "openai" | One of "openai", "anthropic", "gemini", "ollama". |
model | str | None | None | Model name. Uses the provider's default if not set. |
api_key | str | None | None | Reads from the environment variable if not set. |
base_url | str | None | None | Override the provider's API base URL. |
temperature | float | 0.7 | 0.0 = deterministic, 1.0 = creative. |
max_tokens | int | None | None | Cap on output tokens (provider default if unset). |
system_prompt | str | None | None | Prepended to every call automatically. |
track_tokens | bool | True | Set False to disable the TokenTracker. |
pricing | str | None | None | Path to a JSON pricing config file. |
max_retries | int | 0 | Retry attempts on failure (0 = no retries). |
retry_delay | float | 1.0 | Base seconds between retries — exponential backoff + jitter. |
fallback | LLM | None | None | Another LLM to try if all retries fail. |
cache | bool | BaseCache | False | True or a BaseCache instance to enable caching. |
timeout | float | None | None | Seconds before raising a timeout error (SDK default if unset). |
Provider defaults
| Provider | Default model |
|---|---|
openai | gpt-4o-mini |
anthropic | claude-haiku-4-5-20251001 |
gemini | gemini-2.0-flash-lite |
ollama | llama3.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
RoleEnum: USER, ASSISTANT, SYSTEM, TOOL. .value is the lowercase string the provider APIs expect.
Messagerole, content, tool_calls (Optional[list[ToolCall]]), tool_result (Optional[ToolResult]). Immutable dataclass.
TokenUsageinput_tokens, output_tokens, and a .total property. Counts come straight from the API — never estimated.
LLMResponsecontent, usage, model, tool_calls, cost_usd, duration_ms, raw, retrieved_chunks (set by GraphLLM when RAG context was injected).
ToolCallid, name, arguments (dict), thought_signature (Gemini thinking models only — must round-trip verbatim).
ToolResulttool_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 reportChat models — USD per 1M tokens
| Model | Input | Output |
|---|---|---|
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
| Model | Input |
|---|---|
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() # wipeKey: 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)fallbackJitter: 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 errorsProviderErrorbase class for provider API errorsAuthError401/403 — invalid or missing API keygoes to fallbackRateLimitError429 — quota or rate limit exceededretried automaticallyInvalidRequestError400 — bad model name, invalid parametersgoes to fallbackServerError5xx — provider-side failureretried automaticallyProviderTimeoutErrorconnection timeout or deadline exceededretried automaticallyAgentErrorbase class for agent loop errors (raised in PyWrapAI-Graph)AgentBudgetExceededErrorAgentConfig.max_cost_usd exceededAgentContextLimitErrorAgentConfig.max_input_tokens exceededToolNotFoundErrormodel requested an unregistered toolToolArgumentErrortool arguments failed schema validationAgentError 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([...])| Provider | Default model | Token counting |
|---|---|---|
openai | text-embedding-3-small | From API response (exact) |
gemini | models/text-embedding-004 | 4-chars-per-token heuristic |
ollama | nomic-embed-text | 4-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 herepywrapai.llmLLM.chat() / _do_chat() log pointspywrapai.agentAgent.run() / arun() (PyWrapAI-Graph)pywrapai.monitorMonitoredGraph / MonitoredAgent (PyWrapAI-Monitor)| Logger | Level | When emitted |
|---|---|---|
pywrapai.llm | DEBUG | Before every chat() / achat() — provider, model, message count, tool names |
pywrapai.llm | DEBUG | After a successful call — tokens, cost, duration, tool_calls count |
pywrapai.llm | DEBUG | On cache hit — model name |
pywrapai.llm | WARNING | On each retryable failure — attempt number, exception type, delay |
pywrapai.llm | WARNING | When retries are exhausted and fallback is invoked |
What PyWrapAI does not do
| Feature | Lives in |
|---|---|
| Conversation history management | PyWrapAI-Graph |
| RAG pipeline / document retrieval / vector stores | PyWrapAI-Graph |
| Document chunking | PyWrapAI-Graph |
| Tool registration (@tool, ToolRegistry) | PyWrapAI-Graph |
| Tool execution / agent ReAct loop | PyWrapAI-Graph |
| System prompt versioning (file-based) | PyWrapAI-Graph |
| Persistent analytics | PyWrapAI-Monitor |
| Web dashboard | PyWrapAI-Monitor |