When you run a multi-agent workflow, not every agent needs the same model. Your researcher might need a large context window and strong reasoning, while your formatter only needs reliable JSON output. A crewai n4n.ai model per agent setup lets you route each agent to the right model through a single endpoint without rewriting your agent definitions. This guide walks through the complete configuration.
Prerequisites
You need Python 3.10+, an n4n.ai API key, and the CrewAI package installed. The examples assume you’re using the OpenAI-compatible client pattern that n4n.ai exposes.
pip install crewai openai python-dotenvCreate a .env file with your credentials:
N4N_API_KEY=sk-your-key-here
N4N_BASE_URL=https://api.n4n.ai/v1Step 1: Define model routing in a central config
Keep model assignments out of agent definitions. A simple dictionary maps agent roles to model identifiers that n4n.ai recognizes. This makes it trivial to swap models later without touching agent logic.
# config/models.py
AGENT_MODEL_MAP = {
"researcher": "anthropic/claude-3.5-sonnet",
"analyst": "openai/gpt-4o",
"writer": "anthropic/claude-3.5-haiku",
"editor": "openai/gpt-4o-mini",
"formatter": "meta-llama/llama-3.1-8b-instruct",
}
DEFAULT_MODEL = "openai/gpt-4o-mini"The keys match the role parameter you’ll pass when constructing agents. Values are the exact model slugs n4n.ai accepts — provider-prefixed, same as you’d use in the playground.
Step 2: Build a shared LLM factory
CrewAI’s Agent class accepts an llm parameter that can be any object implementing the LangChain BaseLanguageModel interface. The OpenAI client from langchain_openai works directly with n4n.ai when you point it at the custom base URL.
# config/llm_factory.py
import os
from functools import lru_cache
from langchain_openai import ChatOpenAI
from config.models import AGENT_MODEL_MAP, DEFAULT_MODEL
@lru_cache(maxsize=32)
def get_llm_for_role(role: str) -> ChatOpenAI:
"""Return a ChatOpenAI instance configured for the given agent role."""
model = AGENT_MODEL_MAP.get(role, DEFAULT_MODEL)
return ChatOpenAI(
model=model,
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
temperature=0.2,
max_tokens=4096,
timeout=60,
max_retries=2,
)The @lru_cache avoids recreating clients for agents sharing the same role. Adjust temperature and max_tokens per role if needed — just extend the factory.
Step 3: Construct agents with role-specific LLMs
Now wire the factory into your agent definitions. Each agent gets its own LLM instance bound to the model you designated.
# agents/crew.py
from crewai import Agent
from config.llm_factory import get_llm_for_role
def build_researcher() -> Agent:
return Agent(
role="researcher",
goal="Find authoritative sources on the given topic and extract key claims with citations",
backstory=(
"You are a meticulous research analyst. You prioritize primary sources, "
"flag conflicting information, and never hallucinate citations."
),
llm=get_llm_for_role("researcher"),
verbose=True,
allow_delegation=False,
)
def build_analyst() -> Agent:
return Agent(
role="analyst",
goal="Synthesize research findings into structured insights with confidence scores",
backstory=(
"You evaluate evidence quality, identify gaps, and produce "
"decision-ready summaries for stakeholders."
),
llm=get_llm_for_role("analyst"),
verbose=True,
allow_delegation=False,
)
def build_writer() -> Agent:
return Agent(
role="writer",
goal="Draft clear, engaging content from the analyst's structured output",
backstory=(
"You write for a technical audience. You avoid fluff, use active voice, "
"and match the requested tone and format exactly."
),
llm=get_llm_for_role("writer"),
verbose=True,
allow_delegation=False,
)
def build_editor() -> Agent:
return Agent(
role="editor",
goal="Polish the draft for clarity, consistency, and adherence to style guide",
backstory=(
"You enforce the project style guide. You catch passive voice, "
"ambiguous pronouns, and formatting drift."
),
llm=get_llm_for_role("editor"),
verbose=True,
allow_delegation=False,
)
def build_formatter() -> Agent:
return Agent(
role="formatter",
goal="Convert the final text into the requested output format (JSON, Markdown, HTML)",
backstory=(
"You output valid, well-structured markup only. No commentary, no preamble."
),
llm=get_llm_for_role("formatter"),
verbose=True,
allow_delegation=False,
)Each agent’s role string must match a key in AGENT_MODEL_MAP. If you add a new role, update the map — the factory falls back to DEFAULT_MODEL with a warning you can log.
Step 4: Assemble the crew and define tasks
Tasks stay model-agnostic. They describe what needs doing; the agent’s bound LLM determines how.
# tasks/content_tasks.py
from crewai import Task
from agents.crew import (
build_researcher, build_analyst, build_writer, build_editor, build_formatter
)
researcher = build_researcher()
analyst = build_analyst()
writer = build_writer()
editor = build_editor()
formatter = build_formatter()
research_task = Task(
description=(
"Research the topic: {topic}. Find 5-7 authoritative sources. "
"For each source, extract: title, URL, publication date, key claims, "
"and any data points with numbers. Flag contradictory claims across sources."
),
expected_output=(
"A JSON list of source objects with fields: title, url, date, claims[], "
"data_points[], contradictions[]."
),
agent=researcher,
)
analysis_task = Task(
description=(
"Synthesize the research into a structured brief. Include: executive summary, "
"key themes with supporting evidence, confidence scores (0-1) per theme, "
"identified gaps, and recommended angles for coverage."
),
expected_output="A Markdown document with the sections above.",
agent=analyst,
context=[research_task],
)
writing_task = Task(
description=(
"Write a 1500-word article from the analyst's brief. Target audience: "
"senior software engineers. Tone: authoritative, practical, no fluff. "
"Include code examples where relevant. Use H2/H3 headings."
),
expected_output="A complete Markdown article.",
agent=writer,
context=[analysis_task],
)
editing_task = Task(
description=(
"Edit the article for clarity, flow, and style guide compliance. "
"Check: active voice, consistent terminology, no orphaned acronyms, "
"heading hierarchy, code block language tags."
),
expected_output="The edited Markdown article.",
agent=editor,
context=[writing_task],
)
formatting_task = Task(
description=(
"Convert the edited article to the requested output format: {format}. "
"If JSON, output: {{'title': str, 'sections': [{'heading': str, 'content': str}]}}. "
"If HTML, emit semantic HTML5 with <article>, <section>, <pre><code>."
),
expected_output="Valid {format} output only.",
agent=formatter,
context=[editing_task],
)Note how context chains tasks — each agent receives the prior task’s output automatically. The {topic} and {format} placeholders are filled at kickoff.
Step 5: Run the crew with runtime overrides
The kickoff call is where you can inject per-run model overrides via n4n.ai’s routing headers. This is useful for A/B testing or fallback logic without changing code.
# main.py
import os
import json
from crewai import Crew, Process
from tasks.content_tasks import (
research_task, analysis_task, writing_task, editing_task, formatting_task
)
def run_crew(topic: str, output_format: str = "markdown", model_overrides: dict | None = None):
"""Execute the content pipeline with optional per-agent model overrides."""
# Apply runtime overrides by patching the LLM factory cache
if model_overrides:
from config.llm_factory import get_llm_for_role
get_llm_for_role.cache_clear()
# Rebuild agents with overridden models
# (In production, you'd pass overrides into the factory directly)
crew = Crew(
agents=[
research_task.agent,
analysis_task.agent,
writing_task.agent,
editing_task.agent,
formatting_task.agent,
],
tasks=[
research_task,
analysis_task,
writing_task,
editing_task,
formatting_task,
],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff(inputs={"topic": topic, "format": output_format})
return result
if __name__ == "__main__":
import sys
topic = sys.argv[1] if len(sys.argv) > 1 else "async python patterns"
output = run_crew(topic, "json")
print(output)Run it:
python main.py "crewai n4n.ai model per agent setup"Step 6: Verify model routing with response headers
n4n.ai returns provider metadata in response headers. Capture these to confirm each agent hit the intended model.
# utils/verify_routing.py
import httpx
from config.llm_factory import get_llm_for_role
def check_model_routing():
"""Send a test prompt to each role's LLM and print the resolved model."""
test_prompt = "Reply with only the word 'ok'."
for role in ["researcher", "analyst", "writer", "editor", "formatter"]:
llm = get_llm_for_role(role)
# The underlying client is accessible via llm.client
# We can inspect the request/response by wrapping or using a custom client
# Simpler: invoke and check the response metadata if exposed
response = llm.invoke(test_prompt)
print(f"{role:12} -> model: {llm.model_name}, response: {response.content.strip()}")
if __name__ == "__main__":
check_model_routing()Expected output:
researcher -> model: anthropic/claude-3.5-sonnet, response: ok
analyst -> model: openai/gpt-4o, response: ok
writer -> model: anthropic/claude-3.5-haiku, response: ok
editor -> model: openai/gpt-4o-mini, response: ok
formatter -> model: meta-llama/llama-3.1-8b-instruct, response: okIf any role shows the default model instead of your mapping, check the AGENT_MODEL_MAP key spelling against the agent’s role parameter.
Step 7: Add per-agent routing directives for advanced control
n4n.ai honors x-n4n-routing headers for fine-grained control — preferring specific providers, enabling cache hints, or setting cost ceilings. Extend the factory to attach these per role.
# config/llm_factory.py (extended)
import os
from functools import lru_cache
from langchain_openai import ChatOpenAI
from config.models import AGENT_MODEL_MAP, DEFAULT_MODEL
ROUTING_POLICIES = {
"researcher": {"prefer_provider": "anthropic", "max_cost_per_1k": 0.015},
"analyst": {"prefer_provider": "openai", "max_cost_per_1k": 0.01},
"writer": {"prefer_provider": "anthropic", "max_cost_per_1k": 0.002},
"editor": {"prefer_provider": "openai", "max_cost_per_1k": 0.001},
"formatter": {"prefer_provider": "any", "max_cost_per_1k": 0.0005},
}
class RoutingChatOpenAI(ChatOpenAI):
"""ChatOpenAI subclass that injects n4n.ai routing headers per request."""
def __init__(self, role: str, *args, **kwargs):
self.role = role
super().__init__(*args, **kwargs)
def _get_request_headers(self) -> dict:
headers = super()._get_request_headers() or {}
policy = ROUTING_POLICIES.get(self.role, {})
if policy:
headers["x-n4n-routing"] = json.dumps(policy)
return headers
@lru_cache(maxsize=32)
def get_llm_for_role(role: str) -> RoutingChatOpenAI:
model = AGENT_MODEL_MAP.get(role, DEFAULT_MODEL)
return RoutingChatOpenAI(
role=role,
model=model,
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
temperature=0.2,
max_tokens=4096,
timeout=60,
max_retries=2,
)The x-n4n-routing header accepts JSON with keys like prefer_provider ("anthropic", "openai", "any"), max_cost_per_1k (USD), require_caching (boolean), and fallback_models (array). This lets you encode operational constraints directly in the model factory.
Step 8: Handle fallbacks and degraded providers
When a provider is rate-limited or degraded, n4n.ai automatically falls back to the next available model honoring your routing policy. You can observe this in the response headers — check x-n4n-provider and x-n4n-model on each response to see what actually served the request.
For production crews, add a callback to log fallbacks:
# utils/fallback_logger.py
from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict
class FallbackLogger(BaseCallbackHandler):
def on_llm_end(self, response: Any, **kwargs: Any) -> None:
# The response object may carry provider metadata in response.llm_output
# or in generation_info depending on the integration version
if hasattr(response, 'llm_output') and response.llm_output:
provider = response.llm_output.get('provider')
model = response.llm_output.get('model')
if provider and model:
print(f"[routing] provider={provider} model={model}")
# Attach to any LLM instance:
# llm.callbacks = [FallbackLogger()]This visibility matters when debugging latency spikes or unexpected output quality — you’ll know immediately if a fallback triggered.
Step 7 routing policy forced a provider switch
Step 9: Meter usage per agent for cost attribution
Since each agent uses a different model, you’ll want per-agent token accounting. n4n.ai returns usage in the standard OpenAI usage field. Wrap the crew kickoff to aggregate by role.
# utils/usage_tracker.py
from dataclasses import dataclass, field
from collections import defaultdict
from typing import ClassVar
@dataclass
class UsageTracker:
_totals: ClassVar[defaultdict] = defaultdict(lambda: {"prompt": 0, "completion": 0, "total": 0})
@classmethod
def add(cls, role: str, usage: dict):
cls._totals[role]["prompt"] += usage.get("prompt_tokens", 0)
cls._totals[role]["completion"] += usage.get("completion_tokens", 0)
cls._totals[role]["total"] += usage.get("total_tokens", 0)
@classmethod
def report(cls) -> dict:
return dict(cls._totals)
@classmethod
def reset(cls):
cls._totals.clear()
# Monkey-patch or callback to capture usage
def track_usage_callback(response, role: str):
if hasattr(response, 'usage') and response.usage:
UsageTracker.add(role, response.usage)Attach this via a callback handler or by wrapping the invoke method. At the end of a run, UsageTracker.report() gives you a clean breakdown for chargeback or optimization decisions.
Common pitfalls
Mismatched role keys. The agent’s role parameter must exactly match a key in AGENT_MODEL_MAP. A typo like "research" instead of "researcher" silently falls back to the default model. Validate at startup:
def validate_role_mapping():
from agents.crew import build_researcher, build_analyst, build_writer, build_editor, build_formatter
from config.models import AGENT_MODEL_MAP
agents = [build_researcher(), build_analyst(), build_writer(), build_editor(), build_formatter()]
for agent in agents:
if agent.role not in AGENT_MODEL_MAP:
raise ValueError(f"Role '{agent.role}' missing from AGENT_MODEL_MAP")Shared mutable LLM instances. Without the @lru_cache or explicit instance management, two agents with the same role could share an LLM object with mutated state (callbacks, streaming config). The factory pattern prevents this.
Ignoring context window limits. The researcher on Claude 3.5 Sonnet gets 200k tokens; the formatter on Llama 3.1 8B gets 128k. If your research task output exceeds the downstream agent’s window, the crew fails mid-run. Size your task outputs to the smallest context window in the chain, or add a summarization step.
Hardcoding model names in agents. The whole point of this setup is avoiding that. If you find yourself writing llm=ChatOpenAI(model="gpt-4o") inside an agent definition, move it to the factory.
Verification checklist
Run through these after wiring everything:
- Model resolution — Run
python -m utils.verify_routingand confirm each role maps to the intended model slug. - End-to-end execution —
python main.py "test topic"completes without errors and produces valid output in the requested format. - Routing headers — Inspect a live response (add
print(response.response_metadata)in a callback) and verifyx-n4n-providermatches your policy. - Fallback behavior — Temporarily set an invalid
prefer_providerinROUTING_POLICIESand confirm the crew still completes via fallback. - Usage accounting — After a run,
UsageTracker.report()shows non-zero tokens for each role.
Next steps
From here you can extend the pattern: add a critic agent with a reasoning model for self-consistency checks, implement parallel task branches with Process.hierarchical, or wire the routing policies to a feature flag system for gradual rollouts. The factory abstraction keeps all model decisions in one place — change a slug in AGENT_MODEL_MAP and the entire crew adopts it on the next run.