LLM Providers & Failovers¶

Agentomatic abstracts LLM access behind a provider layer that handles instantiation, caching, failover chains, retry logic, and structured output binding. You never construct LangChain chat-model objects directly — instead you call one of the get_llm* helpers and let the framework manage the rest.

graph LR
    A["Agent / Pipeline"] --> B["get_llm_for_agent()"]
    B --> C{"Resolution order"}
    C -->|1| D["Agent llm_config"]
    C -->|2| E["Stack profile"]
    C -->|3| F["'default' profile"]
    C -->|4| G["Global get_llm() singleton"]
    G --> H["Primary provider"]
    H -->|failure| I["Fallback chain"]
    I -->|all fail| J["Dummy LLM"]

Supported Providers¶

Provider Key	LangChain Class	Package	Streaming	Status
`ollama`	`ChatOllama`	`langchain-ollama`	✅	Stable
`openai`	`ChatOpenAI`	`langchain-openai`	✅	Stable
`openai_compatible`	`ChatOpenAI`	`langchain-openai`	✅	Stable — a `base_url` is required
`azure`	`AzureChatOpenAI`	`langchain-openai`	✅	Stable
`vertex`	`ChatVertexAI`	`langchain-google-vertexai`	✅	Stable
`google_genai`	—	`langchain-google-genai`	—	Not available
`dummy`	`FakeListChatModel`	`langchain-core`	❌	Testing

Google GenAI

The google_genai provider key is accepted by the stack manager configuration but is not yet wired into the low-level _build_llm factory. Use vertex for production Google model access today.

Provider Setup¶

OpenAIOllamaOpenAI-compatibleAzure OpenAIGoogle Vertex AI

# Required environment variables
export OPENAI_API_KEY="sk-..."

from agentomatic.providers import get_llm

llm = get_llm(
    provider="openai",
    model="gpt-4",          # or "gpt-4o", "gpt-4o-mini", etc.
    temperature=0.3,
    api_key="sk-...",       # falls back to OPENAI_API_KEY env var
)

Model name format: OpenAI model slugs — gpt-4, gpt-4o, gpt-4o-mini, gpt-3.5-turbo.

# Start the Ollama server (default: http://localhost:11434)
ollama serve
ollama pull llama3

from agentomatic.providers import get_llm

llm = get_llm(
    provider="ollama",
    model="llama3",
    base_url="http://localhost:11434",   # default
    temperature=0.7,
)

Model name format: Ollama model tags — llama3, mistral, codellama:13b, phi3:medium.

Use this for any OpenAI-API-compatible model service, including oMLX, llama.cpp, vLLM, LM Studio, Groq, Together, or a gateway. Unlike openai, this provider requires an explicit base_url (or its api_base alias).

from agentomatic.providers import get_llm

llm = get_llm(
    provider="openai_compatible",
    model="Qwen3.5-9B-MLX-4bit",
    base_url="http://host.docker.internal:8000/v1",
    api_key="local-service-key",
    temperature=0.2,
)

Install langchain-openai via pip install "agentomatic[openai]".

# Required environment variables
export AZURE_OPENAI_API_KEY="..."
export AZURE_OPENAI_ENDPOINT="https://<resource>.openai.azure.com/"

from agentomatic.providers import get_llm

llm = get_llm(
    provider="azure",
    api_key="...",
    azure_endpoint="https://<resource>.openai.azure.com/",
    azure_deployment="gpt-4o",
    api_version="2024-02-15-preview",   # default
    temperature=0.2,
)

Model name format: Your Azure deployment name, not the underlying OpenAI model id.

# Authenticate with Google Cloud
gcloud auth application-default login

from agentomatic.providers import get_llm

llm = get_llm(
    provider="vertex",
    model="gemini-1.5-pro",
    project="my-gcp-project",
    location="us-central1",       # default
    temperature=0.4,
)

Model name format: Vertex AI model ids — gemini-1.5-pro, gemini-1.5-flash, gemini-2.0-flash.

Custom Provider Registry¶

Register a provider builder when an internal gateway or vendor does not match a built-in client. Provider names are trimmed and case-insensitive, registrations are thread-safe, and stack profiles can use the registered name directly.

from agentomatic import register_llm_provider
from agentomatic.providers import OAuth2ClientCredentialsTokenProvider

