Ir para o conteúdo

Quickstart

Every provider implements the same IChatCompletionProvider port, so the only thing that changes between OpenAI, Anthropic, Gemini, Bedrock, OpenRouter and Azure is the value you pass to the factory.

Pick a provider, ask a question

quickstart.py
import asyncio
import os

from apogee_ai_providers import (
    ChatMessage,
    ChatRequest,
    MessageRole,
    Provider,
    ProviderCredentials,
    ProviderFactory,
)


async def main() -> None:
    credentials = ProviderCredentials(
        api_key=os.environ["AI_PROVIDER_KEY_ANTHROPIC"],
        timeout=30.0,
    )
    provider = ProviderFactory.build(Provider.ANTHROPIC, credentials)

    request = ChatRequest(
        model="claude-haiku-4-5-20251001",
        messages=[
            ChatMessage(role=MessageRole.SYSTEM, content="Answer in one sentence."),
            ChatMessage(role=MessageRole.USER, content="What is RAG?"),
        ],
        max_tokens=128,
        temperature=0.2,
    )

    response = await provider.complete(request)
    print(response.choices[0].message.content)
    print("usage:", response.usage)

    await provider.aclose()


asyncio.run(main())

Swap providers without touching application code

Only the Provider value and the model id change — ChatRequest, the response shape and the error types stay identical.

Python
provider = ProviderFactory.build(Provider.ANTHROPIC, credentials)
model = "claude-haiku-4-5-20251001"
Python
provider = ProviderFactory.build(Provider.OPENAI, credentials)
model = "gpt-4o-mini"
Python
provider = ProviderFactory.build(Provider.OPENROUTER, credentials)
model = "anthropic/claude-haiku-4.5"
Python
provider = ProviderFactory.build(Provider.GEMINI, credentials)
model = "gemini-2.5-pro"

ProviderFactory.build() also accepts a plain string ("anthropic"), which is what the CLI and environment-driven configuration use.

Stream responses

Set stream=True on the request and iterate. Each ChatChunk carries a delta; the final one carries finish_reason.

streaming.py
request = ChatRequest(
    model="claude-haiku-4-5-20251001",
    messages=[ChatMessage(role=MessageRole.USER, content="Count from 1 to 5.")],
    max_tokens=64,
    temperature=0.0,
    stream=True,
)

async for chunk in provider.stream(request):
    if chunk.delta:
        print(chunk.delta, end="", flush=True)

Close the client

Providers hold an HTTP connection pool. Call await provider.aclose() when you are done, or build them inside a DI container that owns the lifecycle.