Provider-agnostic agents: swapping models in CrewAI safely

CrewAI agents work well until you need to swap llm models in crewai across providers. The framework’s LLM abstraction helps, but real-world differences in token limits, tool-calling formats, and response styles break agent behavior silently. This guide shows how to build a provider-agnostic layer that lets you switch between OpenAI, Anthropic, Google, and local models without rewriting your crew definitions.

Step 1: Abstract the model configuration

Don’t hardcode model names in your agent definitions. Create a configuration layer that maps logical roles to concrete model identifiers.

# config/models.py
from dataclasses import dataclass
from typing import Literal

@dataclass(frozen=True)
class ModelSpec:
    provider: Literal["openai", "anthropic", "google", "ollama", "openrouter"]
    model_id: str
    max_tokens: int
    supports_tools: bool
    supports_json_mode: bool
    context_window: int

MODEL_REGISTRY: dict[str, ModelSpec] = {
    "gpt-4o": ModelSpec(
        provider="openai",
        model_id="gpt-4o",
        max_tokens=4096,
        supports_tools=True,
        supports_json_mode=True,
        context_window=128_000,
    ),
    "claude-3-5-sonnet": ModelSpec(
        provider="anthropic",
        model_id="claude-3-5-sonnet-20241022",
        max_tokens=8192,
        supports_tools=True,
        supports_json_mode=True,
        context_window=200_000,
    ),
    "gemini-1.5-pro": ModelSpec(
        provider="google",
        model_id="gemini-1.5-pro",
        max_tokens=8192,
        supports_tools=True,
        supports_json_mode=True,
        context_window=1_000_000,
    ),
    "llama-3.1-70b": ModelSpec(
        provider="ollama",
        model_id="llama3.1:70b",
        max_tokens=4096,
        supports_tools=False,
        supports_json_mode=False,
        context_window=128_000,
    ),
}

ROLE_MAPPING: dict[str, str] = {
    "planner": "gpt-4o",
    "researcher": "claude-3-5-sonnet",
    "coder": "gpt-4o",
    "critic": "claude-3-5-sonnet",
    "local_fallback": "llama-3.1-70b",
}

This registry captures the behavioral differences that matter: tool support, JSON mode, context windows. Your agents reference roles ("planner", "researcher"), not model IDs.

Step 2: Build a provider-agnostic LLM factory

CrewAI’s LLM class accepts an OpenAI-compatible endpoint. Wrap the provider-specific initialization so the rest of your code stays clean.

# llm/factory.py
import os
from crewai import LLM
from config.models import MODEL_REGISTRY, ROLE_MAPPING, ModelSpec

class LLMFactory:
    def __init__(self, default_role: str = "planner"):
        self.default_role = default_role
        self._cache: dict[str, LLM] = {}

    def get_llm(self, role: str | None = None, **overrides) -> LLM:
        role = role or self.default_role
        model_key = ROLE_MAPPING.get(role)
        if not model_key:
            raise ValueError(f"Unknown role: {role}. Available: {list(ROLE_MAPPING.keys())}")

        spec = MODEL_REGISTRY[model_key]
        cache_key = f"{model_key}:{hash(frozenset(overrides.items()))}"
        
        if cache_key in self._cache:
            return self._cache[cache_key]

        llm = self._build_llm(spec, **overrides)
        self._cache[cache_key] = llm
        return llm

    def _build_llm(self, spec: ModelSpec, **overrides) -> LLM:
        base_url = self._get_base_url(spec.provider)
        api_key = self._get_api_key(spec.provider)
        
        params = {
            "model": spec.model_id,
            "base_url": base_url,
            "api_key": api_key,
            "temperature": overrides.get("temperature", 0.7),
            "max_tokens": overrides.get("max_tokens", spec.max_tokens),
        }
        
        # Provider-specific tweaks
        if spec.provider == "anthropic":
            params["model"] = f"anthropic/{spec.model_id}"
        elif spec.provider == "google":
            params["model"] = f"gemini/{spec.model_id}"
        elif spec.provider == "ollama":
            params["model"] = f"ollama/{spec.model_id}"
            params["api_key"] = "ollama"  # dummy key for local
        
        return LLM(**params)

    def _get_base_url(self, provider: str) -> str:
        urls = {
            "openai": "https://api.openai.com/v1",
            "anthropic": "https://api.anthropic.com/v1",
            "google": "https://generativelanguage.googleapis.com/v1beta/openai/",
            "ollama": "http://localhost:11434/v1",
            "openrouter": "https://openrouter.ai/api/v1",
        }
        return urls[provider]

    def _get_api_key(self, provider: str) -> str:
        env_vars = {
            "openai": "OPENAI_API_KEY",
            "anthropic": "ANTHROPIC_API_KEY",
            "google": "GOOGLE_API_KEY",
            "ollama": "ollama",
            "openrouter": "OPENROUTER_API_KEY",
        }
        key = os.getenv(env_vars[provider])
        if not key and provider != "ollama":
            raise RuntimeError(f"Missing {env_vars[provider]} environment variable")
        return key or "ollama"