tokens = OAuth2ClientCredentialsTokenProvider(
    token_url="https://identity.example.com/oauth/token",
    client_id="service-client",
    client_secret="...",
    scope="models.invoke",
    token_params={"audience": "https://models.example.com"},
)


def build_corporate_gateway(**kwargs):
    return CorporateChatModel(
        model=kwargs["model"],
        base_url=kwargs["base_url"],
        token_provider=tokens,
    )


# overwrite=False detects duplicate plugin registration during startup.
register_llm_provider("corporate", build_corporate_gateway, overwrite=False)

llm:
  default:
    provider: corporate
    model: enterprise-chat
    base_url: ${MODEL_GATEWAY_URL}

Use registered_llm_providers() for diagnostics and unregister_llm_provider("corporate") during plugin teardown or tests. Removing a builder does not invalidate already-cached models; call reset_llm() when a live registration is replaced.

OAuth2 token providers¶

OAuth2ClientCredentialsTokenProvider coalesces concurrent refreshes, retries transport errors plus HTTP 429/5xx, validates token response shape and expiry, and never includes the identity-provider response body in raised validation errors. It supports both OAuth client authentication methods:

tokens = OAuth2ClientCredentialsTokenProvider(
    token_url="https://identity.example.com/oauth/token",
    client_id="service-client",
    client_secret="...",
    client_auth_method="client_secret_basic",  # or client_secret_post
    refresh_margin_seconds=30,
)

token = tokens.get_token()
token_in_async_code = await tokens.aget_token()  # does not block the event loop
tokens.invalidate()                              # force refresh after a 401

Non-loopback token endpoints require HTTPS by default. http://localhost and loopback IPs remain available for local testing; the explicit allow_insecure_http=True escape hatch should only be used in a controlled development network. StaticTokenProvider exposes the same sync and async API when the gateway supplies a long-lived bearer token directly.

Environment Variables¶

Variable	Provider	Description
`OPENAI_API_KEY`	`openai`	OpenAI API key
service-specific API key	`openai_compatible`	Pass as `api_key` with the required `base_url`
`AZURE_OPENAI_API_KEY`	`azure`	Azure OpenAI resource key
`AZURE_OPENAI_ENDPOINT`	`azure`	Azure resource endpoint URL
`GOOGLE_CLOUD_PROJECT`	`vertex`	GCP project id
`GOOGLE_APPLICATION_CREDENTIALS`	`vertex`	Path to service-account JSON

Failover Chain Configuration¶

Configure an ordered fallback chain so that if the primary model times out, disconnects, rate-limits, or returns an empty response, Agentomatic retries the next model and logs which one succeeded.

Python API¶

from agentomatic.providers import get_llm

llm = get_llm(
    provider="openai",
    model="gpt-4o",
    fallbacks=[
        "ollama",                                      # provider slug (inherits kwargs)
        "openai/gpt-4o-mini",                          # provider/model shorthand
        {"provider": "vertex", "model": "gemini-2.0-flash"},  # full inline spec
    ],
    fallback_on=["timeout", "connection", "rate_limit", "empty_response"],
    temperature=0.3,
)

Stack YAML¶

llm:
  default:
    provider: openai
    model: gpt-4o
    api_key: ${OPENAI_API_KEY}
    fallbacks:
      - fast                                          # named profile in this stack
      - provider: ollama
        model: mistral:7b
        base_url: http://localhost:11434
    fallback_on:
      - timeout
      - connection
      - rate_limit
      - empty_response
  fast:
    provider: openai
    model: gpt-4o-mini
    api_key: ${OPENAI_API_KEY}

Omitting fallbacks keeps the previous single-model behaviour unchanged.

Under the hood Agentomatic wraps the chain in FallbackLLM. Each step failure calls record_failover(primary, next, error); when a fallback answers, a log line records which model succeeded.

Trigger	Default	Matches
`timeout`	✅	`TimeoutError`, messages containing timeout/deadline
`connection`	✅	`ConnectionError` / `OSError`, connection refused, DNS failures
`rate_limit`	✅	429 / rate-limit style errors
`empty_response`	✅	Blank / whitespace-only content
`any_error`	❌	Every exception (opt-in)

Settings default

