Routing Different Models Through a CrewAI Workflow

CrewAI becomes substantially more useful when each agent can use a model suited to its job.

A researcher may need throughput, an analyst may need stronger reasoning, and a writer may benefit from high-quality long-form generation. Wiring those agents directly to separate providers usually means separate credentials, SDKs, endpoints, billing, and error handling.

I prefer a cleaner boundary:

• CrewAI owns agents, tasks, context, and execution.
• A unified OpenAI-compatible model endpoint owns model access.
• Model IDs define routing.

This post builds a sequential three-agent workflow using Gemini 3.7 Flash, Claude Opus 5, and GPT-5.6 through CometAPI, using one credential and one base URL: https://api.cometapi.com/v1.

The implementation includes:

• Per-agent model selection
• Bounded fallback for transient errors
• CrewAI checkpointing
• Token and execution usage reporting
• Model catalog validation
• Production-oriented error handling

The workflow

The example uses three agents:

Agent	Primary model	Fallback	Responsibility
Market Researcher	gemini-3.7-flash	gpt-5.6	Collect facts and sources
Product Analyst	claude-opus-5	gpt-5.6	Evaluate evidence and trade-offs
Technical Writer	gpt-5.6	gemini-3.7-flash	Produce the final decision memo

The execution chain is:

Topic
  ↓
Research
  ↓
Analysis
  ↓
Final memo

This is a routing example, not a benchmark. Model selection should be based on representative application tasks, context requirements, tool support, structured-output behavior, latency, reliability, cost, and quality.

An OpenAI-compatible endpoint standardizes request shape. It does not make models identical. Context limits, tools, reasoning controls, output behavior, latency, and pricing can still differ.

Install the dependencies

I’m using Python 3.10+:

python -m venv .venv
source .venv/bin/activate

On Windows:

.venv\Scripts\Activate.ps1

Install CrewAI with OpenAI compatibility, the SDK used by the fallback logic, and dotenv:

pip install "crewai[openai]" openai python-dotenv

For a real deployment, pin versions you have tested:

crewai==YOUR_TESTED_VERSION
openai==YOUR_TESTED_VERSION
python-dotenv==YOUR_TESTED_VERSION

CrewAI’s LLM configuration is actively evolving, so verify the constructor and provider configuration against the version you deploy.

Configure the endpoint

Create .env:

COMETAPI_KEY=your_cometapi_key
COMETAPI_BASE_URL=https://api.cometapi.com/v1

Load it in Python:

import os

from dotenv import load_dotenv

load_dotenv()

COMETAPI_KEY = os.environ["COMETAPI_KEY"]
COMETAPI_BASE_URL = os.getenv(
    "COMETAPI_BASE_URL",
    "https://api.cometapi.com/v1",
)

Keep credentials out of source control:

.env
.venv/
__pycache__/

The key should remain a server-side credential.

Keep routing separate from prompts

Model selection belongs in configuration, not in agent prompts:

PRIMARY_MODELS = {
    "researcher": "gemini-3.7-flash",
    "analyst": "claude-opus-5",
    "writer": "gpt-5.6",
}

FALLBACK_MODELS = {
    "researcher": "gpt-5.6",
    "analyst": "gpt-5.6",
    "writer": "gemini-3.7-flash",
}

That mapping can later move to environment configuration, YAML, JSON, a database, feature flags, or a routing service without changing the agent definitions.

Create the CrewAI LLM objects

CrewAI’s LLM object accepts a model name, API key, and custom base URL:

from crewai import LLM


def cometapi_llm(model_id: str) -> LLM:
    return LLM(
        model=model_id,
        base_url=COMETAPI_BASE_URL,
        api_key=COMETAPI_KEY,
        timeout=60.0,
        max_retries=0,
    )

I explicitly set max_retries=0 because fallback is handled by the application. Hidden SDK retries can otherwise delay model switching and multiply requests.

Define the agents and tasks

Each agent receives a model from the routing map:

from crewai import Agent, Task