The factory handles the OpenAI-compatible endpoint translation each provider requires. Anthropic and Google need model prefixes; Ollama needs a dummy key. Your agent code never sees this complexity.

Step 3: Wire agents to roles, not models

Update your crew definition to request LLMs by role. This is where the swap llm models in crewai pattern pays off — changing a model means editing one line in ROLE_MAPPING.

# crews/research_crew.py
from crewai import Agent, Crew, Task
from llm.factory import LLMFactory

llm_factory = LLMFactory()

planner = Agent(
    role="Research Planner",
    goal="Break down complex topics into structured research plans",
    backstory="You're a senior research strategist who excels at decomposing ambiguous problems.",
    llm=llm_factory.get_llm("planner", temperature=0.3),
    verbose=True,
)

researcher = Agent(
    role="Deep Researcher",
    goal="Gather comprehensive, cited information on assigned subtopics",
    backstory="You're a meticulous researcher with access to multiple knowledge sources.",
    llm=llm_factory.get_llm("researcher", temperature=0.2),
    verbose=True,
)

coder = Agent(
    role="Code Implementer",
    goal="Translate research findings into working prototype code",
    backstory="You're a pragmatic engineer who writes clean, tested implementations.",
    llm=llm_factory.get_llm("coder", temperature=0.1),
    verbose=True,
)

critic = Agent(
    role="Critical Reviewer",
    goal="Identify gaps, errors, and improvements in research and code",
    backstory="You're a ruthless but constructive critic who catches what others miss.",
    llm=llm_factory.get_llm("critic", temperature=0.4),
    verbose=True,
)

plan_task = Task(
    description="Create a research plan for: {topic}. Output a JSON list of subtopics.",
    agent=planner,
    expected_output="Valid JSON array of subtopic strings",
)

research_task = Task(
    description="Research each subtopic thoroughly. Return findings with citations.",
    agent=researcher,
    expected_output="Markdown report with citations per subtopic",
    context=[plan_task],
)

code_task = Task(
    description="Implement a prototype based on research findings.",
    agent=coder,
    expected_output="Complete Python file with tests",
    context=[research_task],
)

review_task = Task(
    description="Review the code and research for correctness, completeness, and style.",
    agent=critic,
    expected_output="Markdown review with actionable feedback",
    context=[code_task, research_task],
)

crew = Crew(
    agents=[planner, researcher, coder, critic],
    tasks=[plan_task, research_task, code_task, review_task],
    verbose=True,
)

Each agent gets the right model for its job. The planner uses GPT-4o for structured reasoning; the researcher uses Claude for long-context synthesis. Swap the mapping and the crew behavior shifts accordingly.

Step 4: Handle model-specific quirks at the factory level

Different models fail in different ways. GPT-4o follows JSON schemas reliably. Claude sometimes wraps JSON in markdown. Local models may not support tools at all. Centralize these workarounds.

# llm/adapters.py
import json
import re
from typing import Any

def extract_json(response: str) -> dict[str, Any] | list[Any]:
    """Extract JSON from model response, handling common wrapper formats."""
    # Try direct parse first
    try:
        return json.loads(response)
    except json.JSONDecodeError:
        pass
    
    # Strip markdown code fences
    fence_match = re.search(r"```(?:json)?\s*(\{.*?\}|\[.*?\])\s*```", response, re.DOTALL)
    if fence_match:
        try:
            return json.loads(fence_match.group(1))
        except json.JSONDecodeError:
            pass
    
    # Find first JSON-like structure
    brace_match = re.search(r"(\{.*\}|\[.*\])", response, re.DOTALL)
    if brace_match:
        try:
            return json.loads(brace_match.group(1))
        except json.JSONDecodeError:
            pass
    
    raise ValueError(f"Could not extract JSON from response: {response[:200]}...")