Platform setting llm.fallback_on (env nested key LLM__FALLBACK_ON) documents the same trigger list. Per-profile fallback_on in stack YAML overrides when present; otherwise the library defaults above apply.

Last-resort dummy

If the primary provider fails to build at construction time, the factory still falls back to a FakeListChatModel (dummy) so import / startup does not crash. Runtime invoke failures use the configured fallback chain instead.

graph LR
    P["Primary: openai/gpt-4o"] -->|timeout / 429 / empty| F1["Fallback: openai/gpt-4o-mini"]
    F1 -->|failure| F2["Fallback: ollama/mistral"]
    F2 -->|failure| E["Raise last error"]

Retry & Timeout Settings¶

For transient failures within a single provider, use invoke_with_retry which wraps llm.ainvoke with exponential back-off:

from agentomatic.providers import get_llm, invoke_with_retry

llm = get_llm(provider="openai", model="gpt-4o")

response = await invoke_with_retry(
    llm,
    messages=[{"role": "user", "content": "Hello!"}],
    max_retries=3,       # default
    retry_delay=1.0,     # base delay in seconds (default)
)

The delay doubles after each attempt: delay × 2^attempt.

For custom provider setup, token exchange, and other ordinary callables, use the validated generic retry helpers. RetryConfig rejects invalid attempt, delay, multiplier, jitter, and exception settings at construction time.

from agentomatic.providers import RetryConfig, async_retry_call, retry_call

policy = RetryConfig(
    max_attempts=4,
    base_delay=0.5,
    max_delay=8,
    jitter=0.2,
    retryable_exceptions=(ConnectionError, TimeoutError),
)

result = retry_call(fetch_sync, config=policy)
result = await async_retry_call(fetch_async, config=policy)

The async helper uses asyncio.sleep, accepts a sync or async on_retry hook, and propagates task cancellation immediately.

By default invoke_with_retry strips thinking / reasoning from .content (Qwen3, DeepSeek-R1, tagged <think> blocks, etc.) and stores it on additional_kwargs["thinking"] for debugging. Pass strip_thinking=False to keep the raw payload.

Attempt	Wait before retry
0	1.0 s
1	2.0 s
2	4.0 s

Failover vs. Retry

Retry re-attempts the same provider (good for transient 429 / 503 errors). Failover switches to a different provider (good for persistent outages). Combine both for maximum resilience.

Thinking / reasoning models¶

Modern instruct models (Qwen3.5, Gemma reasoning builds, oMLX servers) may emit chain-of-thought separately from the final answer. Agentomatic normalizes this so agents never dump thinking into user-facing text or JSON parsers by default.

Helper	Purpose
`message_text(result)`	Final answer only
`message_thinking(result)`	Reasoning trail (may be empty)
`llm_result_metadata(result)`	Compact debug dict (`has_thinking`, …)
`strip_thinking_for_json(text)`	Answer text safe for `json.loads`
`astream_with_thinking(llm, msgs)`	Yields `{type: thinking\\|answer\\|done}`
`invoke_with_retry(..., strip_thinking=True)`	Normalize after each invoke

from agentomatic.providers import message_text, message_thinking, invoke_with_retry

result = await invoke_with_retry(llm, messages)
answer = message_text(result)                 # safe for FR UX / JSON
thinking = message_thinking(result)           # optional debug
# or result.additional_kwargs.get("thinking")

Stack `extra:` knobs (OpenAI-compatible / oMLX)¶

Vendor-specific fields on a stack LLM profile are forwarded via extra_body / model_kwargs without breaking other providers:

llm:
  structured:
    provider: openai
    model: Qwen3.5-9B-MLX-4bit
    base_url: ${AI_LLM_BASE_URL}
    api_key: ${AI_LLM_API_KEY}
    extra:
      enable_thinking: false          # recommended for JSON agents
      chat_template_kwargs:
        enable_thinking: false
      # response_format: { type: json_object }
      # default_headers: { X-Custom: "…" }
      # extra_body: { … }             # any other server fields

Set enable_thinking: true only when you intentionally want reasoning in Studio / debug metadata. Structured-output fallback still strips thinking before Pydantic validation.

Named LLM Instances¶

get_named_llm lets you maintain multiple cached LLM instances under human-readable names — for example a fast model for routing and a powerful model for complex reasoning:

