The Requests library for AI — one unified SDK for every LLM provider. Write once, run anywhere.
pip install universal-ai
python
from universal_ai import AI
ai = AI(provider="openai", model="gpt-4o")
response = await ai.chat("What is quantum computing?")
print(response.content)
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:
| 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 |
pip install universal-ai
pip install universal-ai[openai]
pip install universal-ai[anthropic]
pip install universal-ai[gemini]
pip install universal-ai[ollama]
pip install universal-ai[all]
python
import asyncio
from universal_ai import AI
async def main():
ai = AI()
response = await ai.chat("Explain quantum computing in one sentence")
print(response.content)
async for chunk in ai.stream("Write a haiku about programming"):
print(chunk.delta, end="", flush=True)
embed_response = await ai.embed("Hello, world!")
print(f"Embedding dimensions: {len(embed_response.vector)}")
asyncio.run(main())
python
from universal_ai import AI
ai = AI(provider="openai", model="gpt-4o")
response = await ai.chat("Hello!")
ai = AI(provider="anthropic", model="claude-sonnet-4-20250514")
response = await ai.chat("Hello!")
ai = AI(provider="ollama", model="llama3")
response = await ai.chat("Hello!")
python
from universal_ai import AI
ai = AI(provider="openai", model="gpt-4o")
response = ai.chat_sync("Hello!")
print(response.content)
text = ai.stream_sync("Tell me a joke")
print(text)
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."""
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")
response = await ai.chat(
"What's the weather in Paris? Also calculate 15 * 23.",
tools=[get_weather, calculate]
)
print(response.content)
php
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)
response = await ai.chat("Search for Python tutorials")
Multi-turn conversations with automatic history management:
from universal_ai import AI
ai = AI(provider="openai", model="gpt-4o")
conv = ai.conversation(
system_prompt="You are a helpful cooking assistant.",
max_turns=20
)
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)
print(f"Turn count: {conv.turn_count}")
print(f"Messages: {len(conv.history)}")
conv.reset()
export UNIVERSALAI_PROVIDER=openai
export UNIVERSALAI_MODEL=gpt-4o
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_...
export AZURE_OPENAI_API_KEY=...
export AZURE_OPENAI_API_BASE=https://your-resource.openai.azure.com
export AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4o
export OLLAMA_HOST=http://localhost:11434
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-...
python
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)
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")
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))
response = await ai.chat("Hello!")
| 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 |
Automatically select the best provider:
from universal_ai import AI
from universal_ai.router import (
Router,
FallbackStrategy,
RoundRobinStrategy,
LowestLatencyStrategy,
LowestCostStrategy,
)
config = Config(
provider="openai",
fallback_providers=["anthropic", "gemini"]
)
ai = AI(config=config)
router = Router(
providers=["openai", "anthropic", "gemini"],
strategy=FallbackStrategy()
)
| 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 |
Build knowledge-base powered chat:
from universal_ai import AI
from universal_ai.rag import RAG, Text, Directory
rag = RAG(chunk_size=500, chunk_overlap=50, top_k=3)
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")
chunks = await rag.search("What is Python?")
for chunk in chunks:
print(f"Score: {chunk.content[:50]}...")
ai = AI(provider="openai", model="gpt-4o")
augmented_request = await rag.augment_request(chat_request)
response = await ai.chat(augmented_request)
ai = AI(provider="openai", model="gpt-4o")
text = await ai.transcribe("audio.mp3")
print(text)
text = await ai.transcribe(audio_bytes)
audio_bytes = await ai.speak("Hello, world!", voice="alloy")
with open("output.mp3", "wb") as f:
f.write(audio_bytes)
urls = await ai.image("A sunset over mountains", size="1024x1024")
print(urls[0]) # URL to generated image
UniversalAI includes a full-featured command-line tool:
uai chat
uai chat -p openai -m gpt-4o
uai chat "What is machine learning?"
uai providers
uai doctor
uai config show
uai config set provider openai
uai config set-api-key openai
uai benchmark --iterations 10
uai serve --port 8000
ai = AI(provider="openai", model="gpt-4o")
ai = AI(provider="anthropic", model="claude-sonnet-4-20250514")
ai = AI(provider="gemini", model="gemini-2.0-flash")
ai = AI(provider="ollama", model="llama3")
ai = AI(provider="groq", model="llama-3.1-70b-versatile")
ai = AI(provider="mistral", model="mistral-large-latest")
ai = AI(provider="openrouter", model="openai/gpt-4o")
ai = AI(provider="huggingface", model="meta-llama/Llama-2-7b-chat-hf")
ai = AI(provider="azure", model="gpt-4o")
python
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")
UniversalAI validates that messages fit within the provider's context window:
from universal_ai import AI, Config
config = Config(auto_truncate=False)
ai = AI(config=config)
config = Config(auto_truncate=True)
ai = AI(config=config)
ai = AI(provider="openai", model="gpt-4o")
estimated_cost = ai.estimate_cost("Hello, world!")
print(f"Estimated cost: ${estimated_cost:.6f}")
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}")
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(...) |
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
multi_provider.py -
Provider switchingWe welcome contributions! Please see CONTRIBUTING.md for guidelines.
git clone https://github.com/6t9xstar/universal-ai.git
cd universal-ai
pip install -e ".[dev]"
pytest
ruff check .
mypy .
MIT License - see LICENSE for details.