def normalize_tool_calls(response: dict, provider: str) -> list[dict]:
    """Normalize tool call formats across providers."""
    if provider == "anthropic":
        # Anthropic returns tool_use blocks in content array
        tool_calls = []
        for block in response.get("content", []):
            if block.get("type") == "tool_use":
                tool_calls.append({
                    "id": block["id"],
                    "name": block["name"],
                    "arguments": block["input"],
                })
        return tool_calls
    
    # OpenAI format (also used by OpenRouter, Ollama with tools)
    return response.get("tool_calls", [])

def supports_feature(spec, feature: str) -> bool:
    """Check if a model spec supports a feature."""
    feature_map = {
        "tools": spec.supports_tools,
        "json_mode": spec.supports_json_mode,
        "vision": "vision" in spec.model_id.lower() or "gpt-4o" in spec.model_id,
    }
    return feature_map.get(feature, False)

Use these adapters in your task callbacks or custom tools. The planner task expects JSON — extract_json handles the variance. Tool-calling agents need normalize_tool_calls when processing responses manually.

Step 5: Implement graceful degradation with fallback chains

Production crews need resilience. When a provider hits rate limits or degrades, fall back automatically without losing crew state.

# llm/fallback.py
from llm.factory import LLMFactory, MODEL_REGISTRY
from config.models import ROLE_MAPPING

class FallbackLLMFactory(LLMFactory):
    def __init__(self, default_role: str = "planner", fallback_chains: dict[str, list[str]] | None = None):
        super().__init__(default_role)
        self.fallback_chains = fallback_chains or {
            "planner": ["gpt-4o", "claude-3-5-sonnet", "gemini-1.5-pro"],
            "researcher": ["claude-3-5-sonnet", "gpt-4o", "gemini-1.5-pro"],
            "coder": ["gpt-4o", "claude-3-5-sonnet"],
            "critic": ["claude-3-5-sonnet", "gpt-4o"],
        }
        self._failure_counts: dict[str, int] = {}

    def get_llm(self, role: str | None = None, **overrides):
        role = role or self.default_role
        chain = self.fallback_chains.get(role, [ROLE_MAPPING[role]])
        
        for model_key in chain:
            if self._is_healthy(model_key):
                try:
                    spec = MODEL_REGISTRY[model_key]
                    return self._build_llm(spec, **overrides)
                except Exception as e:
                    self._record_failure(model_key)
                    continue
        
        # All failed — return last attempt anyway, let it error naturally
        spec = MODEL_REGISTRY[chain[-1]]
        return self._build_llm(spec, **overrides)

    def _is_healthy(self, model_key: str) -> bool:
        return self._failure_counts.get(model_key, 0) < 3

    def _record_failure(self, model_key: str):
        self._failure_counts[model_key] = self._failure_counts.get(model_key, 0) + 1

    def reset_health(self, model_key: str | None = None):
        if model_key:
            self._failure_counts.pop(model_key, None)
        else:
            self._failure_counts.clear()

The fallback chain tries models in order, skipping ones that have failed recently. Three failures marks a model unhealthy. This is a simple circuit breaker — replace with a proper one (like pybreaker) for production.

Step 6: Verify the swap works end-to-end

Write a verification script that exercises each role with its assigned model and validates expected behaviors.

# scripts/verify_models.py
import os
from llm.factory import LLMFactory
from llm.adapters import extract_json, supports_feature
from config.models import MODEL_REGISTRY, ROLE_MAPPING

def test_role(role: str, factory: LLMFactory):
    print(f"\n=== Testing role: {role} ===")
    llm = factory.get_llm(role)
    spec = MODEL_REGISTRY[ROLE_MAPPING[role]]
    
    print(f"Model: {spec.model_id} ({spec.provider})")
    print(f"Context window: {spec.context_window:,}")
    print(f"Tools: {spec.supports_tools}, JSON mode: {spec.supports_json_mode}")
    
    # Test basic completion
    response = llm.call("Reply with exactly: OK")
    print(f"Basic completion: {response.strip()}")
    assert "OK" in response
    
    # Test JSON mode if supported
    if spec.supports_json_mode:
        json_prompt = 'Return JSON: {"status": "success", "count": 42}'
        response = llm.call(json_prompt)
        try:
            data = extract_json(response)
            assert data["status"] == "success"
            assert data["count"] == 42
            print("JSON mode: PASS")
        except Exception as e:
            print(f"JSON mode: FAIL - {e}")
    else:
        print("JSON mode: NOT SUPPORTED (skipped)")
    
    # Test tool calling format if supported
    if spec.supports_tools:
        print("Tool calling: SUPPORTED")
    else:
        print("Tool calling: NOT SUPPORTED")

