Providers

# Primary providers
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export CEREBRAS_API_KEY="..."
 
# OpenRouter (multi-model gateway)
export OPENROUTER_API_KEY="sk-or-..."
export OPENROUTER_HTTP_REFERER="https://your-app.com"
export OPENROUTER_X_TITLE="Your App Name"
 
# Anthropic Subscription Proxy (flat-rate Claude access)
export ANTHROPIC_SUB_PROXY_URL="http://localhost:5005/v1"
export ANTHROPIC_SUB_API_KEY="sk-proxy"
 
# Proxy providers
export YYDS_API_KEY="..."
export YYDS_API_BASE="https://api.yyds.example.com/v1"
 
# Local providers
export OLLAMA_HOST="http://localhost:11434"
 
# Image/video providers
export REPLICATE_API_TOKEN="..."

# openai.yaml
openai:
  enabled: true
  api_key: ${OPENAI_API_KEY}
  api_base_url: https://api.openai.com/v1
  organization: ${OPENAI_ORG_ID}  # Optional
  default_model: gpt-4o-mini
  temperature: 0.7
  max_tokens: 4096
  rate_limit_rpm: 500
  rate_limit_tpm: 200000

from isa_model.inference.ai_factory import AIFactory
 
factory = AIFactory()
 
# Standard chat
llm = factory.get_service("text", "openai", "gpt-4o-mini")
 
# Reasoning model
llm_reasoning = factory.get_service("text", "openai", "o4-mini")
response = await llm_reasoning.ainvoke(
    "Solve this math problem step by step",
    show_reasoning=True
)
 
# Deep research (requires o4-deep-research)
llm_research = factory.get_service("text", "openai", "o4-deep-research")
result = await llm_research.invoke(
    "Research the latest developments in quantum computing",
    task="deep_research"
)

# Claude models
llm = factory.get_service("text", "anthropic", "claude-3-sonnet")
 
# Long context (200K tokens)
llm_long = factory.get_service("text", "anthropic", "claude-3-opus")
response = await llm_long.ainvoke(very_long_document)

# Ultra-fast Llama inference
llm = factory.get_service("text", "cerebras", "llama-3.3-70b")
 
# Great for real-time applications
async for token in llm.astream("Hello!"):
    print(token, end="")  # Very fast token generation

# Access 200+ models from a single API
llm = factory.get_service("text", "openrouter", "anthropic/claude-sonnet-4-6")
 
# Reasoning models detected automatically (deepseek-r1, o1/o3, :thinking suffix)
llm_reasoning = factory.get_service("text", "openrouter", "deepseek/deepseek-r1")
response = await llm_reasoning.ainvoke(
    "Solve this step by step",
    show_reasoning=True,  # Include reasoning tokens in response
)
 
# Token usage includes reasoning_tokens
usage = llm_reasoning.get_token_usage()
print(f"Reasoning tokens: {usage['reasoning_tokens']}")

# Routes through local CLIProxyAPI (OpenAI-compatible endpoint)
llm = factory.get_service("text", "anthropic-sub", "claude-sonnet-4-20250514")
 
# Built-in rotation retry for subscription account management
# Automatically retries on RateLimitError with 2-second delay
response = await llm.ainvoke("Hello!")
 
info = llm.get_model_info()
print(info["subscription_proxy"])  # True

# Same models as OpenAI, lower cost
llm = factory.get_service("text", "yyds", "gpt-4o-mini")
 
# Useful as fallback provider
response = await llm.ainvoke("Hello!")

# Requires Ollama running locally
llm = factory.get_service("text", "ollama", "llama3")
 
# Private, no data leaves your machine
response = await llm.ainvoke("Analyze this sensitive document")
 
# Custom models
llm_custom = factory.get_service("text", "ollama", "my-finetuned-model")

# Failover is automatic when using the /invoke API
# Configure in inference.py:
 
FALLBACK_PROVIDERS = {
    "openai": ["yyds", "cerebras"],    # OpenAI -> YYDS -> Cerebras
    "yyds": ["openai", "cerebras"],
    "cerebras": ["yyds", "openai"],
    "anthropic": ["openai", "yyds"],
}
 
# Model mapping for failover compatibility
FAILOVER_MODEL_MAPPING = {
    "gpt-4.1-mini": {
        "openai": "gpt-4.1-mini",
        "yyds": "gpt-4o-mini",
        "cerebras": "llama-3.3-70b",
    },
}

# my_provider_llm_service.py
from .base_llm_service import BaseLLMService
 
class MyProviderLLMService(BaseLLMService):
    def __init__(self, model_name: str, **kwargs):
        super().__init__("my_provider", model_name, **kwargs)
        # Initialize client
 
    async def ainvoke(self, input_data, **kwargs):
        # Implement inference
        pass
 
    async def astream(self, input_data):
        # Implement streaming
        pass

# my_provider.yaml
my_provider:
  enabled: true
  api_key: ${MY_PROVIDER_API_KEY}
  api_base_url: https://api.myprovider.com/v1
  default_model: my-model-v1

# In ai_factory.py
from .services.llm.my_provider_llm_service import MyProviderLLMService
 
SERVICE_MAP = {
    "text": {
        "my_provider": MyProviderLLMService,
        # ...
    }
}

# Check provider availability
from isa_model.inference.ai_factory import AIFactory
 
factory = AIFactory()
 
# Get health status
health = await factory.check_provider_health("openai")
print(f"OpenAI healthy: {health['status']}")
print(f"Latency: {health['latency_ms']}ms")

# Check all providers
curl http://localhost:8082/api/v1/failover/health
 
# Check specific provider
curl http://localhost:8082/api/v1/failover/circuit/openai

# Provider config
openai:
  rate_limit_rpm: 500    # Requests per minute
  rate_limit_tpm: 200000 # Tokens per minute

# Get usage with cost
usage = llm.get_last_usage_with_cost()
print(f"Provider: {usage['provider']}")
print(f"Model: {usage['model']}")
print(f"Tokens: {usage['total_tokens']}")
print(f"Cost: ${usage['cost_usd']:.4f}")