from agentomatic.providers import get_named_llm

fast  = get_named_llm("fast",  provider="openai", model="gpt-4o-mini", temperature=0.1)
judge = get_named_llm("judge", provider="openai", model="gpt-4o",      temperature=0.0)

Subsequent calls with the same name return the cached instance (thread-safe). Call reset_llm() to clear all named instances.

Agent-Specific LLMs¶

get_llm_for_agent resolves the LLM for a particular agent using a strict priority order:

from agentomatic.providers import get_llm_for_agent

llm = get_llm_for_agent(
    agent_name="planner",
    role="default",           # or "judge", "fast", etc.
    stack_manager=manager,    # optional StackManager instance
)

Resolution order:

Agent's own llm_config — defined in the agent manifest.
Stack profile — the named profile matching role in the active stack.
"default" profile — the stack's default profile.
Global singleton — get_llm() with default settings.

Note

This layered resolution means you can override LLM settings at any granularity — per-agent, per-stack, or globally — without touching other agents' configuration.

Structured Output¶

get_structured_llm binds a Pydantic model to the LLM so every response is automatically parsed and validated:

from pydantic import BaseModel
from agentomatic.providers import get_structured_llm


class SentimentResult(BaseModel):
    label: str
    score: float


llm = get_structured_llm(
    response_model=SentimentResult,
    provider="openai",
    model="gpt-4o-mini",
)

result = await llm.ainvoke("Analyze: 'I love this product!'")
# result -> SentimentResult(label="positive", score=0.97)

Fallback wrapper

If the underlying provider does not support native structured output, Agentomatic transparently wraps it with a StructuredOutputFallbackWrapper that prompts the model for JSON and parses the response against your Pydantic schema.

Custom LLM Injection¶

Every get_llm* function accepts an instance= keyword argument so you can bypass the factory and inject a pre-built LLM directly. This is the recommended approach for custom models, fine-tuned endpoints, or any LLM that doesn't fit the built-in provider system.

Inject a Global Custom LLM¶

The simplest way is set_llm() — it stores your model as the global singleton used by all agents, pipelines, and the platform:

from agentomatic.providers import set_llm, get_llm

# Any object with ainvoke/invoke works (LangChain models, etc.)
from langchain_openai import ChatOpenAI

set_llm(ChatOpenAI(model="gpt-4o", temperature=0.2))

# Now every get_llm() call returns your custom model
llm = get_llm()  # → your ChatOpenAI instance

Or equivalently via get_llm(instance=...):

from agentomatic.providers import get_llm

llm = get_llm(instance=ChatOpenAI(model="gpt-4o"))

Named Custom Instances¶

Use get_named_llm(instance=...) for per-role models:

from agentomatic.providers import get_named_llm

get_named_llm("judge", instance=my_judge_model)
get_named_llm("fast",  instance=my_fast_model)

# Later lookups return the cached instance
judge = get_named_llm("judge")  # → my_judge_model

Structured Output with Custom LLMs¶

from agentomatic.providers import get_structured_llm

structured = get_structured_llm(
    SentimentResult,
    instance=my_custom_llm,   # bypasses the factory
)

Async / Sync Callables¶

Any async or sync callable matching (prompt, *, system_prompt=None) → str works everywhere:

from agentomatic.providers import set_llm

# Async callable
async def my_llm(prompt: str, *, system_prompt: str | None = None) -> str:
    return await my_inference_api(prompt, system=system_prompt)

set_llm(my_llm)

# Sync callable (run in executor automatically)
def my_sync_llm(prompt: str, *, system_prompt: str | None = None) -> str:
    return requests.post("https://my-api/v1/chat", json={"prompt": prompt}).text

set_llm(my_sync_llm)

LLMSpec — Custom Models in the Optimize Pipeline¶

The optimize module (prompt optimizers, metrics, synthesizers) uses the LLMSpec type — a union of str | LLMCallable — so you can pass custom callables to every optimization component:

from agentomatic.optimize import (
    LLMSpec,
    LLMCallable,
    call_llm,
    call_llm_json,
    PromptOptimizer,
    PromptFitter,
)

# Custom callable
async def my_eval_llm(prompt: str, *, system_prompt: str | None = None) -> str:
    return await my_api.complete(prompt, system=system_prompt)