def main():
    if not any(os.getenv(k) for k in ["OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GOOGLE_API_KEY", "OPENROUTER_API_KEY"]):
        print("No API keys set. Set at least one provider key to run verification.")
        return
    
    factory = LLMFactory()
    
    for role in ROLE_MAPPING:
        if role == "local_fallback":
            continue  # Skip unless Ollama running
        try:
            test_role(role, factory)
        except Exception as e:
            print(f"Role {role} FAILED: {e}")

if __name__ == "__main__":
    main()

Run this with python scripts/verify_models.py. It confirms each role resolves to the right model, basic completion works, JSON extraction succeeds where supported, and tool-calling capability matches the registry. Add this to your CI pipeline.

Step 7: Add runtime model switching for A/B testing

Sometimes you need to compare models on the same task without redeploying. Expose a runtime override via environment variable or request header.

# llm/runtime.py
import os
from llm.factory import LLMFactory

class RuntimeLLMFactory(LLMFactory):
    def get_llm(self, role: str | None = None, **overrides):
        role = role or self.default_role
        
        # Check for runtime override: MODEL_OVERRIDE_PLANNER=claude-3-5-sonnet
        override_key = f"MODEL_OVERRIDE_{role.upper()}"
        if override_key in os.environ:
            override_model = os.environ[override_key]
            if override_model in MODEL_REGISTRY:
                print(f"[Runtime] Overriding {role} -> {override_model}")
                spec = MODEL_REGISTRY[override_model]
                return self._build_llm(spec, **overrides)
            else:
                print(f"[Runtime] Unknown override model: {override_model}")
        
        return super().get_llm(role, **overrides)

Replace LLMFactory with RuntimeLLMFactory in your crew definition. Now you can test MODEL_OVERRIDE_PLANNER=gemini-1.5-pro python run_crew.py without code changes. This is invaluable for regression testing when providers release new model versions.

Step 8: Meter usage per model and role

You can’t optimize costs you don’t measure. Wrap the LLM call to capture token usage by role and model.

# llm/metered.py
from dataclasses import dataclass, field
from threading import Lock
from llm.factory import LLMFactory
from config.models import MODEL_REGISTRY

@dataclass
class UsageStats:
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    calls: int = 0

class MeteredLLMFactory(LLMFactory):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._usage: dict[str, UsageStats] = {}
        self._lock = Lock()

    def get_llm(self, role: str | None = None, **overrides):
        role = role or self.default_role
        base_llm = super().get_llm(role, **overrides)
        model_key = ROLE_MAPPING[role]
        spec = MODEL_REGISTRY[model_key]
        
        return MeteredLLM(base_llm, f"{role}:{model_key}", self._record_usage)

    def _record_usage(self, key: str, prompt: int, completion: int):
        with self._lock:
            if key not in self._usage:
                self._usage[key] = UsageStats()
            stats = self._usage[key]
            stats.prompt_tokens += prompt
            stats.completion_tokens += completion
            stats.total_tokens += prompt + completion
            stats.calls += 1

    def get_usage_report(self) -> dict:
        with self._lock:
            return {
                k: {
                    "prompt_tokens": v.prompt_tokens,
                    "completion_tokens": v.completion_tokens,
                    "total_tokens": v.total_tokens,
                    "calls": v.calls,
                }
                for k, v in self._usage.items()
            }

    def reset_usage(self):
        with self._lock:
            self._usage.clear()


class MeteredLLM:
    def __init__(self, wrapped, key: str, recorder):
        self._wrapped = wrapped
        self._key = key
        self._recorder = recorder

    def __getattr__(self, name):
        return getattr(self._wrapped, name)

    def call(self, *args, **kwargs):
        response = self._wrapped.call(*args, **kwargs)
        # CrewAI's LLM.call returns string; usage is in response metadata if available
        # For OpenAI-compatible endpoints, usage is often in a separate attribute
        usage = getattr(self._wrapped, "_last_usage", None)
        if usage:
            self._recorder(self._key, usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0))
        return response

The metered factory wraps each LLM and intercepts calls. Usage aggregates by role-model pair. Call factory.get_usage_report() at the end of a crew run to see cost distribution. This works with any OpenAI-compatible endpoint that returns usage metadata — including n4n.ai’s gateway which forwards provider cache-control hints and per-token metering.

Step 9: Lock model versions for reproducibility

Provider model aliases (like gpt-4o) can point to different snapshots over time. Pin exact versions in your registry for reproducible runs.

