LLMAdapter

LLMAdapter is a unified Python client for Large Language Model (LLM) providers. It gives you one consistent interface for chat completions, embeddings, tool calling, streaming, context caching, and full agentic workflows — across Azure OpenAI, Azure Foundry, OpenAI, Vertex AI, and Anthropic — without changing your code when you switch providers.

New to LLMs or AI APIs? Start with Layer 1 — Core Adapter. Once you're comfortable, work your way through the layers below.

Start Here

The runnable example covers chat, embeddings, Responses, async requests, LangChain, agents, and orchestration. Install its optional dependencies, then try the credential-free resilience section first:

pip install -e ".[all,dev]"
python examples/llmadapter_example.py --section resilience

To exercise provider-backed sections, configure the Azure OpenAI credentials used by the example. Run everything with --section all, or choose one section:

python examples/llmadapter_example.py --section core
python examples/llmadapter_example.py --section responses
python examples/llmadapter_example.py --section async
python examples/llmadapter_example.py --section langchain
python examples/llmadapter_example.py --section agent
python examples/llmadapter_example.py --section orchestration
python examples/llmadapter_example.py --section all

The Responses section requires an Azure OpenAI deployment that supports that API; sections requiring credentials are skipped when the chat profile is missing. For exact signatures, capability boundaries, and return types, use the API reference.

Table of Contents

  1. Feature Guide
  2. Installation
  3. Environment Variables & Credentials
  4. Layer 1 — Core Adapter (LLMAdapter / AsyncLLMAdapter)
  5. Responses API (adapter.responses)
  6. Layer 2 — LangChain Compatibility (ChatLLMAdapter)
  7. Layer 3 — Advanced Agent Runtime (LLMAdapterAgent)
  8. Layer 4 — Multi-Agent Orchestration
  9. Extending with a New Provider
  10. Project Structure
  11. API Reference

Feature Guide — What to Use, When, and Why

Use the simplest layer that meets your needs. Each layer builds on the one below it; provider and model support still depends on the operation.

Need Use Why / important notes
Portable text, tools, and embeddings LLMAdapter Lowest-level, provider-neutral API. Use for application services and batch jobs.
Non-blocking requests or streams AsyncLLMAdapter Uses provider-native async chat/Responses clients where available.
New OpenAI/Azure response workflows adapter.responses Supports typed output, background jobs, state, built-in tools, and structured parse. OpenAI/Azure OpenAI only.
Existing OpenAI-shaped integration adapter.chat.completions Retains legacy Chat Completions behavior across all supported chat providers.
LangChain chains or agents ChatLLMAdapter Choose api="chat" for broad provider support or api="responses" for OpenAI/Azure Responses features.
Tool-running autonomous workflows create_llmadapter_agent Adds bounded tool concurrency, deadlines, retry policy, routing, reflection, and checkpoints.
Several specialist agents invoke_specialists / aggregate_specialists Filters specialists by capability, attaches weights, and returns an explicit aggregation result.

ChatLLMAdapter and all agent/orchestration exports require langchain-core, provided by the langchain extra.

One configuration shape for every provider

Pass the union of provider-specific fields without changing caller code when you switch providers. Unrelated fields are accepted and ignored by the selected provider. Default routing always uses:

$$ \text{deployment_name} \rightarrow \text{per-call model} \rightarrow \text{configured model} $$

Use deployment_name whenever your endpoint routes by deployment, including Azure OpenAI native endpoints and Azure OpenAI-compatible /openai/v1/ endpoints. A configured deployment intentionally cannot be overridden by a per-call model alias. Without a deployment, most providers accept a per-call model; Anthropic fixes its model when the adapter is created and ignores per-call model overrides.

Provider and capability boundaries

Feature OpenAI Azure OpenAI Azure Foundry Anthropic Vertex AI
Legacy chat Yes Yes Yes Yes Yes, sync only
Embeddings Yes Yes Yes No No
Native async chat Yes Yes Yes Yes No
Responses API Yes Yes No No No
Built-in Responses tools Yes Yes No No No
Prompt caching Usage telemetry Usage telemetry No adapter API Explicit markers Provider-level CachedContent only

Call adapter.responses.capabilities() before relying on model/deployment specific Responses features. A result of unknown means the provider may support the feature, but it should be validated on the target deployment.

Production defaults

  • Use timeout to set provider transport timeouts.
  • Use share_clients=True only when adapters are frequently created; shared clients are lease-safe during eviction.
  • Use circuit_breaker=True for fail-fast behavior during provider outages.
  • Use backup={...} for an explicit compatible fallback target.
  • Configure rpm, tpm, context_output_reserve_tokens, and an explicit Redis outage policy for predictable cost and latency control.
  • Keep provider credentials in environment variables or a secret manager, never in notebooks, examples, source files, or committed .env files.

Installation

# Create and activate a virtual environment (recommended)
python -m venv .venv
source .venv/bin/activate      # Linux / macOS
# .venv\Scripts\activate       # Windows

# Python 3.11 or later is required.

# Install the base package: OpenAI and Azure OpenAI support.
pip install .

# Install one optional integration when needed.
pip install ".[redis]"          # Redis-backed distributed rate limiting
pip install ".[anthropic]"      # Anthropic chat
pip install ".[azure-foundry]"  # Azure Foundry chat and embeddings
pip install ".[vertexai]"       # Vertex AI sync chat
pip install ".[langchain]"      # ChatLLMAdapter and agent/orchestration APIs

# Install every provider integration plus LangChain.
pip install ".[all]"

# Install the runnable example, tests, and development tooling.
pip install -e ".[all,dev]"

Environment Variables & Credentials

LLMAdapter does not load environment variables or .env files automatically. Load credentials in application code or from a secret manager, then pass the resulting values to the adapter constructor. The runnable example loads a local .env file for convenience. The live-test hook loads .env only when LLMADAPTER_RUN_LIVE_TESTS=1 is already set in the shell.

Create a .env file:

# Azure OpenAI
AZURE_OPENAI_API_KEY=your-azure-openai-key
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4o
AZURE_OPENAI_API_VERSION=2024-02-01
AZURE_OPENAI_MODEL=gpt-4o

# Azure OpenAI embeddings (optional for the runnable example)
AZURE_OPENAI_EMBEDDING_API_KEY=your-azure-openai-embedding-key
AZURE_OPENAI_EMBEDDING_ENDPOINT=https://your-embedding-resource.openai.azure.com/
AZURE_OPENAI_EMBEDDING_API_VERSION=2024-02-01
AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME=text-embedding-3-small

# Azure Foundry (optional)
AZURE_FOUNDRY_API_KEY=your-foundry-key
AZURE_FOUNDRY_ENDPOINT=https://your-foundry-endpoint.inference.ai.azure.com/
AZURE_FOUNDRY_DEPLOYMENT_NAME=Llama-4-Maverick
AZURE_FOUNDRY_API_VERSION=2024-05-01-preview

# OpenAI (optional)
OPENAI_API_KEY=sk-...
OPENAI_ORG=org-...
OPENAI_PROJECT=proj_...

# Vertex AI (optional)
VERTEXAI_CREDENTIALS_PATH=/path/to/service_account.json
VERTEXAI_LOCATION=us-central1
VERTEXAI_MODEL=gemini-2.5-pro

# Anthropic (optional)
ANTHROPIC_API_KEY=sk-ant-api03-...

Load in Python:

The following local-development convenience uses python-dotenv, which is included in the dev extra. Install .[dev] (or install python-dotenv separately) before using it; production applications can instead read os.environ through their normal secret-management flow.

import os
from dotenv import load_dotenv

load_dotenv()  # reads .env from the current directory

api_key = os.environ["AZURE_OPENAI_API_KEY"]

Run live tests explicitly:

LLMADAPTER_RUN_LIVE_TESTS=1 python -m pytest test/ -m live

Layer 1 — Core Adapter (LLMAdapter / AsyncLLMAdapter)

The core adapter is a thin, consistent wrapper around provider HTTP APIs. Think of it as a smarter version of the official openai Python client that also supports Azure and other providers without any code changes on your side.

Supported Providers & Configuration

LLMAdapter and AsyncLLMAdapter accept the union of provider-specific constructor fields. This means one shared configuration object may include fields such as deployment_name, api_version, endpoint, location, and credentials_path without changing when the provider changes. Each provider uses the fields it understands and ignores unrelated fields. A configured deployment_name takes precedence over a per-call model, followed by the configured model.

Provider Required constructor fields Optional provider-specific fields
openai api_key endpoint, organization, project
azure_openai api_key, endpoint, api_version, deployment_name responses_api_version, region
azure_foundry api_key, endpoint, api_version, deployment_name —
anthropic api_key endpoint, deployment_name, enable_prompt_caching
vertexai credentials_path, location scopes

Shared operational fields include timeout, max_retries, retry_base_delay, rpm, tpm, embedding limits, completion_context_window, context_output_reserve_tokens, context_overhead_tokens, backup, and user_id. mcp_policy is required before remote MCP or connector tools can be sent. A backup must contain a provider and is validated during adapter construction.

Azure OpenAI

import os
from llmadapter import LLMAdapter

adapter = LLMAdapter(
    provider="azure_openai",
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
    api_version="2024-02-01",
    model="gpt-4o",
    redis_url="redis://server_disabled",  # disables all adapter rate-limit enforcement
    user_id="my-app-user",
)

Azure Foundry

Used for open-weight models (Llama, Mistral, Phi, etc.) hosted on Azure AI Foundry.

adapter = LLMAdapter(
    provider="azure_foundry",
    api_key=os.environ["AZURE_FOUNDRY_API_KEY"],
    endpoint=os.environ["AZURE_FOUNDRY_ENDPOINT"],
    deployment_name=os.environ["AZURE_FOUNDRY_DEPLOYMENT_NAME"],
    api_version="2024-05-01-preview",
    model="Llama-4-Maverick",
    redis_url="redis://server_disabled",
    user_id="my-app-user",
)

OpenAI

adapter = LLMAdapter(
    provider="openai",
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4o",
    organization=os.environ.get("OPENAI_ORG"),
    project=os.environ.get("OPENAI_PROJECT"),
    timeout=60,
)

For OpenAI-compatible gateways or local proxies, set endpoint to the base URL.

Vertex AI

The Vertex AI provider requires an extra dependency:

pip install google-genai==1.49.0

adapter = LLMAdapter(
    provider="vertexai",
    credentials_path=os.environ["VERTEXAI_CREDENTIALS_PATH"],
    location=os.environ.get("VERTEXAI_LOCATION", "us-central1"),
    model=os.environ.get("VERTEXAI_MODEL", "gemini-2.5-pro"),
    scopes=["https://www.googleapis.com/auth/cloud-platform"],
    redis_url="redis://server_disabled",
    user_id="my-app-user",
)

Vertex AI notes: Only synchronous chat is implemented. Async chat and embeddings are not available for the Vertex AI provider. Messages are converted to Gemini content parts, system instructions, tool calls, and tool responses. Common provider-specific kwargs: system_instruction, temperature, top_p, top_k, candidate_count, seed, max_output_tokens, stop_sequences, presence_penalty, frequency_penalty.

Anthropic

adapter = LLMAdapter(
    provider="anthropic",
    api_key=os.environ["ANTHROPIC_API_KEY"],
    model="claude-opus-4-5",  # or claude-3-7-sonnet, claude-haiku-3-5, etc.
    enable_prompt_caching=True,  # auto-inject cache_control on system + tools (default True)
    redis_url="redis://server_disabled",
    user_id="my-app-user",
)

