The Requests library for AI one Unified Python SDK for every LLM provider
# UniversalAI
**The Requests library for AI** — one unified SDK for every LLM provider. Write once, run anywhere.
pip install universal-ai
from universal_ai import AI
ai = AI(provider="openai", model="gpt-4o")
response = await ai.chat("What is quantum computing?")
print(response.content)
## Why UniversalAI?
Building AI applications today means juggling multiple provider SDKs, each with different APIs, error handling, and quirks. UniversalAI gives you **one clean interface** that works across all major providers:
* **Same code** works with OpenAI, Anthropic, Gemini, Ollama, Groq, Mistral, OpenRouter, HuggingFace, and Azure OpenAI
* **Switch providers** by changing one string — no code rewrite
* **Built-in resilience** with retry, caching, rate limiting, and circuit breaker middleware
* **Tool calling** works identically across all providers that support it
## Features
Feature | Description
---|---
**9 Providers** | OpenAI, Anthropic, Gemini, Ollama, Groq, Mistral, OpenRouter, HuggingFace, Azure OpenAI
**Async-first** | Full `async`/`await` with synchronous wrappers for scripts and notebooks
**Streaming** | Real-time token streaming from any provider
**Tool Calling** | `@tool` decorator with automatic execution loop
**Middleware** | Retry, cache, rate limit, circuit breaker, cost tracking, logging
**Routing** | Fallback, round-robin, lowest latency, lowest cost strategies
**Vision** | Image-aware chat with OpenAI, Anthropic, Gemini
**Audio** | Transcription (Whisper) and TTS with OpenAI
**Image Generation** | DALL-E 3 support
**Embeddings** | OpenAI, Gemini, Mistral, HuggingFace, Azure, OpenRouter
**RAG** | Built-in retrieval-augmented generation with chunking and vector store
**Agents** | Multi-agent orchestration with coordinator pattern
**Context Safety** | Automatic validation and optional truncation
**Cost Tracking** | Per-request and cumulative cost estimation
**CLI** | Full-featured `uai` command-line tool
## Quick Start
### Installation
# Core SDK (auto-detects available providers)
pip install universal-ai
# With specific provider support
pip install universal-ai[openai]
pip install universal-ai[anthropic]
pip install universal-ai[gemini]
pip install universal-ai[ollama]
# Everything
pip install universal-ai[all]
### Basic Usage
import asyncio
from universal_ai import AI
async def main():
# Auto-detect provider from environment
ai = AI()
# Chat
response = await ai.chat("Explain quantum computing in one sentence")
print(response.content)
# Streaming
async for chunk in ai.stream("Write a haiku about programming"):
print(chunk.delta, end="", flush=True)
# Embeddings
embed_response = await ai.embed("Hello, world!")
print(f"Embedding dimensions: {len(embed_response.vector)}")
asyncio.run(main())
### With Specific Provider
from universal_ai import AI
# OpenAI
ai = AI(provider="openai", model="gpt-4o")
response = await ai.chat("Hello!")
# Anthropic
ai = AI(provider="anthropic", model="claude-sonnet-4-20250514")
response = await ai.chat("Hello!")
# Local Ollama
ai = AI(provider="ollama", model="llama3")
response = await ai.chat("Hello!")
### Synchronous Usage
from universal_ai import AI
ai = AI(provider="openai", model="gpt-4o")
# Synchronous wrappers for scripts/notebooks
response = ai.chat_sync("Hello!")
print(response.content)
# Sync streaming (returns full text)
text = ai.stream_sync("Tell me a joke")
print(text)
## Tool Calling
Define tools with the `@tool` decorator and let the AI use them:
from universal_ai import AI, tool
@tool
def get_weather(city: str, unit: str = "celsius") -> str:
"""Get current weather for a city."""
# In a real app, call a weather API
return f"Weather in {city}: 22°{unit[0].upper()}, sunny"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression."""
return str(eval(expression))
ai = AI(provider="openai", model="gpt-4o")
# The AI will automatically call your tools
response = await ai.chat(
"What's the weather in Paris? Also calculate 15 * 23.",
tools=[get_weather, calculate]
)
print(response.content)
### Manual Tool Execution
from universal_ai import AI, tool
@tool
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
ai = AI(provider="openai", model="gpt-4o")
ai.register_tool(search)
# Tools are auto-executed in the tool loop
response = await ai.chat("Search for Python tutorials")
## Conversations
Multi-turn conversations with automatic history management:
from universal_ai import AI
ai = AI(provider="openai", model="gpt-4o")
# Create a conversation
conv = ai.conversation(
system_prompt="You are a helpful cooking assistant.",
max_turns=20
)
# Send messages
response = await conv.send(message="What should I cook for dinner?")
print(response.content)
response = await conv.send(message="Can you give me a recipe?")
print(response.content)
# Access history
print(f"Turn count: {conv.turn_count}")
print(f"Messages: {len(conv.history)}")
# Reset
conv.reset()
## Configuration
### Environment Variables
# Provider selection
export UNIVERSALAI_PROVIDER=openai
export UNIVERSALAI_MODEL=gpt-4o
# API keys (provider-specific)
export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-ant-...
export GEMINI_API_KEY=...
export GROQ_API_KEY=gsk_...
export MISTRAL_API_KEY=...
export OPENROUTER_API_KEY=sk-or-...
export HF_API_KEY=hf_...
# Azure OpenAI
export AZURE_OPENAI_API_KEY=...
export AZURE_OPENAI_API_BASE=https://your-resource.openai.azure.com
export AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4o
# Ollama (local)
export OLLAMA_HOST=http://localhost:11434
### Config File
# ~/.config/universalai/config.yaml
provider: openai
model: gpt-4o
temperature: 0.7
max_tokens: 4096
timeout: 30
max_retries: 3
auto_truncate: true
fallback_providers:
- anthropic
- gemini
provider_api_keys:
openai: sk-...
anthropic: sk-ant-...
### Programmatic Configuration
from universal_ai import AI, Config
config = Config(
provider="openai",
model="gpt-4o",
temperature=0.7,
max_tokens=4096,
timeout=30,
max_retries=3,
auto_truncate=True,
provider_api_keys={
"openai": "sk-...",
"anthropic": "sk-ant-...",
}
)
ai = AI(config=config)
## Middleware
Add resilience and observability to your requests:
from universal_ai import AI
from universal_ai.middleware import (
RetryMiddleware,
CacheMiddleware,
RateLimitMiddleware,
CircuitBreakerMiddleware,
CostTrackingMiddleware,
LoggingMiddleware,
)
ai = AI(provider="openai", model="gpt-4o")
# Add middleware in order (executed top to bottom)
ai.add_middleware(LoggingMiddleware())
ai.add_middleware(CostTrackingMiddleware())
ai.add_middleware(RetryMiddleware(max_retries=3, base_delay=1.0))
ai.add_middleware(CacheMiddleware(ttl=300))
ai.add_middleware(RateLimitMiddleware(requests_per_minute=60))
ai.add_middleware(CircuitBreakerMiddleware(failure_threshold=5))
# All requests now go through the middleware pipeline
response = await ai.chat("Hello!")
### Middleware Reference
Middleware | Purpose | Key Options
---|---|---
`RetryMiddleware` | Retry failed requests | `max_retries`, `base_delay`, `max_delay`, `jitter`
`CacheMiddleware` | Cache responses | `ttl`, `backend` (memory/sqlite/redis)
`RateLimitMiddleware` | Limit request rate | `requests_per_minute`, `burst`
`CircuitBreakerMiddleware` | Stop cascading failures | `failure_threshold`, `recovery_timeout`
`CostTrackingMiddleware` | Track API costs | —
`LoggingMiddleware` | Log requests/responses | `log_level`
## Routing Strategies
Automatically select the best provider:
from universal_ai import AI
from universal_ai.router import (
Router,
FallbackStrategy,
RoundRobinStrategy,
LowestLatencyStrategy,
LowestCostStrategy,
)
# Configure fallback in config
config = Config(
provider="openai",
fallback_providers=["anthropic", "gemini"]
)
ai = AI(config=config)
# Or use router directly
router = Router(
providers=["openai", "anthropic", "gemini"],
strategy=FallbackStrategy()
)
### Strategy Options
Strategy | Behavior
---|---
`FallbackStrategy` | Try first provider, failover to next on error
`RoundRobinStrategy` | Distribute requests evenly across providers
`LowestLatencyStrategy` | Always use the fastest responding provider
`LowestCostStrategy` | Always use the cheapest provider
## RAG (Retrieval-Augmented Generation)
Build knowledge-base powered chat:
from universal_ai import AI
from universal_ai.rag import RAG, TextLoader, DirectoryLoader
# Initialize RAG
rag = RAG(chunk_size=500, chunk_overlap=50, top_k=3)
# Add content
rag.add_text("Python is a high-level programming language...")
rag.add_document(Document(content="...", source="docs.txt"))
rag.add_folder("./knowledge_base")
rag.add_url("https://example.com/article.txt")
rag.add_github("owner/repo")
# Search
chunks = await rag.search("What is Python?")
for chunk in chunks:
print(f"Score: {chunk.content[:50]}...")
# Use with AI
ai = AI(provider="openai", model="gpt-4o")
augmented_request = await rag.augment_request(chat_request)
response = await ai.chat(augmented_request)
## Audio & Image
### Transcription (Whisper)
ai = AI(provider="openai", model="gpt-4o")
# Transcribe audio file
text = await ai.transcribe("audio.mp3")
print(text)
# Transcribe from bytes
text = await ai.transcribe(audio_bytes)
### Text-to-Speech
# Generate speech
audio_bytes = await ai.speak("Hello, world!", voice="alloy")
with open("output.mp3", "wb") as f:
f.write(audio_bytes)
### Image Generation
# Generate image
urls = await ai.image("A sunset over mountains", size="1024x1024")
print(urls[0]) # URL to generated image
## CLI Usage
UniversalAI includes a full-featured command-line tool:
# Chat interactively
uai chat
# Chat with specific provider
uai chat -p openai -m gpt-4o
# Send a single message
uai chat "What is machine learning?"
# List available providers
uai providers
# Run diagnostics
uai doctor
# Manage configuration
uai config show
uai config set provider openai
uai config set-api-key openai
# Benchmark providers
uai benchmark --iterations 10
# Start local API server
uai serve --port 8000
## Provider Details
### OpenAI
ai = AI(provider="openai", model="gpt-4o")
# Features: Chat, Streaming, Vision, Tools, Embeddings, Audio, Image Gen
# Requires: OPENAI_API_KEY
### Anthropic
ai = AI(provider="anthropic", model="claude-sonnet-4-20250514")
# Features: Chat, Streaming, Vision, Tools
# Requires: ANTHROPIC_API_KEY
### Gemini
ai = AI(provider="gemini", model="gemini-2.0-flash")
# Features: Chat, Streaming, Vision, Tools, Embeddings
# Requires: GEMINI_API_KEY
### Ollama (Local)
ai = AI(provider="ollama", model="llama3")
# Features: Chat, Streaming, Embeddings
# Requires: Ollama running locally
# Install: https://ollama.ai
### Groq
ai = AI(provider="groq", model="llama-3.1-70b-versatile")
# Features: Chat, Streaming, Tools
# Requires: GROQ_API_KEY
### Mistral
ai = AI(provider="mistral", model="mistral-large-latest")
# Features: Chat, Streaming, Tools, Embeddings
# Requires: MISTRAL_API_KEY
### OpenRouter
ai = AI(provider="openrouter", model="openai/gpt-4o")
# Features: Chat, Streaming, Vision, Tools, Embeddings
# Requires: OPENROUTER_API_KEY
### HuggingFace
ai = AI(provider="huggingface", model="meta-llama/Llama-2-7b-chat-hf")
# Features: Chat, Streaming, Embeddings
# Requires: HF_API_KEY
### Azure OpenAI
ai = AI(provider="azure", model="gpt-4o")
# Features: Chat, Streaming, Vision, Tools, Embeddings
# Requires: AZURE_OPENAI_API_KEY, AZURE_OPENAI_API_BASE
## Error Handling
from universal_ai import AI
from universal_ai.exceptions import (
AuthenticationError,
RateLimitError,
ContextWindowExceededError,
ProviderError,
TimeoutError,
)
ai = AI(provider="openai", model="gpt-4o")
try:
response = await ai.chat("Hello!")
except AuthenticationError as e:
print(f"Invalid API key: {e}")
except RateLimitError as e:
print(f"Rate limited, retry after: {e.retry_after}s")
except ContextWindowExceededError as e:
print(f"Context too long: {e.estimated_tokens} > {e.context_window}")
except ProviderError as e:
print(f"Provider error: {e}")
except TimeoutError:
print("Request timed out")
## Context Window Safety
UniversalAI validates that messages fit within the provider's context window:
from universal_ai import AI, Config
# Option 1: Raise error if too long (default)
config = Config(auto_truncate=False)
ai = AI(config=config)
# Option 2: Auto-truncate to fit
config = Config(auto_truncate=True)
ai = AI(config=config)
## Cost Estimation
ai = AI(provider="openai", model="gpt-4o")
# Estimate cost before sending
estimated_cost = ai.estimate_cost("Hello, world!")
print(f"Estimated cost: ${estimated_cost:.6f}")
# Track actual costs with middleware
from universal_ai.middleware import CostTrackingMiddleware
cost_middleware = CostTrackingMiddleware()
ai.add_middleware(cost_middleware)
response = await ai.chat("Hello!")
print(f"Actual cost: ${response.usage.estimated_cost:.6f}")
print(f"Total cost: ${cost_middleware.total_cost:.6f}")
## Sync Wrappers
For scripts and notebooks where you can't use `async`:
Async Method | Sync Wrapper
---|---
`await ai.chat(...)` | `ai.chat_sync(...)`
`async for chunk in ai.stream(...)` | `ai.stream_sync(...)`
`await ai.embed(...)` | `ai.embed_sync(...)`
`await ai.chat_with_tools(...)` | `ai.chat_with_tools_sync(...)`
`await ai.chat_json(...)` | `ai.chat_json_sync(...)`
## Examples
See the examples/ directory for complete working examples:
* `basic_chat.py` - Simple chat usage
* `streaming.py` - Real-time streaming
* `tool_calling.py` - Tool definition and execution
* `middleware_demo.py` - Middleware configuration
* `rag_demo.py` - RAG with document loading
* `multi_provider.py` - Provider switching
## Contributing
We welcome contributions! Please see CONTRIBUTING.md for guidelines.
# Clone the repo
git clone https://github.com/6t9xstar/universal-ai.git
cd universal-ai
# Install dev dependencies
pip install -e ".[dev]"
# Run tests
pytest
# Run linting
ruff check .
# Run type checking
mypy .
## License
MIT License - see LICENSE for details.