def build_crew(model_map: dict[str, str]):
    researcher = Agent(
        role="Market Researcher",
        goal="Collect the facts needed to answer the topic",
        backstory=(
            "You create concise, source-aware research briefs "
            "and distinguish facts from assumptions."
        ),
        llm=cometapi_llm(model_map["researcher"]),
        max_iter=3,
        allow_delegation=False,
    )

    analyst = Agent(
        role="Product Analyst",
        goal="Turn research into a defensible recommendation",
        backstory=(
            "You evaluate evidence, assumptions, risks, "
            "and trade-offs."
        ),
        llm=cometapi_llm(model_map["analyst"]),
        max_iter=3,
        allow_delegation=False,
    )

    writer = Agent(
        role="Technical Writer",
        goal="Produce a concise technical decision memo",
        backstory=(
            "You write clear technical explanations "
            "without unnecessary marketing language."
        ),
        llm=cometapi_llm(model_map["writer"]),
        max_iter=3,
        allow_delegation=False,
    )

    research_task = Task(
        description=(
            "Research this topic: {topic}. "
            "Return the key facts, uncertainties, "
            "and relevant sources."
        ),
        expected_output=(
            "A concise research brief containing facts, "
            "uncertainties, and source references."
        ),
        agent=researcher,
    )

    analysis_task = Task(
        description=(
            "Using the research brief, analyze {topic}. "
            "Identify the strongest conclusion and explain "
            "the major trade-offs."
        ),
        expected_output=(
            "A decision outline with evidence, assumptions, "
            "risks, and trade-offs."
        ),
        agent=analyst,
        context=[research_task],
    )

    writing_task = Task(
        description=(
            "Write a concise technical decision memo about {topic}. "
            "State the recommendation early and preserve "
            "important caveats."
        ),
        expected_output="A polished technical decision memo in Markdown.",
        agent=writer,
        context=[research_task, analysis_task],
    )

    return (
        [researcher, analyst, writer],
        [research_task, analysis_task, writing_task],
    )

The analyst receives the research output. The writer receives both the research and analysis outputs.

Build the crew with sequential execution:

from crewai import Crew, Process


def build_crew(model_map: dict[str, str]) -> Crew:
    agents, tasks = build_agents_and_tasks(model_map)

    return Crew(
        agents=agents,
        tasks=tasks,
        process=Process.sequential,
        verbose=True,
    )

If using the previous function exactly as written, define it under the name build_agents_and_tasks:

def build_agents_and_tasks(model_map: dict[str, str]):
    # Move the agent and task construction from the previous snippet here.
    ...

For a complete runnable implementation, the consolidated example later in this post avoids that split.

Fallback should only handle transient failures

This is too broad:

Any exception
  ↓
Switch model

Changing models will not repair:

400 Bad Request
401 Unauthorized
403 Forbidden
404 Not Found
422 Validation Error

Fallback is more appropriate for temporary failures:

408 Request Timeout
429 Rate Limit
500 Internal Server Error
502 Bad Gateway
503 Service Unavailable
504 Gateway Timeout
Connection error
Timeout

The fallback model must also support the same request contract: tools, structured output, parameters, and other features required by the agent.

Use the OpenAI SDK exception types:

from collections.abc import Iterator

from openai import (
    APIConnectionError,
    APIStatusError,
    APITimeoutError,
)


def exception_chain(error: BaseException) -> Iterator[BaseException]:
    current: BaseException | None = error
    seen: set[int] = set()

    while current is not None and id(current) not in seen:
        seen.add(id(current))
        yield current
        current = current.__cause__ or current.__context__


def should_fallback(error: BaseException) -> bool:
    for current in exception_chain(error):
        if isinstance(
            current,
            (APIConnectionError, APITimeoutError),
        ):
            return True

        if isinstance(current, APIStatusError):
            return (
                current.status_code in {408, 429}
                or current.status_code >= 500
            )

    return False

This excludes ordinary 4xx configuration errors while allowing connection failures, timeouts, rate limits, and 5xx responses.

Crew-level versus task-level recovery

A simple retry rebuilds the entire crew:

Start crew
  ↓ failure
Change route
  ↓
Run crew again

That can repeat completed work:

Research → completed
Analysis → completed
Writer   → failed

A full kickoff() retry may execute all three tasks again, increasing latency, token usage, cost, and possible side effects.

For expensive workflows, checkpoint instead:

Research
  ↓ checkpoint
Analysis
  ↓ checkpoint
Writer fails
  ↓
