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
azure	AzureChatOpenAI	langchain-openai	✅	Stable
vertex	ChatVertexAI	langchain-google-vertexai	✅	Stable
google_genai	—	langchain-google-genai	✅	Planned
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

OpenAIOllamaAzure 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.

# 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_name="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.

---

Environment Variables

Variable	Provider	Description
OPENAI_API_KEY	openai	OpenAI API key
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. Available in agentomatic >= 1.8.0 (install from git / editable checkout until PyPI publishes that release).

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.

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:

1. Agent's own llm_config — defined in the agent manifest.
2. Stack profile — the named profile matching role in the active stack.
3. "default" profile — the stack's default profile.
4. 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
(other)	DeterministicFakeEmbedding	Deterministic dummy vectors

Call reset_embeddings() to clear the cached instance.

---

Streaming Support

All LangChain providers support token-level streaming via astream_events. Agentomatic exposes this through an SSE (Server-Sent Events) endpoint:

POST /invoke/stream
Content-Type: application/json

{"input": "Explain quantum computing", "config": {"provider": "openai"}}

Enable streaming in your settings:

# settings.yaml
platform:
  enable_streaming: true

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.

---

Related Documentation

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