Skip to main content

Unified streaming interface for multiple LLM providers

Project description

HyperDev LLMRouter

Unified streaming interface for multiple LLM providers with a simple, consistent API.

Features

  • Multi-Provider Support: OpenAI, Claude (Anthropic), Gemini (Google), and OpenRouter
  • Unified Streaming Interface: Same API for all providers
  • Configuration Files: JSON-based config for easy customization
  • Environment Variables: Secure API key management
  • Type-Safe: Built with Pydantic for data validation
  • Memory Efficient: Iterator-based streaming
  • Easy to Extend: Abstract base class for adding new providers
  • Langfuse Integration: Optional monitoring, debugging, and analytics (see Langfuse Guide)

Installation

pip install hyperdev

Or install from source:

git clone https://github.com/HyperDev/llmrouter.git
cd llmrouter
pip install -e .

Quick Start

1. Configure Your Providers

Create config files for the providers you want to use. Example for OpenAI:

{
  "model": "gpt-4",
  "api_key_env_var": "OPENAI_API_KEY",
  "max_tokens": 1024,
  "temperature": 0.7,
  "additional_params": {
    "top_p": 1.0
  }
}

Save as openai_chat_config.json in your config directory.

See config_examples/ for all provider examples.

2. Set Environment Variables

export OPENAI_API_KEY="your-api-key"
export ANTHROPIC_API_KEY="your-api-key"
export GOOGLE_API_KEY="your-api-key"
export OPENROUTER_API_KEY="your-api-key"

3. Use in Code

from pathlib import Path
from hyperdev.llmrouter import chat_stream, LLMProvider

# Stream from OpenAI
for chunk in chat_stream(
    prompt="Explain quantum computing",
    llm_provider=LLMProvider.OPENAI,
    llm_string="gpt-4",
    config_dir=Path("./config")  # Directory with *_chat_config.json files
):
    print(chunk.content, end="", flush=True)

API Reference

chat_stream()

Main function for streaming LLM responses.

Parameters:

  • prompt (str): User message/prompt
  • llm_provider (LLMProvider): Provider enum (OPENAI, CLAUDE, GEMINI, OPENROUTER)
  • llm_string (str): Model identifier:
    • OpenAI: "gpt-4", "gpt-3.5-turbo"
    • Claude: "claude-sonnet-4-5-20250929"
    • Gemini: "gemini-2.5-flash"
    • OpenRouter: "anthropic/claude-3-sonnet"
  • config_dir (Path, optional): Config directory. Defaults to current working directory

Yields:

  • StreamChunk: Objects with attributes:
    • content (str): Text chunk
    • finish_reason (str | None): Finish reason if stream ended
    • model (str): Model identifier
    • provider (str): Provider name

Raises:

  • ConfigurationError: Invalid config or missing env vars
  • ValidationError: Invalid inputs
  • ProviderError: Provider initialization failed
  • StreamingError: Streaming error occurred

LLMProvider Enum

Supported providers:

  • LLMProvider.OPENAI
  • LLMProvider.CLAUDE
  • LLMProvider.GEMINI
  • LLMProvider.OPENROUTER

StreamChunk Model

Result of streaming operation:

from hyperdev.llmrouter import StreamChunk

chunk = StreamChunk(
    content="Hello world",
    finish_reason=None,
    model="gpt-4",
    provider="openai"
)

Configuration Format

All config files follow this JSON schema:

{
  "model": "model-identifier",
  "api_key_env_var": "ENV_VAR_NAME",
  "max_tokens": 1024,
  "temperature": 0.7,
  "additional_params": {
    "top_p": 1.0
  }
}
  • model: Default model for this provider (can be overridden via llm_string)
  • api_key_env_var: Environment variable containing the API key
  • max_tokens: Maximum output tokens
  • temperature: Sampling temperature (0.0-2.0)
  • additional_params: Provider-specific parameters

Config files are loaded from: {config_dir}/{provider}_chat_config.json

For example:

  • openai_chat_config.json
  • claude_chat_config.json
  • gemini_chat_config.json
  • openrouter_chat_config.json

Examples

Basic Usage

from hyperdev.llmrouter import chat_stream, LLMProvider

for chunk in chat_stream(
    "Explain quantum computing in one sentence",
    LLMProvider.OPENAI,
    "gpt-4"
):
    print(chunk.content, end="", flush=True)

Stream from Multiple Providers

from hyperdev.llmrouter import chat_stream, LLMProvider

providers = [
    (LLMProvider.OPENAI, "gpt-4"),
    (LLMProvider.CLAUDE, "claude-sonnet-4-5-20250929"),
]

for provider, model in providers:
    print(f"\n{provider.value.upper()}:")
    for chunk in chat_stream("Hi!", provider, model):
        print(chunk.content, end="", flush=True)

Error Handling

from hyperdev.llmrouter import chat_stream, LLMProvider
from hyperdev.llmrouter.exceptions import ConfigurationError, ValidationError

try:
    for chunk in chat_stream("Hello", LLMProvider.OPENAI, "gpt-4"):
        print(chunk.content, end="", flush=True)
except ConfigurationError as e:
    print(f"Config error: {e}")