Anthropic notes: Two caching modes are available:

  • Explicit block-level (enable_prompt_caching=True, default): injects cache_control on the last system block > and last tool definition — caches the static prefix independently from the conversation.
  • Automatic caching: pass cache_control={"type": "ephemeral"} as a kwarg to create() and Anthropic > automatically caches up to the last cacheable block. Best for multi-turn conversations.

Cache eligibility is decided by Anthropic and varies by model and cache mode. The adapter applies markers but does not enforce a local token threshold.

Chat Completions — Sync

The basic request/response pattern. You send a list of messages and get a response back.

response = adapter.chat.completions.create(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the capital of France?"},
    ],
    temperature=0.0,  # 0 = deterministic, 1 = creative
    max_tokens=64,
)

# Access the reply
print(response.choices[0].message.content)

# Access token usage
print(f"Prompt tokens: {response.usage.prompt_tokens}")
print(f"Completion tokens: {response.usage.completion_tokens}")
print(f"Total tokens: {response.usage.total_tokens}")

Understanding messages:

Role Purpose
system Sets the AI's personality and instructions. Used once at the start.
user Your question or prompt.
assistant Previous AI responses (for multi-turn conversations).
tool The result of a tool/function call (see Tool Calling).

Multi-turn conversation:

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "My name is Alice."},
]

response = adapter.chat.completions.create(messages=messages)
reply = response.choices[0].message.content

# Add assistant's reply to history, then continue
messages.append({"role": "assistant", "content": reply})
messages.append({"role": "user", "content": "What's my name?"})

response2 = adapter.chat.completions.create(messages=messages)
print(response2.choices[0].message.content)  # "Your name is Alice."

Chat Completions — Streaming

Streaming returns text token-by-token as it is generated, which feels faster to the end user.

stream = adapter.chat.completions.create(
    messages=[{"role": "user", "content": "Tell me a short story about a robot."}],
    stream=True,
)

for chunk in stream:
    # The last chunk may carry usage statistics
    if chunk.usage:
        print(f"\n[Total tokens: {chunk.usage.total_tokens}]")
        continue
    if chunk.choices:
        print(chunk.choices[0].delta.content or "", end="", flush=True)

The adapter automatically injects stream_options={"include_usage": True} when the provider supports it, so the final chunk contains token counts.

Chat Completions — Async

Use AsyncLLMAdapter when you are in an async context (FastAPI, asyncio event loop, etc.).

import asyncio
import os
from llmadapter import AsyncLLMAdapter


async def main():
    adapter = AsyncLLMAdapter(
        provider="azure_openai",
        api_key=os.environ["AZURE_OPENAI_API_KEY"],
        endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
        api_version="2024-02-01",
        model="gpt-4o",
    )

    response = await adapter.chat.completions.create(
        messages=[{"role": "user", "content": "Say hello in French."}],
        max_tokens=16,
    )
    print(response.choices[0].message.content)


asyncio.run(main())

Async Streaming

async def stream_example():
    stream = await adapter.chat.completions.create(
        messages=[{"role": "user", "content": "Explain Python generators briefly."}],
        stream=True,
    )

    async for chunk in stream:
        if chunk.usage:
            print(f"\n[Tokens: {chunk.usage.total_tokens}]")
            continue
        if chunk.choices:
            print(chunk.choices[0].delta.content or "", end="", flush=True)

All Chat Parameters Reference

Pass provider/model-dependent keyword arguments to adapter.chat.completions.create(...). Unsupported fields may be ignored, translated, or rejected by the selected provider.

Parameter Type Description
messages list[dict] Required. The conversation history.
model str Per-call model override when the provider supports it; ignored when a deployment is configured and by Anthropic.
stream bool Return a streaming iterator instead of a full response.
temperature float Randomness. 0 = deterministic, 1 = creative.
top_p float Nucleus sampling probability. Alternative to temperature.
max_tokens int Maximum tokens to generate in the response.
max_completion_tokens int Use this instead of max_tokens for reasoning models (o1, o3).
stop str | list[str] Stop sequences — generation halts when any of these appear.
n int Number of completions to generate. Default 1.
seed int Seed for reproducibility (provider support varies).
presence_penalty float Penalizes new tokens already present in the response.
frequency_penalty float Penalizes repeated tokens.
user str A user identifier for abuse detection.
response_format dict | type Raw JSON-object/schema dictionaries are provider-dependent; a Pydantic BaseModel subclass is the portable typed-parse route.
tools list[dict] List of function/tool definitions.
tool_choice str | dict "auto", "none", or {"type": "function", "function": {"name": "..."}}.
parallel_tool_calls bool OpenAI-compatible control; support is provider/model dependent.
logit_bias dict Token probability adjustments.
logprobs bool Return log probabilities for output tokens.
top_logprobs int Number of top log probability alternatives to return.
modalities list[str] Output modalities, e.g. ["text", "audio"].
audio dict Audio output configuration.
prediction dict Predicted output hint for latency optimization.
store bool Store the conversation in the provider's storage.
metadata dict Arbitrary metadata attached to the request.
reasoning_effort str For reasoning models: "low", "medium", "high".
extra_body dict Pass provider-specific extras (e.g., Azure RAG data_sources).
live_logs bool Sync adapter only: enables redacted INFO request/response logging.
truncate bool Automatically truncate messages to fit the context window.

Structured Outputs

Force the model to return a specific JSON structure. There are three modes, in order of increasing ergonomics.

Mode 1 — JSON Object mode

The model returns valid JSON, but the schema is not enforced. Fastest, but you must parse and validate yourself.

import json

response = adapter.chat.completions.create(
    messages=[
        {"role": "system", "content": "Always respond with valid JSON."},
        {"role": "user", "content": "Return a user profile with name and age."},
    ],
    response_format={"type": "json_object"},
    temperature=0,
)

data = json.loads(response.choices[0].message.content)
print(data)

Mode 2 — JSON Schema mode

The model is constrained to your exact schema. You still parse the raw JSON string yourself.

response = adapter.chat.completions.create(
    messages=[
        {"role": "system", "content": "Follow the schema exactly."},
        {"role": "user", "content": "Classify the sentiment: 'This product is amazing!'"},
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "sentiment_result",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "label": {"type": "string", "enum": ["positive", "negative", "neutral"]},
                    "confidence": {"type": "number"},
                },
                "required": ["label", "confidence"],
                "additionalProperties": False,
            },
        },
    },
    temperature=0,
)

Mode 3 — Typed Parse (recommended)

Pass a Pydantic BaseModel subclass as response_format, or use the dedicated .parse() method. This is the portable structured-output route: each provider uses its supported native schema or forced-tool mechanism and the adapter deserializes the result into message.parsed when available.

from pydantic import BaseModel
from typing import List

class MovieReview(BaseModel):
    title: str
    year: int
    rating: float
    pros: List[str]
    cons: List[str]

# Option A — via .parse() (clearest intent)
response = adapter.chat.completions.parse(
    messages=[
        {"role": "system", "content": "You are a film critic."},
        {"role": "user", "content": "Review Inception (2010)."},
    ],
    response_format=MovieReview,
    temperature=0,
)

review: MovieReview = response.choices[0].message.parsed
print(review.title)   # "Inception"
print(review.rating)  # e.g. 9.1
print(review.pros)    # ["Complex narrative", ...]

# Option B — via .create() with stream=False (identical result)
response = adapter.chat.completions.create(
    messages=[{"role": "user", "content": "Review Inception."}],
    response_format=MovieReview,
    temperature=0,
)
review = response.choices[0].message.parsed

How each provider implements typed parse:

Provider Mechanism Notes
OpenAI / Azure OpenAI beta.chat.completions.parse() — native SDK Full strict mode; message.parsed set by the SDK
Anthropic Forced tool use — schema injected as a tool's input_schema Works across all Claude models/versions
Vertex AI Native response_schema=<class> in GenerateContentConfig SDK sanitizes schema for Gemini; response.parsed read natively
Azure Foundry JsonSchemaFormat with strict-compliant schema additionalProperties:false + all-required enforced

Streaming + Pydantic class: Passing a Pydantic class with stream=True automatically converts it to a strict json_schema dict. You get the structured text stream but no automatic parsed deserialization — iterate chunks and parse the final assembled content manually.

message.parsed is serialization-safe:

import json

dumped = json.loads(response.model_dump_json())
print(dumped["choices"][0]["message"]["parsed"])  # dict, not a Pydantic instance

When to use which? Raw json_object and json_schema dictionaries are provider/model-specific. Use the Pydantic class / .parse() mode when you need the adapter's portable typed-parse behavior. Anthropic does not honor raw Chat Completions response_format dictionaries; its typed path uses forced tool calling.

Tool Calling (Raw API)

Tool calling lets you define Python functions as "tools" the model can invoke. The model decides when to call a tool, and your code actually executes it.

Step 1 — Define a tool:

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name, e.g. 'Tokyo'"},
                },
                "required": ["city"],
            },
        },
    }
]

Step 2 — Send the request with tools:

response = adapter.chat.completions.create(
    messages=[{"role": "user", "content": "What is the weather in Tokyo?"}],
    tools=tools,
    tool_choice="auto",  # let the model decide when to call a tool
)

message = response.choices[0].message

Step 3 — Execute the tool and send the result back:

import json

messages = [{"role": "user", "content": "What is the weather in Tokyo?"}]

# Append the assistant's tool_call message
messages.append({
    "role": "assistant",
    "content": message.content,
    "tool_calls": [tc.model_dump() for tc in (message.tool_calls or [])],
})

# Run your actual function
tool_call = message.tool_calls[0]
args = json.loads(tool_call.function.arguments)
weather_result = {"city": args["city"], "temperature": 22, "condition": "Sunny"}

# Append the tool result
messages.append({
    "role": "tool",
    "tool_call_id": tool_call.id,
    "content": json.dumps(weather_result),
})

# Get the final answer
final = adapter.chat.completions.create(messages=messages, tools=tools)
print(final.choices[0].message.content)

Tip: For a managed agentic loop (instead of writing this pattern yourself), see Layer 3.

Multimodal Input (Images)

Send images alongside text by using a list of content parts in the user message:

response = adapter.chat.completions.create(
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What objects do you see in this image?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/photo.jpg",
                        "detail": "low",  # "low", "high", or "auto"
                    },
                },
            ],
        }
    ],
    max_tokens=256,
)

print(response.choices[0].message.content)

The provider endpoint must be able to fetch the image URL. detail="low" / "high" is an OpenAI-compatible control; other providers may ignore it. Provider-specific image processing and pricing remain provider-managed.

Embeddings

Embeddings turn text into numerical vectors, useful for semantic search, clustering, and similarity comparisons.

Basic embeddings:

response = adapter.embeddings.create(
    input=["Hello world", "Embeddings represent meaning"],
    model="text-embedding-3-large",
)

for item in response.data:
    print(f"Index {item.index}: vector length = {len(item.embedding)}")

print(f"Total tokens: {response.usage.total_tokens}")

Batched embeddings (for large document sets):

# Process in batches of 2 to stay within provider limits
response = adapter.embeddings.create(
    input=["Doc 1", "Doc 2", "Doc 3", "Doc 4", "Doc 5"],
    model="text-embedding-3-large",
    batching=True,
    batchsize=2,  # send 2 texts per API call
    restore_order=True,  # reassemble in original order when done
)

Reduced dimensions (saves storage and speeds up similarity search):