# config/models.py (updated)
MODEL_REGISTRY: dict[str, ModelSpec] = {
    "gpt-4o": ModelSpec(
        provider="openai",
        model_id="gpt-4o-2024-08-06",  # Pinned snapshot
        max_tokens=4096,
        supports_tools=True,
        supports_json_mode=True,
        context_window=128_000,
    ),
    "claude-3-5-sonnet": ModelSpec(
        provider="anthropic",
        model_id="claude-3-5-sonnet-20241022",  # Pinned snapshot
        max_tokens=8192,
        supports_tools=True,
        supports_json_mode=True,
        context_window=200_000,
    ),
    # ... others
}

Update these pins deliberately during a scheduled dependency review. Treat model versions like any other dependency — test before pinning, document the change, and roll back if behavior regresses.

Step 10: Run a full crew with swapped models

Put it all together. This script runs the research crew with three different model configurations to demonstrate the swap.

# scripts/run_comparison.py
import os
from crews.research_crew import crew
from llm.factory import LLMFactory
from llm.fallback import FallbackLLMFactory
from llm.runtime import RuntimeLLMFactory
from llm.metered import MeteredLLMFactory
from config.models import ROLE_MAPPING

def run_with_factory(factory_class, name: str, **factory_kwargs):
    print(f"\n{'='*60}")
    print(f"Running crew with {name}")
    print(f"{'='*60}")
    
    # Rebuild crew with new factory
    from crews.research_crew import llm_factory
    import crews.research_crew as crew_module
    
    crew_module.llm_factory = factory_class(**factory_kwargs)
    
    # Re-instantiate agents with new LLMs
    for agent in crew.agents:
        role = agent.role.lower().replace(" ", "_")
        if role in ROLE_MAPPING:
            agent.llm = crew_module.llm_factory.get_llm(role)
    
    result = crew.kickoff(inputs={"topic": "Model-agnostic agent architectures"})
    print(f"\nResult preview: {str(result)[:500]}...")
    
    if hasattr(crew_module.llm_factory, "get_usage_report"):
        print(f"\nUsage: {crew_module.llm_factory.get_usage_report()}")
    
    return result

if __name__ == "__main__":
    # Configuration 1: Default (OpenAI planner/coder, Anthropic researcher/critic)
    run_with_factory(MeteredLLMFactory, "default mapping")
    
    # Configuration 2: All Anthropic
    os.environ["MODEL_OVERRIDE_PLANNER"] = "claude-3-5-sonnet"
    os.environ["MODEL_OVERRIDE_CODER"] = "claude-3-5-sonnet"
    os.environ["MODEL_OVERRIDE_RESEARCHER"] = "claude-3-5-sonnet"
    os.environ["MODEL_OVERRIDE_CRITIC"] = "claude-3-5-sonnet"
    run_with_factory(RuntimeLLMFactory, "all-claude override")
    
    # Configuration 3: With fallback chain
    run_with_factory(FallbackLLMFactory, "fallback-enabled", 
                     fallback_chains={
                         "planner": ["gpt-4o", "claude-3-5-sonnet"],
                         "researcher": ["claude-3-5-sonnet", "gpt-4o"],
                         "coder": ["gpt-4o", "claude-3-5-sonnet"],
                         "critic": ["claude-3-5-sonnet", "gpt-4o"],
                     })

Run this to see the same crew produce different outputs with different model assignments. The metered factory shows token costs per configuration. The fallback factory survives a simulated provider outage (kill your API key mid-run to test).

What you’ve built

You now have a provider-agnostic CrewAI setup where:

1. Roles map to models — change ROLE_MAPPING to swap llm models in crewai globally
2. Factory handles provider differences — endpoint URLs, auth, model prefixes, capability flags
3. Adapters normalize behavior — JSON extraction, tool-call formats, feature detection
4. Fallback chains provide resilience — automatic failover on provider degradation
5. Runtime overrides enable A/B testing — environment-variable model switching
6. Metering reveals costs — per-role, per-model token accounting
7. Pinned versions ensure reproducibility — no surprise model updates

The pattern scales. Add new providers by extending MODEL_REGISTRY and the factory’s _get_base_url/_get_api_key methods. Add new roles by updating ROLE_MAPPING. Your crew definitions never change.

Verification checklist

• python scripts/verify_models.py passes for all configured providers
• python scripts/run_comparison.py completes with three distinct outputs
• Fallback triggers when you invalidate an API key mid-run
• Runtime override (MODEL_OVERRIDE_PLANNER=gemini-1.5-pro) changes the planner’s model
• Usage report shows non-zero tokens for each role
• JSON tasks produce valid JSON across all models that claim support
• Tool-calling agents work on models with supports_tools=true

If all seven pass, your crew is genuinely provider-agnostic. Swap models freely.