# Use in optimizer
optimizer = PromptOptimizer(
    agent="my_agent",
    llm=my_eval_llm,         # Custom callable ✓
    rewrite_llm="openai/gpt-4o",  # String spec ✓
)

# Use directly
text = await call_llm(my_eval_llm, "Hello")
data = await call_llm_json(my_eval_llm, "Return {\"ok\": true}")

Graceful degradation

If your callable raises an exception, call_llm() catches it and returns an empty string "" with a warning log — matching the resilience behavior of the string-model path. Your optimisation pipeline will never crash due to a transient LLM failure.

Embedding Providers¶

Embeddings follow the same singleton pattern via get_embeddings:

from agentomatic.providers.embeddings import get_embeddings, reset_embeddings

embeddings = get_embeddings(provider="ollama", model="nomic-embed-text")
vectors = embeddings.embed_documents(["Hello world", "Goodbye world"])

Provider Key	Class	Notes
`ollama`	`OllamaEmbeddings`	Requires running Ollama
`openai`	`OpenAIEmbeddings`	Requires `langchain-openai` and an API key/base URL as appropriate
`azure_openai`	`AzureOpenAIEmbeddings`	Azure deployment plus endpoint and API version
`hash`	`HashEmbedder`	Local deterministic hash vectors; default, no network dependency
`dummy`	`DeterministicFakeEmbedding`	Test-only deterministic vectors (falls back to `HashEmbedder` without LangChain)
registered name	Custom builder	Add with `register_embedding_provider(name, builder)`

Call reset_embeddings() to clear the cached instance.

Streaming Support¶

Agentomatic exposes framework streaming through an SSE (Server-Sent Events) endpoint. Use the normal agent invocation request shape; provider selection is configured by the agent, stack, or server, not by a request-body config object:

POST /api/v1/{agent}/invoke/stream
Content-Type: application/json

{"query": "Explain quantum computing"}

The stream format and event richness depend on the registered agent framework; see the Studio guide and API reference for the event and endpoint contracts.

Note

The dummy provider does not support streaming. Use a real provider when testing streaming behaviour.

Failover Telemetry¶

Agentomatic tracks every failover event so you can monitor provider reliability in production:

from agentomatic.providers import (
    get_failover_count,
    record_failover,
    reset_llm,
)

# Manually record a failover (usually called internally)
record_failover(
    primary="openai",
    fallback="ollama",
    error="RateLimitError: 429",
)

# Query cumulative failover count
count = get_failover_count()   # -> int

# Reset everything (instances + telemetry)
reset_llm()

Each record_failover call emits a loguru warning with the primary provider, the fallback that took over, and the error message — making it easy to set up alerts in your logging pipeline.

Troubleshooting¶

API key not found — AuthenticationError

Verify the environment variable is set in the shell where your process runs. Common mistakes:

Setting the variable in .bashrc but running the process in a different shell (e.g. a VS Code terminal that sources .zshrc).
Passing api_key as a kwarg with a typo.
Using OPENAI_API_KEY when the provider is azure (needs AZURE_OPENAI_API_KEY).

echo $OPENAI_API_KEY   # should print your key

Model not available — NotFoundError

Each provider uses a different model naming scheme:

OpenAI: gpt-4o, not gpt4o or gpt-4-o.
Azure: Use your deployment name, not the OpenAI model id.
Ollama: Run ollama list to see locally pulled models.
Vertex: Ensure the model is enabled in your GCP project.

Timeout errors — TimeoutError / ReadTimeout

Increase max_retries and retry_delay in invoke_with_retry.
For Ollama, ensure the server is running (ollama serve) and the model is fully loaded (first invocation can take 30 s+).
For cloud providers, check your network / proxy configuration.

Rate limited — 429 Too Many Requests

invoke_with_retry handles transient 429s automatically via exponential back-off.

For sustained rate-limiting, add a cheaper provider as a fallback:

llm = get_llm(provider="openai", model="gpt-4o", fallbacks=["ollama"])

Consider using get_named_llm to route high-volume, low-stakes calls to a faster / cheaper model.

Topic	Page
Agent Manifest & Config	Agent Structure
Stack Manager	Stacks & Profiles
Prompt Engineering	Prompt Management
Platform Settings	Configuration
API Reference — Providers	API Reference