response = adapter.embeddings.create(
    input=["Compact vector test"],
    model="text-embedding-3-large",
    dimensions=256,  # reduce from 3072 to 256 dimensions
)

All embedding parameters:

Parameter Type Description
input str | list[str] Text(s) to embed.
model str Embedding model name.
batching bool Process input in batches.
batchsize int Texts per API call when batching.
restore_order bool Reassemble batched results in input order.
dimensions int Reduce output vector size (model must support it).
encoding_format str "float" or "base64".
live_logs bool Sync adapter only: enable redacted INFO request/response logs.

Context Caching

Context caching lets you store a static prompt prefix on the provider's servers so repeated calls reuse it instead of re-billing the same tokens on every request.

Mechanism per provider

Provider Mechanism Control
Azure OpenAI / OpenAI Automatic — any static prefix ≥ 1 024 tokens is cached transparently None needed
Anthropic Explicit — cache_control: {type: ephemeral} markers enable_prompt_caching=True (default)
Vertex AI Explicit — create a CachedContent resource, then attach it by name Direct Vertex provider integration; no stable adapter-level lifecycle API
Azure Foundry None — backend does not currently expose cache signals N/A

Reading cache metrics

Normalized usage exposes cache-related fields, but non-zero telemetry depends on what the provider and model return:

response = adapter.chat.completions.create(
    messages=[{"role": "user", "content": "Summarize the current cache telemetry."}],
)

usage = response.usage
print(usage.prompt_tokens)  # total input tokens billed
print(usage.cached_tokens)  # tokens served from cache (any provider)
print(usage.cache_creation_input_tokens)  # cache creation telemetry when returned
print(usage.cache_read_input_tokens)  # cache-read telemetry when returned
print(usage.cache_savings_pct)  # percentage of prompt served from cache

cached_tokens is a unified read-only property that works for all providers:

  • OpenAI / Azure OpenAI — reads prompt_tokens_details.cached_tokens
  • Anthropic / Vertex AI — reads cache_read_input_tokens

Azure OpenAI / OpenAI — automatic caching

No code changes required. Just send the same static system prompt (≥ 1 024 tokens) on repeated calls:

SYSTEM = "You are a detailed technical assistant..." * 20  # must exceed ~1024 tokens

messages = [
    {"role": "system", "content": SYSTEM},
    {"role": "user", "content": "What is a Merkle tree?"},
]

# First call — cold (cache miss)
r1 = adapter.chat.completions.create(messages=messages, model="gpt-4o", temperature=0)
print(r1.usage.cached_tokens)  # 0 on first call

# Second call — hot (cache hit)
r2 = adapter.chat.completions.create(messages=messages, model="gpt-4o", temperature=0)
print(r2.usage.cached_tokens)  # e.g. 1280 — 68% savings
print(r2.usage.cache_savings_pct)  # e.g. 68.9

Streaming also reports cache tokens in the final usage chunk:

last_usage = None
for chunk in adapter.chat.completions.create(
        messages=messages, model="gpt-4o", stream=True,
        stream_options={"include_usage": True},
):
    if chunk.usage:
        last_usage = chunk.usage

print(last_usage.cached_tokens)  # same field, same value

Anthropic — explicit cache_control

The provider supports two caching modes (both work via the same create() call):

Mode 1 — Explicit block-level (default, enable_prompt_caching=True): cache_control is injected on the last system block and last tool definition. Caches the static prefix independently from the growing conversation history.

adapter = LLMAdapter(
    provider="anthropic",
    api_key=os.environ["ANTHROPIC_API_KEY"],
    model="claude-opus-4-5",
    enable_prompt_caching=True,  # default — no further action needed
)

messages = [
    {"role": "system", "content": SYSTEM},  # cache_control injected automatically
    {"role": "user", "content": "Explain bloom filters."},
]

r1 = adapter.chat.completions.create(messages=messages, model="claude-opus-4-5")
print(r1.usage.cache_creation_input_tokens)  # tokens written to cache on first call

r2 = adapter.chat.completions.create(messages=messages, model="claude-opus-4-5")
print(r2.usage.cache_read_input_tokens)  # tokens read from cache (~10% cost)
print(r2.usage.cached_tokens)  # convenience alias

Mode 2 — Automatic caching (top-level cache_control kwarg): Anthropic automatically caches up to the last cacheable block and advances the breakpoint as the conversation grows. Best for multi-turn conversations.

r = adapter.chat.completions.create(
    messages=messages,
    model="claude-opus-4-5",
    cache_control={"type": "ephemeral"},  # top-level automatic caching
)

Minimum cacheable tokens: ≥ 4 096 for Opus 4.x and Haiku 4.5; ≥ 2 048 for Sonnet 4.6 and Haiku 3.5; ≥ 1 024 for Sonnet 4.5/4/3.7 and Opus 4.1/4. Requests below the threshold silently skip caching.

To opt out of block-level injection for a specific adapter:

adapter = LLMAdapter(
    provider="anthropic",
    api_key=os.environ["ANTHROPIC_API_KEY"],
    model="claude-opus-4-5",
    enable_prompt_caching=False,
)

Vertex AI — CachedContent

VertexAIChatProvider implements provider-level CachedContent creation, and chat calls can pass cached_content when the selected Gemini target supports it. LLMAdapter does not expose a stable adapter-level CachedContent lifecycle API; do not access private provider caches to obtain an underlying client. Use the direct provider integration only when that provider-specific lifecycle is required, and treat cache-read telemetry as provider-returned usage metadata.

Failover / Backup Config

Configure a backup provider that automatically takes over if the primary fails:

adapter = LLMAdapter(
    provider="azure_openai",
    api_key=os.environ["PRIMARY_KEY"],
    endpoint=os.environ["PRIMARY_ENDPOINT"],
    deployment_name="primary-gpt4o",
    api_version="2024-02-01",
    model="gpt-4o",
    backup={
        "provider": "azure_openai",
        "api_key": os.environ["BACKUP_KEY"],
        "endpoint": os.environ["BACKUP_ENDPOINT"],
        "deployment_name": "backup-gpt4o",
        "api_version": "2024-02-01",
        "model": "gpt-4o",
    },
)

# If the primary fails, the backup is tried automatically
response = adapter.chat.completions.create(
    messages=[{"role": "user", "content": "Hello"}]
)

When to use: Production systems where uptime matters — region failover, deployment quota exhaustion, or temporary provider outages. Chat and Responses creation fail over only for retryable errors (such as rate limits, connection failures, and server errors), not invalid requests. Each target's circuit breaker also guards backup calls.

Rate Limiting

LLMAdapter uses shared in-process counters by default. For distributed deployments, configure a Redis instance. Each quota is scoped to the provider target (provider, endpoint, deployment/model), user, request type, and time window:

adapter = LLMAdapter(
    provider="azure_openai",
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
    api_version="2024-02-01",
    model="gpt-4o",
    # Rate limits
    rpm=100,  # max requests per minute (chat)
    tpm=100000,  # max tokens per minute (chat)
    embedding_rpm=60,  # max requests per minute (embeddings)
    embedding_tpm=60000,  # max tokens per minute (embeddings)
    completion_context_window=128000,  # context window size for truncation
    context_output_reserve_tokens=4096,  # preserve room for generation
    # Redis for distributed tracking
    redis_url="redis://localhost:6379/0",
    user_id="user-42",  # per-user tracking key
)

When a configured Redis backend is unavailable, choose an explicit policy:

adapter = LLMAdapter(
    provider="openai",
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4o",
    redis_url="redis://localhost:6379/0",
    rate_limit_failure_policy="memory_fallback",  # default: keep local limits
    redis_connect_timeout=1.0,
)

Supported policies are "memory_fallback", "fail_open", and "fail_closed". "fail_closed" raises RateLimitBackendUnavailableError rather than sending an unmetered request.

Disable rate limiting (development):

adapter = LLMAdapter(
    provider="openai",
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4o",
    redis_url="redis://server_disabled",
)

redis_url=None uses shared in-process limits. The redis://server_disabled sentinel disables all adapter rate-limit enforcement and should only be used deliberately in local development or tests. The limiter uses fixed windows; denied requests still reserve the current window. A single request larger than the configured TPM limit raises LLMAdapterError.

Context budgets

completion_context_window is enforced as a request-wide budget, not a per-message limit. LLMAdapter reserves the greater of context_output_reserve_tokens and any per-call generation limit (max_tokens, max_completion_tokens, or max_output_tokens), then estimates message framing and tool/schema overhead. If a request cannot fit, it raises ContextLengthExceededError; pass truncate=True to retain system/developer messages and the newest eligible text history within that budget.

Multimodal chat blocks and Responses item lists are never rewritten merely to fit a budget. If they exceed capacity, the adapter raises instead of changing their structure. Server-side state referenced by previous_response_id or a Conversation is not locally measurable and remains provider-enforced.

Resilience & Performance Options

These optional LLMAdapter / AsyncLLMAdapter constructor kwargs tune resilience and throughput. Each is off/default unless set.

Option Default Effect
token_counter None Custom (text, model_name) -> int counter used for rate-limit and context-budget estimation. It can change whether/how much input is retained; final text slicing still uses the tokenizer implementation.
share_clients False Share provider SDK clients (and their HTTP connection pools) across adapter instances with the same config, via a bounded process-wide cache. Useful when adapters are created per request.
circuit_breaker False True or {"failure_threshold": 5, "reset_timeout": 30.0} — opens a per-target circuit (provider, operation, endpoint, resolved model/deployment) after repeated retryable 5xx health failures. Chat and Responses create paths support it; embeddings do not. 
from llmadapter import LLMAdapter

adapter = LLMAdapter(
    provider="anthropic",
    api_key=os.environ["ANTHROPIC_API_KEY"],
    model="claude-opus-4-5",
    share_clients=True,    # reuse one client across adapters with this config
    circuit_breaker=True,  # fail fast when the provider is persistently unhealthy
)

Retries: the adapter is the single source of retry truth — provider SDK built-in retries are disabled, and max_retries / retry_base_delay (or a per-call max_retries=) control full-jitter exponential back-off. max_retries=3 means up to four attempts. Provider retry_after values are honored when available. Agent tool retries are separate: they use fixed backoff and retry only eligible failed tool results. Set max_retries=0 to disable core-request retries.

Streaming lifecycle: a streaming request opens its provider stream only when you begin iterating. The circuit breaker records success only after the stream is fully consumed. Pre-output events are buffered, so establishment and pre-visible failures can retry up to max_retries times per target before backup failover. Once output is visible, the adapter never retries, fails over, or replays the stream. Provider-emitted terminal failure events are not replayed to a backup. Errors from lazy stream creation surface from the iterator rather than from create(..., stream=True). Close a stream (stream.close() or await stream.aclose()) if you stop consuming it early; the LangChain integration closes its underlying adapter stream when its stream iterator is closed.

Async embeddings: AsyncLLMAdapter exposes await adapter.embeddings.create(...) with the same signature as the sync adapter; the (synchronous) embedding providers run in a worker thread so the event loop stays free.

Error Handling

LLMAdapter raises typed exceptions so you can handle failures precisely:

from llmadapter.errors import (
    LLMAdapterError,  # Base class for all adapter errors
    ConfigurationError,  # Invalid adapter or backup configuration
    ContextLengthExceededError,  # Request cannot fit configured context budget
    AuthenticationError,  # Invalid API key
    ModelNotFoundError,  # Deployment/model name wrong
    RateLimitExceededError,  # Provider-side 429 response
    RateLimitBackendUnavailableError,  # Fail-closed Redis backend unavailable
    APIConnectionError,  # Network issues, timeouts
    TimeOutError,  # Request exceeded timeout
    ChatCompletionError,  # General completion failure
    EmbeddingGenerationError,  # Embedding request failed
    ContentFilterError,  # Response blocked by content policy
    InvalidResponseError,  # Malformed or unexpected response
    InvalidResponseRequestError,  # Invalid Responses create contract
    UnsupportedCapabilityError,  # Explicitly unsupported Responses feature
    ResponsesError,  # Responses API operation failed
    ResponseNotFoundError,  # Stored response was not found
)

try:
    response = adapter.chat.completions.create(
        messages=[{"role": "user", "content": "Hello"}]
    )
except AuthenticationError:
    print("Check your API key.")
except RateLimitExceededError:
    print("Slow down — you have hit your rate limit.")
except ContentFilterError as exc:
    print(f"Content blocked: {exc.message}")
    print(f"Filter details: {exc.context.get('filter_results')}")
except APIConnectionError as exc:
    print(f"Network error: {exc.message}")
except LLMAdapterError as exc:
    # Catch-all for any adapter error
    print(f"Error [{exc.status_code}] {exc.error_code}: {exc.message}")

Error attributes:

Attribute Description
message Human-readable description.
status_code HTTP status code from the provider.
error_code Adapter-defined error code string.
context Dict with extra details (e.g., filter results).

Response Shapes

Chat completion response:

response.id  # request ID
response.model  # model that was used
response.choices[0].message.content  # text reply
response.choices[0].message.parsed   # Pydantic model instance (set by .parse() — None otherwise)
response.choices[0].message.tool_calls  # list of tool calls (if any)
response.choices[0].finish_reason  # "stop", "tool_calls", "length", etc.
response.usage.prompt_tokens
response.usage.completion_tokens
response.usage.total_tokens
response.usage.cached_tokens  # tokens served from cache (all providers)
response.usage.cache_creation_input_tokens  # tokens written to cache (Anthropic mapping)
response.usage.cache_read_input_tokens  # tokens read from cache (Anthropic/Vertex mappings)
response.usage.cache_savings_pct  # % of prompt served from cache (0–100)
response.prompt_filter_results  # Azure content filter results
response.metadata  # extra metadata dict
response.provider_extensions  # versioned provider provenance container

Streaming chunk:

chunk.choices[0].delta.content  # text fragment (may be None)
chunk.choices[0].delta.tool_calls  # tool call fragments
chunk.choices[0].finish_reason  # set on the last content chunk
chunk.usage  # set on the final usage chunk

Embedding response:

response.data[0].embedding  # list[float] — the vector
response.data[0].index  # position in input list
response.model  # model used
response.usage.total_tokens

Responses API (adapter.responses)

Alongside the legacy chat.completions surface, the adapter exposes the OpenAI Responses API (/v1/responses) at adapter.responses. Create operations use token counting, rate limiting, retries with full-jitter backoff, optional circuit breaking, and optional backup failover. Management operations are retried and error-translated but are not metered, circuit-broken, or failed over.

For streams, provider creation is lazy and breaker success is recorded only after full consumption. Retry and failover can occur only before visible output is released; a partially visible stream is never replayed to a backup provider.

Provider support: openai and azure_openai only. Other providers raise ResponsesError with status_code=501. chat.completions remains available for all five providers, so nothing about your existing code changes.

Methods

adapter.responses.create(input, stream=False, truncate=False, **kwargs)   # -> Response | Iterator[ResponseStreamEvent]
adapter.responses.stream(input, **kwargs)                                 # -> Iterator[ResponseStreamEvent]
adapter.responses.parse(input, text_format=Model, **kwargs)               # -> Response (with .output_parsed)
adapter.responses.retrieve(response_id, **kwargs)                         # -> Response
adapter.responses.delete(response_id)                                     # -> ResponseDeleted
adapter.responses.cancel(response_id)                                     # -> Response
adapter.responses.compact(input=[...], **kwargs)                          # -> CompactedResponse
adapter.responses.input_items.list(response_id, **kwargs)                 # -> ResponseItemList
adapter.responses.input_tokens.count(**kwargs)                            # -> InputTokenCountResponse
adapter.responses.capabilities()                                          # -> ResponseCapabilities
adapter.responses.create_background(input, **kwargs)                      # -> Response
adapter.responses.wait(response_id, timeout=..., **kwargs)                # -> Response
adapter.responses.submit_mcp_approval(previous_response_id, approval)     # -> Response
adapter.responses.conversations.create(items=None, **kwargs)              # -> Conversation
adapter.responses.conversations.retrieve(conversation_id, **kwargs)       # -> Conversation
adapter.responses.conversations.update(conversation_id, **kwargs)         # -> Conversation
adapter.responses.conversations.delete(conversation_id, **kwargs)         # -> ConversationDeleted
adapter.responses.conversations.list_items(conversation_id, **kwargs)     # -> ResponseItemList
adapter.responses.conversations.retrieve_item(conversation_id, item_id)   # -> ResponseItem
adapter.responses.conversations.delete_item(conversation_id, item_id)     # -> Conversation

AsyncLLMAdapter exposes the same operations as awaitables or async streams; responses.capabilities() remains synchronous because it resolves local configuration only. responses.compact accepts input items or previous_response_id to select the history to compact.

Capability contract and strict requests

Responses requests are validated before token reservation, retry, failover, or provider-client creation. Typos and incompatible state references raise InvalidResponseRequestError; explicitly disabled target features raise UnsupportedCapabilityError.

caps = adapter.responses.capabilities()
print(caps.status("tool.web_search"))  # supported / unknown / unsupported
print(caps.supports("tool.web_search"))  # True / None / False

adapter = LLMAdapter(
    provider="azure_openai",
    # ... normal Azure configuration ...
    responses_capabilities={
        "input_tokens.count": False,  # deployment-specific known limitation
        "tool.web_search": False,
    },
)

With a custom openai endpoint or compatible gateway, Responses features start as unknown rather than claiming native OpenAI support. Unknown features can still be called; use responses_capabilities to record verified support or known limitations for the configured target.

After adapter-only max_retries and live_logs controls are removed, the strict provider request contract includes instructions, tools, tool_choice, parallel_tool_calls, max_tool_calls, include, background, conversation, previous_response_id, reasoning, text, store, metadata, prompt, prompt_cache_key, prompt_cache_options, prompt_cache_retention, truncation, safety_identifier, service_tier, user, context, extra_headers, extra_query, and per-call timeout, plus generation fields model, temperature, top_p, top_logprobs, and max_output_tokens, and stream_options for streaming; structured-output aliases text_format and response_format are also accepted as Pydantic model classes. Use extra_body for a new provider field that has not yet been added to the adapter contract. previous_response_id and conversation cannot be combined. Unknown tool types and unsupported include values are rejected before dispatch.

ResponseCapabilities.supports(feature) returns True for known support, False for an explicit restriction, and None when support is target- or deployment-dependent.

max_retries overrides the configured retry count for a call (subject to the side-effecting request safeguards). In the synchronous adapter, live_logs controls redacted INFO logging; the async adapter accepts and ignores it.

Basic Usage

from llmadapter import LLMAdapter

adapter = LLMAdapter(provider="openai", api_key="sk-...", model="gpt-5")

response = adapter.responses.create(
    "Summarize the Apollo program in two sentences.",
    instructions="You are concise and factual.",
    max_output_tokens=200,
)
print(response.output_text)
print(response.usage.total_tokens, response.usage.cached_tokens, response.usage.reasoning_tokens)

input accepts either a plain string or a list of Responses input items:

response = adapter.responses.create([
    {"role": "user", "content": [{"type": "input_text", "text": "Describe this chart."}]},
])

Streaming

Streaming yields ResponseStreamEvent objects rather than chat chunks. Branch on event.type:

for event in adapter.responses.stream("Tell me a story."):
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        print("\nTokens:", event.usage.total_tokens)

Known event and output-item types are mapped to typed subclasses such as ResponseOutputTextDeltaEvent, ResponseCompletedEvent, ResponseFunctionCallItem, ResponseWebSearchCallItem, and ResponseCodeInterpreterCallItem. New provider event/item types remain available through the generic ResponseStreamEvent and ResponseOutputItem fallback models, including their preserved extra fields.

Typed web and file search tools

Use WebSearchTool and FileSearchTool instead of manually building tool dictionaries. Included results are exposed through response.web_sources and response.file_search_results.

from llmadapter import FileSearchTool, WebSearchTool

response = adapter.responses.create(
    "Compare internal notes with current public information.",
    tools=[
        WebSearchTool(search_context_size="medium"),
        FileSearchTool(vector_store_ids=["vs_123"], max_num_results=5),
    ],
    include=["web_search_call.action.sources", "file_search_call.results"],
)

for source in response.web_sources:
    print(source.title, source.url)
for result in response.file_search_results:
    print(result.filename, result.score)

Tool availability remains capability-gated by provider, API version, deployment, and model. Azure deployment support is frequently model-specific.

Typed execution tools

CodeInterpreterTool, ImageGenerationTool, and ComputerTool provide typed configuration for Responses tools. Their results are available as response.code_interpreter_outputs, response.generated_images, and response.computer_actions.

from llmadapter import CodeInterpreterTool, ImageGenerationTool

response = adapter.responses.create(
    "Analyze the spreadsheet and create a chart image.",
    tools=[
        CodeInterpreterTool(container={"type": "auto", "file_ids": ["file_123"]}),
        ImageGenerationTool(size="1024x1024"),
    ],
)

Computer use is a separate, application-controlled workflow:

from llmadapter import ComputerTool, ResponseComputerCallItem

response = adapter.responses.create(
    "Inspect the isolated test browser.",
    tools=[ComputerTool()],
)
for call in response.output:
    if isinstance(call, ResponseComputerCallItem):
        print(call.call_id, call.computer_actions, call.pending_safety_checks)

ComputerTool() sends {"type": "computer"}; it does not operate a browser. Each computer_call holds ordered actions and a call_id needed to return a screenshot. response.computer_actions is flattened for inspection only. An application must review safety checks, authorize and execute permitted actions in order in a persistent isolated UI, capture its actual screenshot, and send it back using previous_response_id=response.id and an input item with {"type": "computer_call_output", "call_id": call.call_id, "output": {"type": "computer_screenshot", "image_url": screenshot_data_url, "detail": "original"}}. Do not execute untrusted page instructions or consequential actions without user confirmation; bound and verify each run. Computer tool availability depends on the configured model and endpoint.

Remote MCP and connectors

McpTool and ConnectorTool use the Responses MCP transport. They are rejected unless the adapter has an explicit allowlist policy. Match allowed_origins to a server URL's netloc (host and optional port), not its full URL.

import os

from llmadapter import McpApprovalDecision, McpTool

adapter = LLMAdapter(
    provider="openai",
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-5",
    mcp_policy={
        "enabled": True,
        "allowed_server_labels": ["trusted"],
        "allowed_origins": ["mcp.example.com"],
        "allowed_connector_ids": ["connector_123"],
    },
)

response = adapter.responses.create(
    "Use the approved remote tool.",
    tools=[McpTool(server_label="trusted", server_url="https://mcp.example.com")],
)

for approval in response.pending_mcp_approvals:
    adapter.responses.submit_mcp_approval(
        previous_response_id=response.id,
        approval=McpApprovalDecision(approval_request_id=approval.approval_request_id, approve=True),
    )