except ValidationError as e:
    print(f"Invalid input: {e}")

Custom Config Directory

from pathlib import Path
from hyperdev.llmrouter import chat_stream, LLMProvider

config_path = Path.home() / ".llmrouter" / "configs"

for chunk in chat_stream(
    "What is AI?",
    LLMProvider.CLAUDE,
    "claude-sonnet-4-5-20250929",
    config_dir=config_path
):
    print(chunk.content, end="", flush=True)

Langfuse Integration (Optional)

Track all your LLM calls with Langfuse for analytics, debugging, and monitoring:

pip install hyperdev[langfuse]

Then enable tracing:

from hyperdev.llmrouter import chat_stream, LLMProvider

for chunk in chat_stream(
    "Explain quantum computing",
    LLMProvider.OPENAI,
    "gpt-4",
    langfuse_public_key="pk_...",
    langfuse_secret_key="sk_...",
    langfuse_user_id="user_123",
    langfuse_session_id="session_456",
    langfuse_metadata={"feature": "essay_generation"},
    langfuse_tags=["production"],
):
    print(chunk.content, end="", flush=True)

This logs:

  • Full input/output of each call
  • Token usage and latency
  • User and session information
  • Custom metadata and tags

See Langfuse Integration Guide for detailed documentation.

Adding New Providers

Extend BaseProvider to add support for new LLM providers:

from hyperdev.llmrouter.providers.base import BaseProvider
from hyperdev.llmrouter.types import ProviderConfig, StreamChunk
from typing import Iterator

class CustomProvider(BaseProvider):
    def stream_chat(self, prompt: str, llm_string: str) -> Iterator[StreamChunk]:
        # Implement streaming logic
        pass

    def validate_model(self, llm_string: str) -> bool:
        # Validate model string
        pass

Then register it in src/hyperdev/llmrouter/providers/__init__.py:

PROVIDER_REGISTRY[LLMProvider.CUSTOM] = CustomProvider

Dependencies

Required:

  • openai>=1.0.0 - OpenAI and OpenRouter
  • anthropic>=0.18.0 - Claude
  • google-genai>=0.2.0 - Gemini
  • pydantic>=2.0.0 - Data validation
  • python-dotenv>=1.0.0 - Environment management

Optional:

  • langfuse>=2.0.0 - LLM tracing and analytics (install with pip install hyperdev[langfuse])

Project Structure

hyperdev_pip/
├── pyproject.toml              # Package configuration
├── README.md                   # This file
├── .gitignore                  # Git ignore rules
├── src/
│   └── hyperdev/
│       ├── __init__.py         # Package init
│       └── llmrouter/
│           ├── __init__.py     # Exports: chat_stream, LLMProvider, etc.
│           ├── enums.py        # LLMProvider enum
│           ├── types.py        # Pydantic models
│           ├── exceptions.py   # Custom exceptions
│           ├── config.py       # ConfigLoader class
│           ├── chat_stream.py  # Main streaming function
│           └── providers/
│               ├── __init__.py # Provider registry
│               ├── base.py     # Abstract base class
│               ├── openai.py   # OpenAI implementation
│               ├── claude.py   # Claude implementation
│               ├── gemini.py   # Gemini implementation
│               └── openrouter.py # OpenRouter implementation
├── config_examples/            # Example config files
├── examples/                   # Usage examples
│   ├── basic_usage.py
│   └── all_providers.py

Testing

Run tests with pytest:

pip install -e ".[dev]"
pytest tests/

License

MIT

Contributing

Contributions welcome! Please ensure:

  • Code follows the project style
  • Tests are included for new features
  • Documentation is updated

Support

For issues and questions:

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

hyperdev-0.1.2.tar.gz (22.0 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

hyperdev-0.1.2-py3-none-any.whl (20.5 kB view details)

Uploaded Python 3

File details

Details for the file hyperdev-0.1.2.tar.gz.

File metadata

  • Download URL: hyperdev-0.1.2.tar.gz
  • Upload date:
  • Size: 22.0 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.12

File hashes

Hashes for hyperdev-0.1.2.tar.gz
Algorithm Hash digest
SHA256 06b9d0cf55a6d4a1102f902c9376995ca8e021b79d1ee2a20109118f708d4b3b
MD5 da5a00c95d5e9ab928c076817a7d8b6f
BLAKE2b-256 e47fd034882ce0d3fec70f725ee2058bd9e61bf1e1187c97a53bb55d240f05ad

See more details on using hashes here.

File details

Details for the file hyperdev-0.1.2-py3-none-any.whl.

File metadata

  • Download URL: hyperdev-0.1.2-py3-none-any.whl
  • Upload date:
  • Size: 20.5 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.12

File hashes

Hashes for hyperdev-0.1.2-py3-none-any.whl
Algorithm Hash digest
SHA256 a4f335c4a8e14a890993873f2fba970ce4fceca26497e89e2698fb18dbe8ca62
MD5 ce06a6da01cfc3870969ba0de06bb027
BLAKE2b-256 01947097324f6ca1a8db4ea740a7866abb12134ddd67a5e741e0d033550bec78

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page