UniLLM

Lightweight LLM access layer with provider normalization, caching, and observability. It uses a unified endpoint format (model@provider) so application code can switch providers without rewriting call sites, while still allowing developers to choose the model provider or compatible endpoint they want to use.

What layer is this?

UniLLM is the model-access layer of unify, a self-improving agent harness: every LLM call the harness makes goes through it. It normalizes requests across providers (OpenAI, Anthropic, Vertex AI, etc.), provides response caching for test determinism, and attaches attribution metadata to each call's events.

If you're here from the unify quickstart, this is the layer that talks to model providers. OpenAI and Anthropic are the simplest documented paths, but the point of UniLLM is that the provider choice is yours, including other supported providers and compatible local endpoints.

Installation

Install from Git:

pip install "unillm @ git+https://github.com/unifyai/unillm.git"

Or with uv:

uv add "unillm @ git+https://github.com/unifyai/unillm.git"

The import name is unillm:

import unillm

Configuration

API Keys

Set credentials for whichever providers you want to use. OpenAI models are available through OpenRouter; UniLLM has no native OpenAI chat transport.

export OPENROUTER_API_KEY=<your-key>
export ANTHROPIC_API_KEY=<your-key>
# ... other provider keys

To send OpenRouter calls to an OpenAI-compatible gateway instead of OpenRouter, set both UNILLM_LLM_GATEWAY_URL (its base URL) and UNILLM_LLM_GATEWAY_KEY (its key) in the process environment. The gateway then holds the provider key, and other providers are still called directly.

Google Cloud / Vertex AI

For Vertex AI models (Gemini, Claude on Vertex, etc.), authenticate using Google Cloud Application Default Credentials:

# One-time setup: authenticate with your Google Cloud account
gcloud auth application-default login

# Set your GCP project and location
export VERTEXAI_PROJECT=<your-project-id>
export VERTEXAI_LOCATION=<your-location>  # e.g., us-central1, europe-west1

Alternatively, use a service account JSON file:

export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json

Basic Usage

import unillm

# Sync client
client = unillm.Unify("openai/gpt-4o@openrouter")
response = client.generate(
    messages=[{"role": "user", "content": "Hello!"}]
)

# Async client
async_client = unillm.AsyncUnify("claude-sonnet-5@anthropic")
response = await async_client.generate(
    messages=[{"role": "user", "content": "Hello!"}]
)

Features

Unified Endpoint Format

All models use a consistent model@provider format:

client = unillm.Unify("openai/gpt-4o@openrouter")
client = unillm.Unify("claude-sonnet-5@anthropic")
client = unillm.Unify("gemini-2.0-flash@vertexai")

Provider-Specific Preprocessing

Automatic handling of provider quirks (message format normalization, parameter translation, etc.) before requests are sent.

Prompt Caching

Every Claude call, direct (@anthropic) or through OpenRouter (Bedrock and Vertex Claude included), carries Anthropic's cache_control breakpoints with the one-hour TTL, because Claude reads nothing from its prompt cache without them. A client places three by default: after the last tool, after the last static system block, and on the last message, so each turn of a conversation reads the turn before it. Other models are sent no breakpoints. OpenAI's through OpenRouter (measured on GPT-6 Sol) read from the cache only as far as an earlier request's whole prompt ran: a conversation that grows turn by turn reads each turn before it, but calls that share an opening and differ after it read none of it until one call has sent the shared messages alone, after which every call that starts with exactly those messages reads them. For every model, what earns cache reads is a request ordered from most to least shared, with nothing that varies between calls (a timestamp, a run id) ahead of shared content.

A caller marks any further shared block, such as a document every call of a task reads, with "_cache_breakpoint": True on the content block (or on the message, for string content):

messages = [
    {"role": "system", "content": RUBRIC},
    {
        "role": "user",
        "content": [
            {"type": "text", "text": transcript, "_cache_breakpoint": True},
            {"type": "text", "text": question},
        ],
    },
]
client = unillm.Unify("anthropic/claude-opus-5.5@openrouter")
client.generate(messages=messages)

A block that already carries a cache_control of its own counts as a mark too. Every breakpoint gets the one-hour TTL, since Anthropic rejects a one-hour breakpoint after a five-minute one. Anthropic accepts four breakpoints per request; when the caller's marks would exceed that, the earliest of them are dropped. A system block marked "_static": False (a clock, say) ends the cached system prefix before it. prompt_caching=["system"] places only the listed defaults, and prompt_caching=None sends no breakpoints at all. Breakpoints are left out of the response cache key, so a recorded response serves a request whatever its breakpoints.

A caller that posts a Claude request body itself, outside a client, gives it the same breakpoints with unillm.place_cache_breakpoints(body).

Calls that share a prefix and start at the same moment each write it and read none of it: send the first alone, and start the rest once its reply has begun. For OpenAI's models that first call carries the shared messages alone. Each completion's usage.prompt_tokens_details.cached_tokens says how much was read from cache.

Response Caching

Built-in caching to avoid redundant LLM calls. Cache entries are stored as NDJSON (.cache.ndjson). At runtime UniLLM keeps a hash→offset sidecar index (.cache.ndjson.idx) so processes do not load multi-GB key strings into RAM. The first open of a cache file builds or validates the sidecar; after consolidating or hand-editing .cache.ndjson, delete any *.ndjson.idx so the next process rebuilds cleanly.

client = unillm.Unify("openai/gpt-4o@openrouter", cache=True)