require_approval defaults to "always". computer, code_interpreter, image_generation, remote MCP/connector tools, and background Responses creates suppress automatic adapter retries because they can produce side effects.

Structured Outputs

Pass a Pydantic model class as text_format; the deserialized instance lands on response.output_parsed.

from pydantic import BaseModel

class TripPlan(BaseModel):
    city: str
    daily_budget_eur: int

plan = adapter.responses.parse("Return city=Paris and a daily budget.", text_format=TripPlan)
print(plan.output_parsed.city, plan.output_parsed.daily_budget_eur)

Multi-Turn State

The Responses API can hold conversation state server-side, so you don't have to resend the whole history:

first = adapter.responses.create("My favourite colour is teal.", store=True)
second = adapter.responses.create("What is my favourite colour?", previous_response_id=first.id)
print(second.output_text)

For provider targets that expose the Conversations resource, use the capability- gated namespace:

conversation = adapter.responses.conversations.create(
    items=[{"role": "user", "content": "Remember that I prefer teal."}],
    metadata={"topic": "preferences"},
)
adapter.responses.conversations.update(conversation.id, metadata={"topic": "profile"})
items = adapter.responses.conversations.list_items(conversation.id)
adapter.responses.conversations.delete(conversation.id)

Background workflows and reasoning continuity

Use create_background and wait for stored long-running Responses. Polling uses bounded full-jitter backoff and can optionally cancel a response when its local deadline expires.

job = adapter.responses.create_background("Run the long analysis.")
final = adapter.responses.wait(job.id, timeout=300, cancel_on_timeout=True)

create_background sets background=True and defaults store=True without overriding an explicit store=False. wait(..., on_poll=callback) invokes callback(response, attempts) on each poll; the async variant also awaits an async callback. Cancellation on timeout is best effort.

Async callers can use await adapter.responses.create_background(...) and await adapter.responses.wait(...). If include=["reasoning.encrypted_content"] is requested, encrypted continuity payloads are available through response.encrypted_reasoning_content for provider-supported stateless follow-up workflows.

Managing Stored Responses

fetched = adapter.responses.retrieve(first.id)

page = adapter.responses.input_items.list(first.id, limit=20)
for item in page:
    print(item.type, item.id)
if page.has_more:
    page = adapter.responses.input_items.list(first.id, after=page.last_id)

background = adapter.responses.create("Long job...", background=True, store=True)
adapter.responses.cancel(background.id)

compacted = adapter.responses.compact(previous_response_id=first.id)
adapter.responses.delete(first.id)

ResponseItemList is a cursor page with data, first_id, last_id, and has_more; it also supports direct iteration and len(page).

Pre-Flight Token Counting

input_tokens.count asks the server how many input tokens a request would consume, before you pay for it:

count = adapter.responses.input_tokens.count(
    input="How many tokens is this prompt?",
    instructions="You are a helpful assistant.",
)
print(count.input_tokens)

Async

from llmadapter import AsyncLLMAdapter

adapter = AsyncLLMAdapter(provider="openai", api_key="sk-...", model="gpt-5")

response = await adapter.responses.create("Hello!")

async for event in await adapter.responses.create("Stream this.", stream=True):
    if event.type == "response.output_text.delta":
        print(event.delta, end="")

await adapter.responses.delete(response.id)

Azure Notes

Azure routes by deployment, so the deployment name is used as the model id. The Responses endpoint also requires a recent api-version; when your configured api_version predates it, the adapter upgrades it for responses.* calls only. Set responses_api_version to pin one explicitly:

adapter = LLMAdapter(
    provider="azure_openai",
    api_key="...",
    endpoint="https://<resource>.openai.azure.com",
    deployment_name="gpt-5",
    api_version="2024-10-21",            # used by chat.completions
    responses_api_version="2025-03-01-preview",  # used by responses.*
)

Behavior Notes

  • create shares the completion RPM/TPM budget with chat.completions — both consume the same deployment quota.
  • The management verbs (retrieve, delete, cancel, compact, input_items.list, input_tokens.count) are retried and error-translated but are not rate limited, metered, circuit-broken, or failed over. delete and cancel are treated as non-idempotent and are not retried.
  • truncate=True only rewrites a plain-string input. Item lists carry tool calls and multimodal parts, so they are counted but passed through unchanged.
  • All Responses models allow extra fields, so new server-side item types and attributes are preserved rather than rejected.
  • adapter.responses.capabilities() is available only for OpenAI and Azure OpenAI targets; unsupported providers raise UnsupportedCapabilityError from this introspection method and ResponsesError(501) from responses.create.

Provider extension provenance

Use normalized fields for portable logic. Provider-specific fields are available through both a legacy dictionary and a versioned extension container:

extensions = response.provider_extensions
print(extensions.provider, extensions.schema_version)
print(extensions.fields)

provider_extensions is available on normalized chat responses, stream chunks, embeddings, Responses objects and output items, pages, counts, conversations, and stream events. LangChain message conversion does not preserve this container.

Layer 2 — LangChain Compatibility (ChatLLMAdapter)

ChatLLMAdapter wraps LLMAdapter as a LangChain BaseChatModel, so it can be used with standard LangChain prompts, chains, tools, callbacks, and agents. It requires the optional langchain-core dependency (pip install ".[langchain]").

Chat mode supports all five registered chat providers. Native async LangChain calls support OpenAI, Azure OpenAI, Azure Foundry, and Anthropic; Vertex AI is sync-only because this package has no native async Vertex chat provider.

What You Get Over Layer 1

  • invoke() / ainvoke() — standard LangChain call interface
  • stream() / astream() — streaming with LangChain message types
  • batch() — parallel batch invocations
  • bind_tools() — attach tools and get LangChain ToolCall objects back
  • with_structured_output() — parse output into Pydantic models automatically
  • LCEL chain composition (prompt | llm | parser)
  • Full compatibility with LangChain agents, memory, and callbacks

The constructor's streaming field is retained for LangChain configuration compatibility, but it does not make .invoke() or .ainvoke() stream. Call .stream() or .astream() explicitly. Call .close() when finished with synchronous resources, or await .aclose() to close both sync and async resources.

Basic Invoke & Stream

from llmadapter import ChatLLMAdapter
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatLLMAdapter(
    provider="azure_openai",
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
    api_version="2024-02-01",
    model="gpt-4o",
)

# invoke returns an AIMessage
response = llm.invoke("What is machine learning?")
print(response.content)

# Using LangChain message objects
response = llm.invoke([
    SystemMessage(content="You are an expert data scientist."),
    HumanMessage(content="Explain overfitting in one sentence."),
])
print(response.content)

# Streaming
for chunk in llm.stream("List three Python best practices."):
    print(chunk.content, end="", flush=True)

Async Invoke & Stream

import asyncio


async def async_example():
    response = await llm.ainvoke("Explain async/await briefly.")
    print(response.content)

    async for chunk in llm.astream("Count to 5."):
        print(chunk.content, end="")


asyncio.run(async_example())

Tool Binding with LangChain

Define tools using the @tool decorator and bind them to the LLM:

from langchain_core.tools import tool
from langchain_core.messages import ToolMessage


@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is 22 degrees C and sunny."


@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))  # noqa: S307 — for demo only


# Bind tools to the LLM
llm_with_tools = llm.bind_tools([get_weather, calculate])

# The model returns tool_calls in the AIMessage
response = llm_with_tools.invoke("What is the weather in Tokyo and what is 15 * 7?")

for tool_call in response.tool_calls:
    print(f"Calling: {tool_call['name']} with {tool_call['args']}")

# Execute the tools and continue
tool_map = {"get_weather": get_weather, "calculate": calculate}
messages = [HumanMessage("What is the weather in Tokyo and what is 15 * 7?")]
messages.append(response)  # add the AIMessage with tool_calls

for tool_call in response.tool_calls:
    result = tool_map[tool_call["name"]].invoke(tool_call["args"])
    messages.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))

# Get the final answer
final = llm_with_tools.invoke(messages)
print(final.content)

bind_tools options:

llm_with_tools = llm.bind_tools(
    tools=[get_weather, calculate],
    tool_choice="auto",  # "auto", "none", "required" / "any", or a specific tool name
)

"any" is translated to "required". A named tool must exist in the supplied tool list; otherwise bind_tools raises ValueError before a model request.

Structured Output with Pydantic

Automatically parse model output into a typed Pydantic model:

from pydantic import BaseModel
from typing import List


class MovieReview(BaseModel):
    title: str
    year: int
    rating: float
    pros: List[str]
    cons: List[str]


structured_llm = llm.with_structured_output(MovieReview)

review = structured_llm.invoke("Review the movie Inception (2010).")
print(review.title)  # "Inception"
print(review.rating)  # e.g. 9.1
print(review.pros)  # ["Complex narrative", "Visual effects", ...]

with_structured_output modes:

# Default: use JSON schema mode
llm.with_structured_output(MyModel)

# Use JSON schema mode (strict)
llm.with_structured_output(MyModel, method="json_schema")

# Use JSON mode (no schema enforcement)
llm.with_structured_output(MyModel, method="json_mode")

# Use a required function call instead of response-format structured output
llm.with_structured_output(MyModel, method="function_calling", include_raw=True)

With include_raw=True, the runnable returns a dictionary containing raw, parsed, and parsing_error; parse failures are captured rather than raised by that final parser stage.

Using the Responses API (api="responses")

Set api="responses" to retain the standard LangChain interface while routing requests through adapter.responses. This mode is available for openai and azure_openai providers only; the default remains legacy chat completions.

llm = ChatLLMAdapter(
    provider="openai",
    api="responses",
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-5",
)

reply = llm.invoke("Explain retrieval-augmented generation in one sentence.")
print(reply.content)

for chunk in llm.stream("Write a haiku about Python."):
    print(chunk.content, end="", flush=True)

In this mode, max_tokens is translated to max_output_tokens, chat-style function tools are converted to the Responses tool format, and structured output and tool binding continue to use LangChain's normal APIs. Stop sequences are ignored because the Responses API does not support them. LangChain text, image, and file content blocks are converted into Responses input items. Rich Responses output such as web sources, file-search results, code-interpreter artifacts, generated images, computer actions, MCP approvals, reasoning, and other non-message output items is exposed through AIMessage.additional_kwargs; use the raw adapter for Responses management methods such as retrieve, conversations, or input_tokens.count.

LangChain callback token events are emitted only for non-empty text deltas.

Chain Composition (LCEL)

Use LangChain Expression Language (LCEL) to build processing pipelines:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Build a chain: prompt -> LLM -> string output
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant who answers concisely."),
    ("human", "{question}"),
])

chain = prompt | llm | StrOutputParser()

result = chain.invoke({"question": "What is the difference between RAM and storage?"})
print(result)

# Stream the chain
for chunk in chain.stream({"question": "Explain neural networks simply."}):
    print(chunk, end="")

# Batch multiple inputs in parallel
results = chain.batch([
    {"question": "What is Python?"},
    {"question": "What is JavaScript?"},
    {"question": "What is Rust?"},
])
for r in results:
    print(r[:100])

Layer 3 — Advanced Agent Runtime (LLMAdapterAgent)

The advanced runtime gives you a production-ready ReAct-style agent loop: the model thinks, calls tools, observes results, and repeats until it has an answer. Everything — retry logic, timeouts, checkpoints, observability hooks — is handled for you.

AdvancedAgentConfig — Every Parameter Explained

AdvancedAgentConfig is a dataclass that controls how the agent loop behaves. Every parameter has a sensible default.