Resume writer with fallback

CrewAI provides checkpointing that persists execution state, skips completed tasks, and resumes downstream work. The design principle is:

> Checkpoint first, fallback second.

Enable it on the crew:

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    checkpoint=True,
    verbose=True,
)

A restored execution can look like:

from crewai import CheckpointConfig

result = crew.kickoff(
    from_checkpoint=CheckpointConfig(
        restore_from="./.checkpoints/checkpoint.json",
    )
)

The exact checkpoint configuration should match the CrewAI version in use.

A bounded crew-level fallback

For a small stateless workflow, a bounded crew-level retry is still useful:

def run_with_fallback(topic: str):
    routes = [
        PRIMARY_MODELS,
        {
            **PRIMARY_MODELS,
            "writer": FALLBACK_MODELS["writer"],
        },
        {
            **PRIMARY_MODELS,
            "analyst": FALLBACK_MODELS["analyst"],
            "writer": FALLBACK_MODELS["writer"],
        },
    ]

    last_error = None

    for attempt, model_map in enumerate(routes, start=1):
        try:
            crew = build_crew(model_map)
            result = crew.kickoff(inputs={"topic": topic})
            return result, model_map

        except Exception as error:
            last_error = error

            if not should_fallback(error):
                raise

            if attempt == len(routes):
                raise

            print(
                f"Transient failure on attempt {attempt}; "
                "trying bounded fallback.",
                flush=True,
            )

    raise RuntimeError(
        "Crew execution failed after all fallback routes."
    ) from last_error

This code does not claim to identify the failed agent. It changes the configured route for the next crew execution. For production workflows with tools, expensive work, or side effects, use task-level recovery and checkpoints instead.

Track usage and execution metadata

At minimum:

result, selected_models = run_with_fallback(topic)

print("Selected models:")
print(selected_models)

print("Final result:")
print(result.raw)

print("Usage:")
print(result.token_usage)

Usage fields depend on the CrewAI version and execution path, so treat the returned result object as the source of truth for the version deployed.

A durable usage record should contain:

job_id
agent
model
input_tokens
output_tokens
total_tokens
latency_ms
fallback_used
fallback_reason
status
created_at

This makes it possible to determine which agent consumes the budget, how often fallback occurs, which model is slowest, and the cost of each workflow.

max_iter=3 bounds an agent’s iteration loop. It should not be interpreted as an exact limit of three API calls or three token budgets.

Other cost controls include:

• Limiting intermediate context
• Summarizing research
• Caching repeatable work
• Restricting input and output sizes
• Limiting tool calls
• Setting user and workflow budgets
• Monitoring fallback frequency

Validate model IDs before deployment

Model IDs can become unavailable, renamed, deprecated, restricted, or incompatible with parameters your application uses.

A catalog request can be made with:

curl -s \
  https://api.cometapi.com/api/models \
  -H "Authorization: Bearer $COMETAPI_KEY"

Your CI or deployment check can verify:

gemini-3.7-flash → available
claude-opus-5    → available
gpt-5.6          → available

Catalog availability is not application compatibility. Test the tools, parameters, context sizes, and output formats your agents actually use.

Complete example

Save this as crewai_multi_model.py:

import json
import os
import sys
from collections.abc import Iterator

from crewai import Agent, Crew, LLM, Process, Task
from dotenv import load_dotenv
from openai import (
    APIConnectionError,
    APIStatusError,
    APITimeoutError,
)

load_dotenv()

COMETAPI_KEY = os.environ["COMETAPI_KEY"]
COMETAPI_BASE_URL = os.getenv(
    "COMETAPI_BASE_URL",
    "https://api.cometapi.com/v1",
)

PRIMARY_MODELS = {
    "researcher": "gemini-3.7-flash",
    "analyst": "claude-opus-5",
    "writer": "gpt-5.6",
}

FALLBACK_MODELS = {
    "researcher": "gpt-5.6",
    "analyst": "gpt-5.6",
    "writer": "gemini-3.7-flash",
}


def cometapi_llm(model_id: str) -> LLM:
    return LLM(
        model=model_id,
        base_url=COMETAPI_BASE_URL,
        api_key=COMETAPI_KEY,
        timeout=60.0,
        max_retries=0,
    )


