Production CrewAI

Ein CrewAI-Prototyp im Notebook ist der erste Schritt. Aber für Production brauchen Sie Error Handling, Retries, Cost Control, Testing und professionelles Deployment. Diese Lektion zeigt, wie Sie CrewAI produktionsreif machen.

Error Handling

Crew-Level Error Handling

from crewai import Crew

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    max_rpm=10,           # Rate Limiting: max 10 Requests/Minute
    max_tokens=50000,     # Token-Limit pro Crew-Run
    verbose=True
)

try:
    result = crew.kickoff(inputs={"topic": "AI Trends 2026"})
except Exception as e:
    logger.error(f"Crew-Fehler: {e}")
    fallback_result = simple_chain.invoke(inputs)

Agent-Level Retries

resilient_agent = Agent(
    role="Resilient Researcher",
    goal="Recherche auch bei temporären Fehlern abschließen",
    backstory="...",
    max_retry_limit=3,       # Maximal 3 Retries
    max_iter=15,             # Maximal 15 Iterationen
    respect_context_window=True  # Automatisches Context-Trimming
)

Cost Control

LLM-Kosten können mit CrewAI schnell steigen, da mehrere Agents parallel arbeiten:

Kosten-Strategien

Model Mixing Beispiel

# Teures Modell nur für komplexe Aufgaben
analyst = Agent(
    role="Senior Analyst",
    llm=ChatOpenAI(model="gpt-4o"),  # Komplex → teures Modell
    # ...
)

# Günstigeres Modell für einfachere Aufgaben
formatter = Agent(
    role="Report Formatter",
    llm=ChatOpenAI(model="gpt-4o-mini"),  # Einfach → günstiges Modell
    # ...
)

Testing-Strategien

Unit Tests für Tools

def test_search_tool():
    result = web_search.run("test query")
    assert isinstance(result, str)
    assert len(result) > 0

def test_database_tool():
    result = db_query.run("SELECT COUNT(*) FROM users")
    assert "count" in result.lower()

Integration Tests für Crews

def test_research_crew():
    crew = Crew(
        agents=[test_researcher],
        tasks=[test_task],
        process=Process.sequential
    )
    result = crew.kickoff(inputs={"topic": "Test Topic"})
    assert result is not None
    assert len(result.raw) > 100
    assert result.token_usage.total_tokens < 10000

Deployment mit Docker

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

FastAPI-Wrapper

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

class CrewRequest(BaseModel):
    topic: str
    output_format: str = "markdown"

@app.post("/run-crew")
async def run_crew(request: CrewRequest, background_tasks: BackgroundTasks):
    job_id = str(uuid4())
    background_tasks.add_task(execute_crew, job_id, request)
    return {"job_id": job_id, "status": "started"}

Monitoring

• Token-Usage: Pro Agent, pro Task, pro Crew-Run tracken
• Latenz: Dauer jeder Task messen und Bottlenecks identifizieren
• Erfolgsrate: Wie oft liefert die Crew das erwartete Ergebnis?
• Kosten: Tägliche/monatliche Kosten pro Crew monitoren

Praxis-Tipp: Setzen Sie von Anfang an ein Token-Budget pro Crew-Run. Ohne Budget kann eine Crew mit Delegation-Loops hunderte Dollar pro Run kosten. Kombinieren Sie max_tokens, max_iter und max_rpm für umfassende Kostenkontrolle.