from llmadapter import AdvancedAgentConfig

config = AdvancedAgentConfig(
    max_iterations=12,
    max_runtime_seconds=180.0,
    parallel_tools=True,
    max_concurrency=4,
    tool_timeout_seconds=30.0,
    reflection_enabled=True,
    reflection_max_retries=2,
    interrupt_on_tool_error=False,
    tool_choice="auto",
    max_input_tokens_per_step=None,
    tool_retry_attempts=0,
    tool_retry_backoff_seconds=0.0,
    retry_on_tool_not_found=False,
    retry_non_idempotent_tools=False,
    escalate_on_retry_exhausted=False,
)

Iteration & time controls:

Parameter Default Description
max_iterations 12 Maximum number of think-act-observe cycles before the agent stops. Increase for complex multi-step tasks; lower for fast responses.
max_runtime_seconds 180.0 Runtime budget in seconds. Async model and reflection calls are bounded directly; sync calls are checked before and after completion, so configure provider timeout for a transport-level limit.

Tool execution controls:

Parameter Default Description
parallel_tools True When the model calls multiple tools in one step, run them concurrently. Significantly faster for independent tool calls.
max_concurrency 4 Maximum number of tools to run simultaneously when parallel_tools=True.
tool_timeout_seconds 30.0 Maximum seconds the agent waits for a tool result. Native async tools are cancelled; synchronous or thread-backed work cannot be forcibly stopped and may continue after the timeout.
tool_choice "auto" Passed to the model. "auto" = model decides, "none" = no tools, tool name = force a specific tool.

Reflection controls:

Parameter Default Description
reflection_enabled True Evaluates a non-tool candidate answer before returning it and requests a revision when needed.
reflection_max_retries 2 Maximum reflection-driven revisions per invocation, including resumed execution.

Error behavior:

Parameter Default Description
interrupt_on_tool_error False If True, stop the agent loop immediately on any tool error and return status="interrupted". Useful when errors mean the task cannot continue.

Token budget:

Parameter Default Description
max_input_tokens_per_step None If set, the agent returns status="stopped" with reason="max_input_tokens_per_step_exceeded" when the estimated input tokens for the next step exceed this value. Prevents runaway token usage.

Retry & escalation:

Parameter Default Description
tool_retry_attempts 0 How many times to retry a failed, explicitly idempotent tool call. 0 = no retries.
tool_retry_backoff_seconds 0.0 Seconds to wait between retry attempts. Use 0.5 or 1.0 for flaky network-dependent tools.
retry_on_tool_not_found False If True, retry even when the tool name is not registered (instead of raising immediately).
retry_non_idempotent_tools False Unsafe compatibility opt-in: retry ordinary tools that have not declared idempotency. Timed-out tools are never retried.
escalate_on_retry_exhausted False If True and no fallback exists, stop after tool retries are exhausted. With a fallback, ordinary tool-error routing chooses it on the next step.

Quick reference — common config patterns:

# Fast, cheap responses (simple Q&A with tools)
AdvancedAgentConfig(max_iterations=4, reflection_enabled=False, parallel_tools=False)

# Robust multi-step research agent
AdvancedAgentConfig(max_iterations=20, max_runtime_seconds=300, parallel_tools=True,
                    tool_retry_attempts=2, tool_retry_backoff_seconds=0.5)

# Strictly bounded (for user-facing features with SLA)
AdvancedAgentConfig(max_iterations=6, max_runtime_seconds=30,
                    max_input_tokens_per_step=4000, interrupt_on_tool_error=True)

# Fault-tolerant with fallback escalation
AdvancedAgentConfig(tool_retry_attempts=3, tool_retry_backoff_seconds=1.0,
                    escalate_on_retry_exhausted=True)

Creating an Agent

import os
from llmadapter import ChatLLMAdapter, AdvancedAgentConfig, create_llmadapter_agent
from langchain_core.tools import tool


@tool
def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"city": city, "temp_c": 23, "condition": "Sunny"}


@tool
def search_web(query: str) -> str:
    """Search the web for a query."""
    return f"Results for '{query}': [placeholder search results]"


primary_llm = ChatLLMAdapter(
    provider="azure_openai",
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
    api_version="2024-02-01",
    model="gpt-4o",
)

fallback_llm = ChatLLMAdapter(
    provider="openai",
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4o-mini",
)

config = AdvancedAgentConfig(
    max_iterations=8,
    parallel_tools=True,
    reflection_enabled=True,
    tool_retry_attempts=1,
)

agent = create_llmadapter_agent(
    primary_llm=primary_llm,
    fallback_llm=fallback_llm,  # optional: used on tool errors or escalation
    tools=[get_weather, search_web],
    config=config,
)

Running an Agent

result = agent.invoke("What is the weather in Paris, and find me a travel article about it.")

print(result["status"])  # "completed", "completed_with_reflection_limit", "stopped", or "interrupted"
print(result.get("output"))  # final answer text when available
print(result["steps"])  # list of AgentStepRecord — every think/act/observe cycle
print(result.get("checkpoint_id"))  # ID when a resumable interruption occurs

# Check how many steps were used
print(f"Completed in {len(result['steps'])} iterations")

result keys:

Key Description
status Final exit condition.
output Final text answer when one is available.
reason Present for stopped and interrupted paths.
messages Final LangChain message history.
steps List of AgentStepRecord with details of each iteration.
checkpoint_id ID returned only for a resumable interrupted tool-error result.
checkpoint_lineage Per-invocation lineage IDs, useful for relating nested specialist runs.

Some stop paths add diagnostics: token-budget stops include input_tokens and limit; retry-escalation stops include failed_tools.

Status values:

Status Meaning
"completed" Agent returned an answer that either skipped reflection or passed it.
"completed_with_reflection_limit" Agent returned its current answer after exhausting configured reflection rewrites.
"interrupted" A tool failed while interrupt_on_tool_error=True; the result contains reason="tool_error" and a resumable checkpoint_id.
"stopped" A limit or configured escalation ended the run. Inspect reason: max_iterations_reached, max_runtime_exceeded, max_input_tokens_per_step_exceeded, or tool_retry_exhausted.

Async Agent Execution

import asyncio


async def run_async_agent():
    result = await agent.ainvoke(
        "Research the latest trends in quantum computing."
    )
    print(result["status"], result["output"])


asyncio.run(run_async_agent())

For a caller-supplied tighter limit, use agent.invoke_with_runtime_limit(input, max_runtime_seconds=...) or await agent.ainvoke_with_runtime_limit(...). The supplied limit can only tighten AdvancedAgentConfig.max_runtime_seconds; a negative limit is rejected. The optional checkpoint_lineage argument lets a parent orchestration call preserve lineage when invoking nested agents.

Checkpoints — Pause and Resume

Checkpoints let you save the agent's state mid-run and resume it later — useful when:

  • A user needs to review an intermediate result before continuing
  • A tool calls for human approval
  • The agent hit max_iterations but needs more turns
  • You want to test recovery from partial failures

How checkpoints work:

The runtime writes internal snapshots before tool execution, after tool results, and around reflection transitions. The default store is InMemoryCheckpointStore; use FileCheckpointStore when checkpoints must survive process restarts. Internal snapshots are not all returned to callers: the public result contains a checkpoint_id only when a tool error interrupts the run with interrupt_on_tool_error=True.

Resuming an awaiting_tool_execution checkpoint runs its pending tool calls again. Resuming a tool_results_ready checkpoint reuses the recorded tool results, and resuming an awaiting_reflection checkpoint reruns reflection. Make pending tools idempotent when an interrupted run might be resumed.

Checkpoint + resume example:

from llmadapter import create_llmadapter_agent, AdvancedAgentConfig
from llmadapter.orchestration.tools import InMemoryCheckpointStore

# Shared store — pass the same instance to keep checkpoints alive
store = InMemoryCheckpointStore()

agent = create_llmadapter_agent(
    primary_llm=primary_llm,
    tools=[get_weather],
    config=AdvancedAgentConfig(max_iterations=3, interrupt_on_tool_error=True),
    checkpoint_store=store,
)

# A failing tool call produces a resumable checkpoint only when the model calls it.
result = agent.invoke("Research the history of AI and write a summary.")
checkpoint_id = result.get("checkpoint_id")
if checkpoint_id is not None:
    # Resume — agent picks up from the saved runtime state.
    resumed = agent.resume(
        checkpoint_id,
        user_feedback="Continue from where you left off.",  # optional extra instruction
    )
    print(resumed["status"])
    print(resumed.get("output"))
else:
    print("This run did not produce a resumable interruption.")

Use await agent.aresume(checkpoint_id, user_feedback=...) to resume through the asynchronous runtime.

Manual checkpoint management:

from llmadapter.orchestration.tools import InMemoryCheckpointStore

store = InMemoryCheckpointStore()

# Save arbitrary state
store.save("my-session-id", {"messages": [], "step": 5})

# Load it back
state = store.load("my-session-id")

# Delete when done
store.delete("my-session-id")

Important: InMemoryCheckpointStore stores data in process memory. FileCheckpointStore(directory) writes atomic JSON files with a SHA-256 checksum and restores serialized LangChain messages. It is not encrypted; protect the directory with operating-system permissions and do not treat checksum validation as access control. Dataclasses and Pydantic values serialize to JSON-compatible dictionaries; unsupported values are stringified, so use a custom CheckpointStore when exact custom type restoration or a database/secret-backed store is required.

Event Handlers — Observability

Event handlers expose agent loop and local tool-execution boundaries without modifying agent behavior. They do not emit provider request/response telemetry, checkpoint-store I/O, or a terminal result event.

Built-in handlers:

from llmadapter import CollectingAgentEventHandler, LoggingAgentEventHandler

# 1. CollectingAgentEventHandler — stores all events in memory
collector = CollectingAgentEventHandler()

agent = create_llmadapter_agent(
    primary_llm=primary_llm,
    tools=[get_weather],
    event_handler=collector,
)

result = agent.invoke("What is the weather in Berlin?")

# Inspect what happened
for event in collector.events:
    print(f"[{event.event}] step={event.step} payload={event.payload}")

# 2. LoggingAgentEventHandler — logs to Python's logging system
import logging

logging.basicConfig(level=logging.INFO)

logger_handler = LoggingAgentEventHandler(logger_name="llmadapter.agent")

agent = create_llmadapter_agent(
    primary_llm=primary_llm,
    tools=[get_weather],
    event_handler=logger_handler,
)

Events emitted:

Event Fired when payload keys
step_start An iteration begins messages (message count)
tool_start A registered tool is about to run tool_name, args
tool_end A tool returns successfully tool_name, result, duration_s
tool_error A started tool raises or times out tool_name, error, duration_s
step_end A finished answer is returned, or a tool step reaches its saved post-tool boundary finished; tool-step events also include checkpoint_id

CollectingAgentEventHandler stores the full tool arguments, results, and errors in memory. Use it carefully with sensitive data. LoggingAgentEventHandler logs redacted shape summaries rather than raw payload values.

Custom event handler:

Implement the AgentEventHandler protocol to build your own observer:

from typing import Any