def build_crew(model_map: dict[str, str]) -> Crew:
    researcher = Agent(
        role="Market Researcher",
        goal="Collect the facts needed to answer the topic",
        backstory=(
            "You create concise, source-aware research briefs "
            "and distinguish facts from assumptions."
        ),
        llm=cometapi_llm(model_map["researcher"]),
        max_iter=3,
        allow_delegation=False,
    )

    analyst = Agent(
        role="Product Analyst",
        goal="Turn research into a defensible recommendation",
        backstory=(
            "You evaluate evidence, assumptions, risks, "
            "and trade-offs."
        ),
        llm=cometapi_llm(model_map["analyst"]),
        max_iter=3,
        allow_delegation=False,
    )

    writer = Agent(
        role="Technical Writer",
        goal="Produce a concise technical decision memo",
        backstory=(
            "You write clear technical explanations "
            "without unnecessary hype."
        ),
        llm=cometapi_llm(model_map["writer"]),
        max_iter=3,
        allow_delegation=False,
    )

    research_task = Task(
        description=(
            "Research this topic: {topic}. "
            "Return the key facts, uncertainties, "
            "and relevant sources."
        ),
        expected_output=(
            "A concise research brief with facts "
            "and open questions."
        ),
        agent=researcher,
    )

    analysis_task = Task(
        description=(
            "Using the research brief, analyze {topic}. "
            "Identify the strongest conclusion and "
            "explain the major trade-offs."
        ),
        expected_output=(
            "A decision outline with evidence, "
            "assumptions, risks, and trade-offs."
        ),
        agent=analyst,
        context=[research_task],
    )

    writing_task = Task(
        description=(
            "Write a concise technical decision memo "
            "about {topic}. State the recommendation early "
            "and preserve important caveats."
        ),
        expected_output=(
            "A polished technical decision memo in Markdown."
        ),
        agent=writer,
        context=[
            research_task,
            analysis_task,
        ],
    )

    return Crew(
        agents=[
            researcher,
            analyst,
            writer,
        ],
        tasks=[
            research_task,
            analysis_task,
            writing_task,
        ],
        process=Process.sequential,
        verbose=True,
    )


def exception_chain(
    error: BaseException,
) -> Iterator[BaseException]:
    current: BaseException | None = error
    seen: set[int] = set()

    while current is not None and id(current) not in seen:
        seen.add(id(current))
        yield current
        current = current.__cause__ or current.__context__


def should_fallback(error: BaseException) -> bool:
    for current in exception_chain(error):
        if isinstance(
            current,
            (
                APIConnectionError,
                APITimeoutError,
            ),
        ):
            return True

        if isinstance(current, APIStatusError):
            return (
                current.status_code in {408, 429}
                or current.status_code >= 500
            )

    return False


def run_with_fallback(topic: str):
    routes = [
        PRIMARY_MODELS,
        {
            **PRIMARY_MODELS,
            "writer": FALLBACK_MODELS["writer"],
        },
        {
            **PRIMARY_MODELS,
            "analyst": FALLBACK_MODELS["analyst"],
            "writer": FALLBACK_MODELS["writer"],
        },
    ]

    last_error = None

    for attempt, model_map in enumerate(routes, start=1):
        try:
            crew = build_crew(model_map)
            result = crew.kickoff(
                inputs={"topic": topic}
            )
            return result, model_map

        except Exception as error:
            last_error = error

            if not should_fallback(error):
                raise

            if attempt == len(routes):
                raise

            print(
                f"Transient failure on attempt {attempt}; "
                "trying fallback.",
                file=sys.stderr,
            )

    raise RuntimeError(
        "No model route completed the crew."
    ) from last_error


def main():
    topic = (
        sys.argv[1]
        if len(sys.argv) > 1
        else (
            "Should a small SaaS add "
            "AI-generated meeting summaries?"
        )
    )

    result, selected_models = run_with_fallback(topic)

    output = {
        "selected_models": selected_models,
        "raw": result.raw,
        "tasks_output": [
            task.raw
            for task in result.tasks_output
        ],
        "token_usage": str(
            result.token_usage
        ),
    }

    print(
        json.dumps(
            output,
            indent=2,
            default=str,
        )
    )