# Cache modes
client.generate(..., cache="read")       # Read from cache only if available
client.generate(..., cache="write")      # Write to cache only
client.generate(..., cache="both")       # Read and write
client.generate(..., cache="read-only")  # Must be in cache, else error

Cache Event Capture

Track cache hit/miss status for observability:

from unillm import capture_cache_events

with capture_cache_events() as events:
    client.generate(messages=[...])

print(events[0]["cache_status"])  # "hit" or "miss"

Streaming

client = unillm.Unify("openai/gpt-4o@openrouter", stream=True)
for chunk in client.generate(messages=[...]):
    print(chunk, end="")

Stateful Conversations

client = unillm.Unify("openai/gpt-4o@openrouter", stateful=True)
client.generate(user_message="What is 2+2?")
client.generate(user_message="And what is that times 3?")  # Maintains history

Tool Calling

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "parameters": {"type": "object", "properties": {...}}
    }
}]
response = client.generate(messages=[...], tools=tools)

Response Formats

from pydantic import BaseModel

class Answer(BaseModel):
    value: int
    explanation: str

response = client.generate(
    messages=[{"role": "user", "content": "What is 2+2?"}],
    response_format=Answer
)

Observability

Console & File Logging

Terminal and file logging are independently controlled:

# Terminal (console) output (default: true)
export UNILLM_TERMINAL_LOG=true

# File-based traces (independent of terminal)
export UNILLM_LOG_DIR=/path/to/logs

When UNILLM_LOG_DIR is set, structured log files are written:

During call (cache enabled): {base}.cache_pending.txt
During call (cache disabled): {base}.pending.txt
After completion: {base}.cache_hit.txt or {base}.cache_miss.txt (cache enabled), or {base}.txt (cache disabled)

Pending files remain as evidence if an LLM call hangs or crashes.

OpenTelemetry Tracing

# Enable OTel tracing
export UNILLM_OTEL=true

# OTLP endpoint (optional)
export UNILLM_OTEL_ENDPOINT=http://localhost:4317

# File-based span export (optional)
export UNILLM_OTEL_LOG_DIR=/path/to/traces

LLM calls create OTel spans that can be correlated with parent application spans and propagated to child services.

Privacy Note

When UNILLM_LOG_DIR is set, full request and response payloads are written to disk. This includes user messages, tool arguments, and model responses — which may contain PII or sensitive data. Be mindful of this when enabling file logging in production environments.

Supported Providers

OpenAI
Anthropic
Vertex AI (Google)
Bedrock (AWS)
DeepSeek
Groq
Mistral
Replicate
Together AI
xAI

Project Structure

unillm/
├── __init__.py              # Public API exports
├── settings.py              # Configuration via env vars
├── helpers.py               # Utility functions
├── costs.py                 # Provider cost computation
├── cost_tracker.py          # Per-call cost event capture
├── tokens.py                # Token counting and context window utilities
├── cache_events.py          # Cache hit/miss event capture
├── llm_events.py            # LLM event hooks for observability
├── logger.py                # File logging and OTel tracing
├── clients/                 # LLM client implementations
│   ├── base.py              # Base client class
│   ├── uni_llm.py           # Unify (sync) and AsyncUnify clients
│   ├── provider_preprocessing.py   # Provider-specific request normalization
│   ├── provider_postprocessing.py  # Response validation and retries
│   └── shared_session.py    # Shared aiohttp session management
├── caching/                 # Response caching system
│   ├── base_cache.py        # Abstract cache backend
│   ├── ndjson_index.py      # Hash→offset index over NDJSON cache files
│   ├── local_cache.py       # File-based NDJSON cache
│   └── local_separate_cache.py  # Split read/write cache (for CI)
├── endpoints/               # Provider-specific model mappings
│   ├── openai.py
│   ├── anthropic.py
│   └── ...
└── types/                   # Type definitions
    ├── cache.py
    ├── prompt.py
    └── prompt_caching.py

Local Development

unillm is a library, not a runnable system. To run it inside the agent harness, follow the install steps in unify-agent. The steps below are for developing this library itself.

This project uses uv for dependency management.

Setup

git clone https://github.com/unifyai/unillm.git
cd unillm
uv sync

Running Tests

Live tests require at minimum an OPENROUTER_API_KEY or ANTHROPIC_API_KEY in your environment (or a .env file). With a populated .cache.ndjson, cached LLM responses are replayed, so tests run quickly and deterministically without making real LLM calls.

uv run pytest tests/ -v

Running Tests in CI

Tests are opt-in to reduce GitHub Actions costs. Tests only run when explicitly requested:

Commit message: Include [run-tests] in your commit message
PR title: Include [run-tests] in your pull request title
Manual trigger: Use the "Run workflow" button in GitHub Actions

Examples:

# Run tests on this commit
git commit -m "Fix caching logic [run-tests]"

# No tests (default)
git commit -m "Update README"

Note: The black formatting check always runs on every push.

LLM cache: CI tests replay cached LLM responses (read-only). Changes that invalidate cache keys require refreshing .github/cache-seed/cache.ndjson and publishing it via llm-cache-refresh.yml before pytest can pass in CI. See CONTRIBUTING.md — LLM cache and CI.

Pre-commit Hooks

Pre-commit hooks run automatically on git commit (Black, isort, autoflake). If a commit fails due to auto-formatting, re-run the commit.

pre-commit install

License

MIT — see LICENSE for details.

GeezerVison

Latest commit

History

Folders and files

Repository files navigation