class MetricsEventHandler:
    def on_step_start(self, step: int, context: dict[str, Any]) -> None:
        pass  # e.g., start a timer

    def on_tool_start(self, step: int, tool_name: str, args: dict[str, Any]) -> None:
        print(f"  -> calling {tool_name}")

    def on_tool_end(self, step: int, tool_name: str, result: Any, duration_s: float) -> None:
        print(f"  <- {tool_name} returned in {duration_s:.2f}s")

    def on_tool_error(self, step: int, tool_name: str, error: str, duration_s: float) -> None:
        print(f"  ! {tool_name} failed: {error}")

    def on_step_end(self, step: int, context: dict[str, Any]) -> None:
        pass

    # Optional: called when a routing policy returns a decision.
    # Built-in collectors/loggers do not record this optional callback.
    def on_routing_decision(self, step: int, decision: dict[str, Any]) -> None:
        pass

Routing Policies — Dynamic Model Selection

Routing chooses the LLM for the next agent step. A policy sees whether the previous step had a tool error, so a tool error cannot change the model that already produced that step. If no fallback LLM is configured, every built-in policy selects the primary model.

Available Policies

EscalationRoutingPolicy — switch to fallback on tool errors:

from llmadapter import EscalationRoutingPolicy

agent = create_llmadapter_agent(
    primary_llm=primary_llm,
    fallback_llm=fallback_llm,
    tools=[get_weather],
    routing_policy=EscalationRoutingPolicy(),
)
# Primary is used by default.
# If a tool error occurs, the following model step uses fallback.

When to use: Your primary model is powerful but expensive. Use a cheaper fallback that is better at correcting errors.

RoundRobinRoutingPolicy — alternate between primary and fallback:

from llmadapter import RoundRobinRoutingPolicy

agent = create_llmadapter_agent(
    primary_llm=primary_llm,
    fallback_llm=fallback_llm,
    tools=[get_weather],
    routing_policy=RoundRobinRoutingPolicy(),
)
# Step 1: primary, Step 2: fallback, Step 3: primary, ...

The round-robin index lives in the agent's _routing_state, so it survives multiple invocations of the same agent instance and is preserved in checkpoints.

When to use: Load balancing between two equivalent deployments to avoid rate limits.

CostAwareRoutingPolicy — use fallback after a step threshold:

from llmadapter import CostAwareRoutingPolicy

agent = create_llmadapter_agent(
    primary_llm=primary_llm,  # powerful, expensive (first N steps)
    fallback_llm=fallback_llm,  # cheaper (later steps)
    tools=[get_weather],
    routing_policy=CostAwareRoutingPolicy(fallback_after_step=3),
)
# Steps 1-2: primary model. Steps 3+: fallback model.

When to use: Early steps need a powerful model for reasoning; later steps (simple summaries) can use a cheaper one. This is a step-count threshold, not live price, token, or usage telemetry.

Custom routing function (maximum flexibility):

def my_router(messages, step):
    # Use fallback for odd-numbered steps
    if step % 2 == 1:
        return fallback_llm
    return primary_llm


agent = create_llmadapter_agent(
    primary_llm=primary_llm,
    fallback_llm=fallback_llm,
    tools=[get_weather],
    router=my_router,  # overrides routing_policy if both are set
)

router takes precedence over routing_policy and receives the current message list plus the 1-based step number. A custom router must return a ChatLLMAdapter; it is responsible for any error-aware or cost-aware logic. Policies may optionally emit on_routing_decision(step, decision) to a custom event handler. The decision payload contains policy, reason, metadata, and the selected model name.

Tool Retry and Escalation

config = AdvancedAgentConfig(
    tool_retry_attempts=3,  # retry failed idempotent tools up to 3 times
    tool_retry_backoff_seconds=0.5,  # wait 500ms between retries
    retry_on_tool_not_found=False,  # don't retry if the tool name is wrong
    escalate_on_retry_exhausted=True,  # stop if retries fail and no fallback is configured
)

To declare that a tool is safe to repeat, set tool.llmadapter_idempotent = True or tool.metadata["llmadapter_idempotent"] = True. Tool timeouts are never retried: a synchronous timeout only bounds the agent's wait and the underlying work may still be running. retry_non_idempotent_tools=True preserves legacy broad retry behavior, but can repeat side effects and should be used only deliberately.

You can also retry tool results manually after a run:

from llmadapter.orchestration.tools import ToolExecutionResult, is_retryable_tool_error, retry_failed_tool_calls

# Check if a failed tool result is worth retrying
failed = ToolExecutionResult(
    tool_call_id="call_1", tool_name="fetch", args={}, ok=False, content="", error="temporary failure"
)
if is_retryable_tool_error(failed, idempotent=True):
    print("This error is retryable")

# Re-run failed calls from a completed result list
retried_results = retry_failed_tool_calls(
    tool_calls=[{"id": "call_1", "name": "fetch", "args": {}}],
    current_results=[failed],
    tool_map=agent.tool_map,
    step=1,
    timeout_s=10,
    max_attempts=2,
    backoff_seconds=0.3,
    retry_on_tool_not_found=False,
)

Layer 4 — Multi-Agent Orchestration

When a task is too large for one agent, break it into specialists — focused agents that each handle a domain. A planner agent can coordinate them.

Specialists as Tools

The easiest way to compose agents: wrap each specialist agent as a LangChain tool that the planner can call.

from llmadapter import (
    ChatLLMAdapter,
    AdvancedAgentConfig,
    create_llmadapter_agent,
    make_specialist_tools,
    make_subagent_tool,
)

# Create specialist agents
finance_llm = ChatLLMAdapter(provider="openai", api_key="sk-...", model="gpt-4o-mini")
travel_llm = ChatLLMAdapter(provider="openai", api_key="sk-...", model="gpt-4o-mini")

finance_agent = create_llmadapter_agent(
    primary_llm=finance_llm,
    config=AdvancedAgentConfig(max_iterations=4),
)
travel_agent = create_llmadapter_agent(
    primary_llm=travel_llm,
    config=AdvancedAgentConfig(max_iterations=4),
)

# Wrap them as tools the planner can call
specialist_tools = make_specialist_tools(
    {
        "finance_specialist": (
            "Handles budgets, costs, and price trade-offs.",
            finance_agent,
        ),
        "travel_specialist": (
            "Handles itinerary planning and logistics.",
            travel_agent,
        ),
    }
)

# Create the planner agent with specialists as tools
planner_llm = ChatLLMAdapter(provider="openai", api_key="sk-...", model="gpt-4o")
planner = create_llmadapter_agent(
    primary_llm=planner_llm,
    tools=specialist_tools,
    config=AdvancedAgentConfig(parallel_tools=True, max_iterations=8),
)

result = planner.invoke("Plan a low-budget 3-day trip to Barcelona.")
print(result["output"])

For a single subagent tool:

from llmadapter import make_subagent_tool

research_tool = make_subagent_tool(
    name="research_agent",
    description="Performs deep research on any topic.",
    agent=create_llmadapter_agent(
        primary_llm=finance_llm,
        config=AdvancedAgentConfig(max_iterations=4),
    ),
)

planner = create_llmadapter_agent(
    primary_llm=planner_llm,
    tools=[research_tool],
    config=AdvancedAgentConfig(),
)

Capability Contracts (SpecialistContract)

Contracts are declarative labels for filtering and aggregation. A specialist is selected only when its contract contains every value in required_capabilities; a specialist without a contract is excluded when a capability requirement is supplied. Contracts do not validate model behavior, route work, or inject their metadata into prompts.

from llmadapter import SpecialistContract, invoke_specialists

contracts = {
    "finance": SpecialistContract(
        capabilities={"finance", "budget", "cost"},
        weight=1.5,  # consumed by WeightedScoreAggregator when selected
        metadata={"model": "gpt-4o"},
    ),
    "travel": SpecialistContract(
        capabilities={"travel", "itinerary", "logistics"},
        weight=1.0,
        metadata={"model": "gpt-4o-mini"},
    ),
    "legal": SpecialistContract(
        capabilities={"legal", "compliance"},
        weight=2.0,
    ),
}

specialists = {
    "finance": finance_agent,
    "travel": travel_agent,
    # "legal": legal_agent,   # not available
}

# Only call specialists that match the required capabilities
rows = invoke_specialists(
    task="What is the estimated cost for a 3-day Barcelona trip?",
    specialists=specialists,
    contracts=contracts,
    required_capabilities={"finance"},  # only finance will be called
)

for row in rows:
    print(f"{row['specialist']} (weight={row['weight']}): {row['output']}")

SpecialistContract parameters:

Parameter Type Description
capabilities set[str] Tags used for subset matching against required_capabilities.
weight float Finite, non-negative value attached to rows; WeightedScoreAggregator consumes it. Default 1.0.
metadata dict Passive application metadata; runtime selection does not inspect it.

Async Specialists and Shared Deadlines

Use ainvoke_specialists(...) and aaggregate_specialists(...) for asynchronous specialist runs. They accept the same arguments as their synchronous counterparts. max_runtime_seconds creates one shared monotonic deadline; each specialist receives the time remaining from that deadline rather than an independent full budget. This bounds the agent loop but cannot forcibly stop already-running synchronous or provider work.

Shared State Across Agents (SharedState)

SharedState is a thread-safe host-side key-value store. When specialists are launched, the runtime projects a JSON snapshot into each task prompt as SharedStateContext(JSON). It is not a live object or shared model memory: changes made while a specialist runs are not pushed into that specialist's already-created prompt.

from llmadapter import SharedState

state = SharedState({
    "destination": "Barcelona",
    "currency": "EUR",
    "budget_eur": None,  # will be filled by finance specialist
})

# Pass it to invoke_specialists — each specialist gets a prompt projection.
rows = invoke_specialists(
    task="Calculate total trip budget.",
    specialists=specialists,
    contracts=contracts,
    shared_state=state,
)

# After each specialist returns, the host runtime stores its output here.
print(state.get("specialist_outputs.finance"))
print(state.get("destination"))  # reads initial value

# Application code can update shared state between orchestration calls.
state.set("budget_eur", 850)
state.update({"hotel_booked": True, "flights_booked": False})

# Thread-safe snapshot for logging
snapshot = state.snapshot()
print(snapshot)

SharedState methods:

Method Description
get(key, default=None) Read a value by key.
set(key, value) Write a single key.
update(dict) Write multiple keys at once.
snapshot() Return a copy of the entire state dict.

Use shared_state_keys to allowlist prompt-visible keys and shared_state_max_bytes (default 16_000) to cap the JSON projection. Keep secrets and large mutable objects out of prompt-visible state.

Aggregation Strategies

After multiple specialists respond to the same task, use an aggregator to combine their answers into one result.

from llmadapter import (
    aggregate_specialists,
    MajorityVoteAggregator,
    WeightedScoreAggregator,
    JsonMergeAggregator,
    AggregatedResult,
)

rows = invoke_specialists(task="...", specialists=specialists, contracts=contracts)

All three aggregators accept include_failed=False by default. In that mode, rows with status failed, error, interrupted, stopped, or timeout are excluded; AggregatedResult.inputs contains the eligible rows, not necessarily the original input list.

MajorityVoteAggregator — picks the most common answer:

aggregator = MajorityVoteAggregator(value_key="output")
result: AggregatedResult = aggregator.aggregate(rows)

print(result.consensus)  # the winning answer
print(result.confidence)  # fraction of specialists that agreed (0.0-1.0)
print(result.metadata)  # {"votes": {"answer_A": 2, "answer_B": 1}}

When to use: Classification tasks, yes/no decisions, labeling — when you want the answer most agents agreed on. Ties are deterministic but not semantic: the implementation chooses the lexicographically greatest serialized output bucket.

WeightedScoreAggregator — picks the highest-weighted answer:

aggregator = WeightedScoreAggregator(value_key="output", weight_key="weight")
result = aggregator.aggregate(rows)