if __name__ == "__main__":
    main()

Run it with:

python crewai_multi_model.py \
  "Should a small SaaS add AI-generated meeting summaries?"

A successful result will have this general shape:

{
  "selected_models": {
    "researcher": "gemini-3.7-flash",
    "analyst": "claude-opus-5",
    "writer": "gpt-5.6"
  },
  "raw": "",
  "tasks_output": [
    "",
    "",
    ""
  ],
  "token_usage": ""
}

The exact output and usage values depend on the input, model behavior, CrewAI version, and execution path. If a retryable failure occurs, selected_models shows the route used for that execution.

Production concerns

Model routing

A useful routing policy balances:

quality
reliability
context fit
tool compatibility
cost
latency

Typical policies include:

Simple task       → economical model
Complex task      → stronger model
Interactive task  → lower-latency model
Background task   → higher-quality model
Research          → Model A
Analysis          → Model B
Writing           → Model C
Code              → Model D

Side effects

Fallback is dangerous when agents modify external systems. Suppose an agent creates a database record, sends an email, calls an API, and then times out. The operation may have succeeded even though the model request failed.

Use:

• Idempotency keys
• Task checkpoints
• Transaction boundaries
• Execution IDs
• Durable task state
• Explicit side-effect confirmation

For example:

job_id  = crew_run_123
task_id = writer_456

Persist those identifiers with external operations so retries can determine whether the operation already completed.

Observability

Log at least:

workflow_id
agent
model
task
start_time
end_time
latency
status
fallback_used
fallback_reason
input_tokens
output_tokens
total_tokens

Do not log API keys, private credentials, unredacted sensitive prompts, private user data, or unrestricted model output.

Track:

Reliability:
  success rate
  timeout rate
  5xx rate
  fallback rate

Performance:
  p50 latency
  p95 latency
  p99 latency

Cost:
  input tokens
  output tokens
  cost per task
  cost per workflow

Quality:
  task success rate
  human evaluation
  structured-output validity
  tool-call success

Error boundaries

Keep these categories distinct:

configuration
  ↓
API transport
  ↓
model execution
  ↓
agent logic
  ↓
tool execution
  ↓
application side effects

A generic except Exception: use_fallback() hides too much. Authentication failures, invalid model IDs, malformed requests, tool failures, and side-effect failures need different responses.

Direct providers versus a unified endpoint

Direct provider APIs make sense when an application needs one provider’s native features and only one credential.

A unified endpoint is more useful when the CrewAI workflow needs models from several providers while the application wants one access layer, centralized authentication, model-ID routing, and consolidated usage visibility.

The responsibility split remains:

CrewAI
  → agents, tasks, context, orchestration

Model access layer
  → credentials, endpoint, model routing

Models
  → generation, reasoning, tools, output

One API key does not mean the models have identical capabilities.

Scaling the routing policy

Once model IDs are separated from agent definitions, adding an agent does not require restructuring the workflow:

PRIMARY_MODELS = {
    "researcher": "gemini-3.7-flash",
    "analyst": "claude-opus-5",
    "writer": "gpt-5.6",
    "coder": "YOUR_CODE_MODEL",
}

The same pattern supports research, analysis, coding, review, writing, and fact-checking agents.

The next step is dynamic selection:

select_model(
    task="analysis",
    budget=budget,
    latency_target=latency_target,
)

That selector can choose from an approved model allowlist using task type, budget, latency, availability, and compatibility requirements.

For production, the important components are:

1. Model allowlist
2. Per-agent routing
3. Bounded retries
4. Task checkpointing
5. Usage tracking
6. Cost controls
7. Compatibility testing
8. Observability
9. Idempotent side effects

That is considerably safer than wrapping crew.kickoff() in an unrestricted retry loop.

Final notes

The useful abstraction is simple:

> CrewAI defines what agents do; model IDs define how each agent performs its work.

A fast model can handle high-volume research, a stronger reasoning model can handle synthesis, and a general-purpose model can produce the final memo. Keeping those choices in configuration makes the workflow easier to test, observe, and change.

For a small stateless workflow, bounded crew-level fallback may be sufficient. For production, combine explicit routing with model validation, checkpoint-based recovery, usage tracking, compatibility tests, and idempotent side effects.