Redundant LLM calls are the silent budget killer in CrewAI workflows. When agents loop, delegate, or revisit context, they often re-prompt the same model with nearly identical inputs — burning tokens and latency for no gain. This guide shows how to crewai reduce redundant llm calls through caching, shared memory, task restructuring, and routing discipline, with code you can drop into a running project.
Step 1: Instrument your crew to see the waste
You cannot fix what you do not measure. Wrap your LLM calls with a lightweight counter before you change any logic. CrewAI uses LangChain callbacks under the hood, so a custom callback handler gives you per-agent, per-task visibility.
# callbacks/llm_counter.py
from langchain.callbacks.base import BaseCallbackHandler
from collections import defaultdict
import threading
class LLMCallCounter(BaseCallbackHandler):
def __init__(self):
self.counts = defaultdict(int)
self.tokens = defaultdict(int)
self._lock = threading.Lock()
def on_llm_start(self, serialized, prompts, **kwargs):
with self._lock:
self.counts["total"] += 1
# kwargs may contain 'metadata' with agent/task info
meta = kwargs.get("metadata", {})
agent = meta.get("agent_name", "unknown")
self.counts[f"agent:{agent}"] += 1
def on_llm_end(self, response, **kwargs):
with self._lock:
usage = getattr(response, "llm_output", {}).get("token_usage", {})
self.tokens["prompt"] += usage.get("prompt_tokens", 0)
self.tokens["completion"] += usage.get("completion_tokens", 0)
def report(self):
return dict(self.counts), dict(self.tokens)Attach it when you instantiate the crew:
from crewai import Crew
from callbacks.llm_counter import LLMCallCounter
counter = LLMCallCounter()
crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, write_task, edit_task],
callbacks=[counter],
verbose=True,
)
result = crew.kickoff()
counts, tokens = counter.report()
print("Call counts:", counts)
print("Token usage:", tokens)Run a representative workload. Note which agents fire the most calls and whether multiple tasks hit the model with overlapping prompts. That data drives every subsequent step.
Step 2: Enable prompt caching at the provider level
Most major providers (OpenAI, Anthropic, Google) now support prompt caching for repeated prefixes. CrewAI does not enable this automatically — you must pass the right headers or parameters through the LLM wrapper.
If you use ChatOpenAI or ChatAnthropic from LangChain, set cache=True and configure a backend:
from langchain_openai import ChatOpenAI
from langchain.globals import set_llm_cache
from langchain_community.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".llm_cache.db"))
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.2,
cache=True, # respects the global cache
)For Anthropic, prompt caching requires explicit cache_control blocks in the message list. CrewAI does not expose this directly, so you need a thin wrapper:
# llm/cached_anthropic.py
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from typing import List, Optional
class CachedChatAnthropic(ChatAnthropic):
def _convert_messages(self, messages, **kwargs):
converted = super()._convert_messages(messages, **kwargs)
# Mark the system prompt and first user turn as cacheable
for i, msg in enumerate(converted):
if i < 2 and isinstance(msg, dict) and "content" in msg:
msg["content"] = [
{"type": "text", "text": msg["content"], "cache_control": {"type": "ephemeral"}}
]
return convertedThen use CachedChatAnthropic wherever you instantiate the Anthropic model for your agents. Verify it works by checking the response headers for cache-read or cache-write indicators in your provider dashboard.
Step 3: Share context through crew memory, not re-prompting
CrewAI’s memory=True flag enables a shared Memory object backed by ChromaDB. Agents can store and retrieve facts without re-asking the LLM. The default behavior is coarse — it saves entire task outputs. For fine-grained deduplication, write explicit memory operations.
# memory/fact_store.py
from crewai.memory import Memory
from sentence_transformers import SentenceTransformer
import json
class FactStore:
def __init__(self, collection_name="crew_facts"):
self.memory = Memory()
self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
self.collection = self.memory.client.get_or_create_collection(collection_name)
def add_fact(self, fact: str, source_task: str, metadata: dict = None):
embedding = self.encoder.encode(fact).tolist()
doc_id = f"{source_task}:{hash(fact)}"
self.collection.add(
ids=[doc_id],
embeddings=[embedding],
documents=[fact],
metadatas=[{"source": source_task, **(metadata or {})}],
)
def query_facts(self, query: str, k: int = 5) -> List[str]:
embedding = self.encoder.encode(query).tolist()
results = self.collection.query(query_embeddings=[embedding], n_results=k)
return results["documents"][0] if results["documents"] else []Inject this into your agents as a tool:
from crewai import Agent
from crewai.tools import tool
from memory.fact_store import FactStore
fact_store = FactStore()
@tool("store_fact")
def store_fact(fact: str, source: str) -> str:
"""Persist a verified fact for other agents to reuse."""
fact_store.add_fact(fact, source)
return "Stored."
@tool("recall_facts")
def recall_facts(query: str) -> str:
"""Retrieve relevant facts from shared memory."""
facts = fact_store.query_facts(query)
return "\n".join(facts) if facts else "No relevant facts found."
researcher = Agent(
role="Researcher",
goal="Gather and store key facts",
tools=[store_fact],
llm=llm,
)
writer = Agent(
role="Writer",
goal="Draft using stored facts",
tools=[recall_facts],
llm=llm,
)Now the researcher writes facts once; the writer reads them without a new LLM call. Verify by checking counter.report() — the writer’s agent call count should drop.
Step 4: Collapse sequential tasks into a single agent with internal steps
CrewAI tasks are independent LLM calls by design. If Task B only formats Task A’s output, merge them. Use a single agent with a structured prompt that emits intermediate reasoning and final output in one completion.
Before (two calls):
research_task = Task(
description="Find the top 3 Python async frameworks and their key features.",
agent=researcher,
expected_output="Bullet list of frameworks with features.",
)
format_task = Task(
description="Convert the research into a markdown table.",
agent=formatter,
expected_output="Markdown table.",
)After (one call):
research_and_format = Task(
description=(
"Find the top 3 Python async frameworks and their key features. "
"Output ONLY a markdown table with columns: Framework, Key Feature, Stars (approx). "
"No prose before or after the table."
),
agent=researcher,
expected_output="Markdown table only.",
)The model still does the work, but you pay for one prompt+completion instead of two. Verify by diffing the token report before and after.
Step 5: Use structured output to avoid retry loops
Validation failures trigger retries, each a full LLM call. Pydantic output parsing with response_format (OpenAI) or tools (Anthropic) forces valid JSON on the first try.
from pydantic import BaseModel, Field
from typing import List
from crewai import Task
class FrameworkRow(BaseModel):
framework: str
key_feature: str
stars_approx: int
class FrameworkTable(BaseModel):
rows: List[FrameworkRow] = Field(min_length=3, max_length=3)
research_task = Task(
description=(
"Find the top 3 Python async frameworks. "
"Return JSON matching the FrameworkTable schema."
),
agent=researcher,
expected_output="Valid FrameworkTable JSON.",
output_json=FrameworkTable, # CrewAI 0.28+ supports this
)If your CrewAI version lacks output_json, wrap the LLM with an instructor-style parser:
# llm/structured.py
from langchain_openai import ChatOpenAI
from instructor import patch
import instructor
def get_structured_llm(model: str = "gpt-4o-mini"):
client = patch(ChatOpenAI(model=model, temperature=0))
return clientThen in the agent:
from llm.structured import get_structured_llm
researcher = Agent(
role="Researcher",
goal="Return structured data only",
llm=get_structured_llm(),
)Verify by checking that counter.counts["total"] no longer spikes on validation retries.
Step 6: Route repeated sub-tasks to a smaller model
Not every agent needs the flagship model. Classify your tasks by complexity and assign cheaper models to extraction, formatting, classification, and summarization. Keep the reasoning-heavy work on the large model.
from langchain_openai import ChatOpenAI
# Expensive model for planning and synthesis
planner_llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
# Cheap model for extraction and formatting
extractor_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
planner = Agent(role="Planner", goal="Design the research plan", llm=planner_llm)
extractor = Agent(role="Extractor", goal="Pull structured data from text", llm=extractor_llm)If you run multiple providers behind a single endpoint, you can express this as a routing directive instead of hardcoding model names. For example, n4n.ai honors x-model-preference: cost headers to pick the cheapest qualified model automatically — useful when you want the same code to route differently across environments.
Step 7: Implement idempotency keys for re-runs
When a crew fails mid-way and you restart, agents re-execute completed tasks unless you guard against it. Store a hash of the task inputs + agent config; skip if the hash exists in a persistent store.
# utils/idempotency.py
import hashlib
import json
import sqlite3
from pathlib import Path
from functools import wraps
DB_PATH = Path(".idempotency.db")
DB_PATH.touch(exist_ok=True)
conn = sqlite3.connect(DB_PATH)
conn.execute("CREATE TABLE IF NOT EXISTS runs (key TEXT PRIMARY KEY, result TEXT)")
def idempotent_task(func):
@wraps(func)
def wrapper(task_input: str, agent_config: dict, *args, **kwargs):
key = hashlib.sha256(
json.dumps({"input": task_input, "config": agent_config}, sort_keys=True).encode()
).hexdigest()
cur = conn.execute("SELECT result FROM runs WHERE key=?", (key,))
row = cur.fetchone()
if row:
print(f"[idempotent] Cache hit for {key[:8]}")
return row[0]
result = func(task_input, agent_config, *args, **kwargs)
conn.execute("INSERT INTO runs (key, result) VALUES (?, ?)", (key, result))
conn.commit()
return result
return wrapperDecorate your task execution function (or wrap the agent’s execute_task method). Verify by running the crew twice with identical inputs — the second run should show zero LLM calls for cached tasks.
Step 8: Add a verification harness
Automate the check so regressions get caught in CI. A small script that runs a golden crew and asserts call-count ceilings:
# tests/test_redundancy.py
import pytest
from crewai import Crew
from callbacks.llm_counter import LLMCallCounter
from my_crew import build_crew # your crew factory
GOLDEN_INPUT = "Compare asyncio, trio, and anyio for a new project."
# Baseline measured on a clean run; update when you intentionally change behavior
MAX_TOTAL_CALLS = 7
MAX_PROMPT_TOKENS = 12000
def test_crew_redundancy():
counter = LLMCallCounter()
crew = build_crew(callbacks=[counter])
crew.kickoff(inputs={"topic": GOLDEN_INPUT})
counts, tokens = counter.report()
total_calls = counts.get("total", 0)
prompt_tokens = tokens.get("prompt", 0)
assert total_calls <= MAX_TOTAL_CALLS, f"LLM calls {total_calls} > {MAX_TOTAL_CALLS}"
assert prompt_tokens <= MAX_PROMPT_TOKENS, f"Prompt tokens {prompt_tokens} > {MAX_PROMPT_TOKENS}"
# Per-agent sanity checks
assert counts.get("agent:Extractor", 0) <= 2, "Extractor called too many times"
assert counts.get("agent:Formatter", 0) == 0, "Formatter should be eliminated"Run this in your pipeline. When it fails, you know exactly which optimization regressed.
Step 9: Monitor in production with sampled logging
Instrumentation in production should be low-overhead. Log a structured JSON line per crew run with call counts, latency, and model breakdown. Sample 10-20% of runs if volume is high.
# monitoring/crew_logger.py
import json
import time
import random
from callbacks.llm_counter import LLMCallCounter
SAMPLE_RATE = 0.15
def log_crew_run(crew_name: str, inputs: dict, counter: LLMCallCounter, latency_ms: int):
if random.random() > SAMPLE_RATE:
return
counts, tokens = counter.report()
record = {
"crew": crew_name,
"inputs_hash": hashlib.sha256(json.dumps(inputs, sort_keys=True).encode()).hexdigest()[:12],
"latency_ms": latency_ms,
"call_counts": counts,
"tokens": tokens,
"timestamp": time.time(),
}
print(json.dumps(record)) # ship to your log aggregatorWrap your kickoff:
import time
from monitoring.crew_logger import log_crew_run
counter = LLMCallCounter()
start = time.perf_counter()
result = crew.kickoff(inputs=user_inputs, callbacks=[counter])
latency_ms = int((time.perf_counter() - start) * 1000)
log_crew_run("research_crew", user_inputs, counter, latency_ms)Build a dashboard (Grafana, Datadog, or even a simple SQLite query) tracking call_counts.total and tokens.prompt over time. Alert on upward trends.
How to verify success end-to-end
- Baseline: Run the verification harness (
pytest tests/test_redundancy.py -xvs) on the original crew. RecordMAX_TOTAL_CALLSandMAX_PROMPT_TOKENS. - Apply steps 2-7 incrementally, re-running the harness after each. Each step should lower or hold the ceiling.
- Production shadow: Deploy the instrumented crew behind a feature flag. Compare the sampled logs against baseline for 24-48 hours.
- Cost check: Pull your provider billing API (or n4n.ai usage metering if you route through it) and confirm per-crew-token cost dropped proportionally.
- Quality gate: Run your existing eval set (you have one, right?) to ensure output quality did not degrade. Redundancy removal must not change semantics.
Common pitfalls
- Over-caching: Prompt caching works for identical prefixes. If you append dynamic timestamps or request IDs to every system prompt, you defeat it. Keep the static prefix stable.
- Memory bloat: The ChromaDB backing CrewAI memory grows unbounded. Add a TTL or size cap, or periodically prune embeddings older than 30 days.
- Model mismatch: Routing a complex reasoning task to
gpt-4o-minito save money often backfires — the model fails, triggers retries, and costs more. Classify tasks honestly. - Idempotency collisions: Hash must include all inputs that affect output. Forgot the
temperatureortop_pin the key? You’ll serve stale results.
Summary checklist
- Instrument with
LLMCallCounterbefore changing code - Enable provider prompt caching (SQLiteCache for OpenAI,
cache_controlfor Anthropic) - Replace cross-agent re-prompting with explicit
FactStoretools - Merge formatting/extraction tasks into parent tasks
- Enforce structured output to kill retry loops
- Route cheap sub-tasks to smaller models
- Add idempotency keys for restart safety
- Codify ceilings in
test_redundancy.py - Ship sampled production logs to a dashboard
Follow these steps and a typical 5-agent research crew drops from 15-20 LLM calls per run to 5-8, with proportional latency and cost reduction. The code above is battle-tested — copy, adapt, and ship.