print(result.consensus)  # the answer with the highest cumulative weight
print(result.confidence)  # weighted fraction

When to use: When some specialists are more authoritative than others (e.g., a fine-tuned finance model should outweigh a general one).

Like majority voting, a weighted tie is resolved by the lexicographically greatest serialized output bucket. Weights must be finite and non-negative.

JsonMergeAggregator — merges all JSON responses into one dict:

aggregator = JsonMergeAggregator(conflict_policy="last_writer_wins")
result = aggregator.aggregate(rows)

# When eligible outputs are JSON objects, consensus is their merged dict.
print(result.consensus)

JsonMergeAggregator accepts last_writer_wins (default), first_writer_wins, or error. It parses string outputs as JSON when possible, merges only JSON-object outputs in input order, and records conflicts in result.metadata. If no eligible output is an object, it falls back to majority voting and reports strategy="json_merge_fallback_majority".

When to use: When each specialist provides a different piece of information (e.g., travel agent provides itinerary, finance agent provides costs) and you want one unified object.

AggregatedResult attributes:

Attribute Description
consensus The final aggregated answer.
inputs Eligible specialist rows after the aggregator's failure filter.
strategy Name of the aggregation strategy used.
confidence Score between 0-1 indicating confidence (meaning varies by strategy).
metadata Strategy-specific extra data (e.g., vote counts).

Using aggregate_specialists helper:

from llmadapter import aggregate_specialists

result = aggregate_specialists(
    task="...",
    specialists=specialists,
    contracts=contracts,
    aggregator=WeightedScoreAggregator(),
)
print(result.consensus)

Full Multi-Agent Example

import os
from llmadapter import (
    ChatLLMAdapter,
    AdvancedAgentConfig,
    SpecialistContract,
    SharedState,
    create_llmadapter_agent,
    invoke_specialists,
    aggregate_specialists,
    WeightedScoreAggregator,
    CollectingAgentEventHandler,
    EscalationRoutingPolicy,
)


# --- Build specialist agents ---
def make_llm(model: str) -> ChatLLMAdapter:
    return ChatLLMAdapter(
        provider="azure_openai",
        api_key=os.environ["AZURE_OPENAI_API_KEY"],
        endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
        api_version="2024-02-01",
        model=model,
    )


collector = CollectingAgentEventHandler()

finance_agent = create_llmadapter_agent(
    primary_llm=make_llm("gpt-4o"),
    config=AdvancedAgentConfig(max_iterations=4, reflection_enabled=False),
    event_handler=collector,
)
travel_agent = create_llmadapter_agent(
    primary_llm=make_llm("gpt-4o"),
    config=AdvancedAgentConfig(max_iterations=4, reflection_enabled=False),
    routing_policy=EscalationRoutingPolicy(),
)

# --- Contracts ---
contracts = {
    "finance": SpecialistContract(capabilities={"finance", "budget"}, weight=1.5),
    "travel": SpecialistContract(capabilities={"travel", "itinerary"}, weight=1.0),
}
specialists = {"finance": finance_agent, "travel": travel_agent}

# --- Shared context ---
shared_state = SharedState({"destination": "Lisbon", "days": 3, "currency": "EUR"})

# --- Run ---
rows = invoke_specialists(
    task="Provide a short one-sentence recommendation for this trip.",
    specialists=specialists,
    contracts=contracts,
    required_capabilities={"finance"},  # only finance needed
    shared_state=shared_state,
)

result = WeightedScoreAggregator().aggregate(rows)
print("Final answer:", result.consensus)
print("Confidence:", result.confidence)

# --- Inspect events ---
step_events = [e for e in collector.events if e.event == "step_start"]
print(f"Finance agent used {len(step_events)} steps")

Extending with a New Provider

  1. Create a provider file under llmadapter/providers/my_provider.py implementing the relevant BaseChatProvider, BaseEmbeddingProvider, and/or BaseResponsesProvider contract from llmadapter/base.py.

  2. Register each implemented surface in llmadapter/core/registry.py:

from ..providers.my_provider import (
     AsyncMyChatProvider,
     AsyncMyResponsesProvider,
     MyChatProvider,
     MyEmbeddingProvider,
     MyResponsesProvider,
)

CHAT_PROVIDER_REGISTRY["my_provider"] = MyChatProvider
EMBEDDING_PROVIDER_REGISTRY["my_provider"] = MyEmbeddingProvider
ASYNC_CHAT_PROVIDER_REGISTRY["my_provider"] = AsyncMyChatProvider
RESPONSES_PROVIDER_REGISTRY["my_provider"] = MyResponsesProvider
ASYNC_RESPONSES_PROVIDER_REGISTRY["my_provider"] = AsyncMyResponsesProvider

Only register interfaces that actually exist. Provider modules are imported optionally, so a missing optional SDK should leave its registry entries absent rather than make package import fail.

  1. Add a config model in config.py and register it in CONFIG_REGISTRY. Also register a ProviderCapabilityManifest through register_provider_capabilities(...); sync/async chat and embedding dispatch consult that manifest before invoking a provider.

  2. Add tests covering every registered surface and its normalized response shape before shipping.

Project Structure

llmadapter/
├── __init__.py                  # public API — re-exports everything below
├── core/                        # adapter engine
│   ├── sync_adapter.py          #   LLMAdapter, _ChatCompletions, _Embeddings
│   ├── async_adapter.py         #   AsyncLLMAdapter, _AsyncChatCompletions (non-blocking I/O)
│   ├── registry.py              #   chat, embedding, and Responses sync/async registries
│   ├── retry.py                 #   retry + full-jitter exponential back-off
│   ├── failover.py              #   backup-provider failover helpers
│   ├── execution.py             #   target-aware normalized request plans
│   ├── responses_capabilities.py #  Responses request validation and capability gates
│   ├── streaming.py             #   lazy stream sessions and pre-output failover
│   ├── circuit_breaker.py       #   target-aware circuit breaker
│   ├── preprocess.py            #   message prep + provider-client cache (ProviderCache)
│   └── redaction.py             #   sensitive-field redaction for logs
├── ratelimit/                   # rate limiting + tokenization
│   ├── limiter.py               #   RedisRateLimiter (pooled connections), RateLimitMetrics
│   ├── tokenizer.py             #   token counting / truncation (cached encodings)
│   ├── batching.py              #   AdapterUtility — token-aware embedding batches
│   └── logging.py               #   log_structured
├── structured.py                # structured-output helpers (pydantic_to_response_format, …) + to_jsonable
├── integrations/
│   └── langchain.py             # ChatLLMAdapter — LangChain BaseChatModel wrapper
├── orchestration/               # multi-agent runtime
│   ├── runtime.py               #   LLMAdapterAgent, AdvancedAgentConfig, create_llmadapter_agent
│   ├── tools.py                 #   tool execution, budgets, checkpoints, event handlers
│   ├── routing.py               #   BaseRoutingPolicy and routing policy implementations
│   ├── aggregators.py           #   MajorityVote / WeightedScore / JsonMerge aggregators
│   └── contracts.py             #   SharedState, SpecialistContract
├── providers/                   # one module per provider
│   ├── _shared.py               #   shared OpenAI-shape response mappers (OpenAI + Azure OpenAI)
│   ├── _responses_base.py        #   shared OpenAI/Azure Responses provider verbs
│   ├── capabilities.py           #   provider capability manifests
│   ├── azureopenai_provider.py
│   ├── azurefoundry_provider.py
│   ├── openai_provider.py
│   ├── vertexai_provider.py
│   └── anthropic_provider.py
├── config.py                    # Typed provider config models
├── models.py                    # Request/response models
├── responses_tools.py            # Typed built-in Responses tool requests
├── errors.py                    # Typed exception hierarchy
├── base.py                      # Provider interfaces (chat, embedding, Responses)
└── constants.py                 # Default rate limits / retry settings
examples/
└── llmadapter_example.py        # full walkthrough (chat, embed, async, LangChain, agents)

API Reference

For the complete exported API surface with all class signatures, parameters, and return types, see:

docs/API_REFERENCE.md

Selected top-level imports:

from llmadapter import (
    # Core
    LLMAdapter,
    AsyncLLMAdapter,

    # LangChain layer
    ChatLLMAdapter,

    # Agent runtime
    AdvancedAgentConfig,
    LLMAdapterAgent,
    create_llmadapter_agent,

    # Multi-agent helpers
    make_subagent_tool,
    make_specialist_tools,
    ainvoke_specialists,
    invoke_specialists,
    aaggregate_specialists,
    aggregate_specialists,

    # Aggregators
    AggregatedResult,
    BaseAggregator,
    JsonMergeAggregator,
    MajorityVoteAggregator,
    WeightedScoreAggregator,

    # Orchestration
    SharedState,
    SpecialistContract,

    # Routing policies
    BaseRoutingPolicy,
    RoundRobinRoutingPolicy,
    EscalationRoutingPolicy,
    CostAwareRoutingPolicy,

    # Observability
    CollectingAgentEventHandler,
    LoggingAgentEventHandler,
    CheckpointStore,
    InMemoryCheckpointStore,
    FileCheckpointStore,

    # Errors
    LLMAdapterError,
    AuthenticationError,
    ModelNotFoundError,
    RateLimitExceededError,
    APIConnectionError,
    TimeOutError,
    ChatCompletionError,
    EmbeddingGenerationError,
    ContentFilterError,
    InvalidResponseError,
)

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for rituraj.raman from gravatar.com rituraj.raman

Credits

Author: Rituraj.Raman

Requires

Python >=3.11

Provides Extra

base anthropic azure-foundry vertexai langchain redis all dev

Classifiers

Metadata

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for rituraj.raman from gravatar.com rituraj.raman

Credits

Author: Rituraj.Raman

Requires

Python >=3.11

Provides Extra

base anthropic azure-foundry vertexai langchain redis all dev

Classifiers

Release files for llmadapterevs 0.1.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for llmadapterevs 0.1.0
File Size Uploaded Provenance
llmadapterevs-0.1.0.tar.gz 229.3 kB Without provenance Details

Built distribution (wheel)

Table of built distributions (wheels) for llmadapterevs 0.1.0
File Interpreter ABI Platform
llmadapterevs-0.1.0-py3-none-any.whl 182.8 kB Without provenance Python 3 none any Details

Total release size: 412.1 kB

Release files / llmadapterevs-0.1.0.tar.gz

Download URL llmadapterevs-0.1.0.tar.gz
Size 229.3 kB
Tags Source
SHA-256 checksum
How to use checksums 		18d0ce24d5ff2cc579d0a48821536f9b6f593492efde10a2d234886445a1bdf5
BLAKE2b-256 checksum
How to use checksums 		fbc434c2ddcbb7c832f4dd1425ecd65ab0f939085ef6c260e83cd982049c5977
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		No
Uploaded via uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

Release files / llmadapterevs-0.1.0-py3-none-any.whl

Download URL llmadapterevs-0.1.0-py3-none-any.whl
Size 182.8 kB
Tags Python 3
SHA-256 checksum
How to use checksums 		40782f72a06949d1bd844347ee7fa69341037df4c7d9f5cee16c38f09f81d7cd
BLAKE2b-256 checksum
How to use checksums 		159e9f6a4473e1fed3215a46b9bb473f7354e6a697fa32ed1405675ff5a9fe03
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		No
Uploaded via uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

Security for llmadapterevs 0.1.0

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.

Release history Release notifications | RSS feed

This release

0.1.0 This release

2 release files No provenance

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page