CrewAI: Multi-Agent Koordinasyon Framework'ü Production Rehberi 2026

1from crewai import Agent, Task, Crew, Process
2from crewai_tools import SerperDevTool, WebsiteSearchTool
3 
4# Temel yapı - 3 agent, 3 task, 1 crew
5researcher = Agent(
6    role="Senior Research Analyst",
7    goal="Kapsamlı ve doğrulanmış bilgi toplama",
8    backstory="""10 yıllık araştırma deneyimine sahip,
9    kaynak doğrulama konusunda takıntılı bir analist.""",
10    tools=[SerperDevTool(), WebsiteSearchTool()],
11    verbose=True,
12    allow_delegation=False,
13    max_iter=5,
14    llm="gpt-4o"
15)
16 
17writer = Agent(
18    role="Technical Content Writer",
19    goal="Teknik bilgiyi okunabilir içeriğe dönüştürme",
20    backstory="Karmaşık konuları basitleştiren deneyimli yazar.",
21    tools=[],
22    verbose=True,
23    allow_delegation=False,
24    llm="gpt-4o-mini"  # Maliyet optimizasyonu: yazı için mini model
25)
26 
27reviewer = Agent(
28    role="Quality Assurance Editor",
29    goal="İçeriğin doğruluğunu ve kalitesini garanti etme",
30    backstory="Hata toleransı sıfır olan titiz editör.",
31    tools=[],
32    verbose=True,
33    allow_delegation=True,  # Düzeltme için writer'a delege edebilir
34    llm="gpt-4o"
35)

1agent = Agent(
2    role="AI Assistant",
3    goal="Help with tasks",
4    backstory="An AI that helps with things."
5)

1senior_ios_architect = Agent(
2    role="Senior iOS Software Architect",
3    goal="""SwiftUI ve UIKit projelerinde kod kalitesini artırmak,
4    technical debt'i azaltmak ve team'in architectural decision'larını
5    rehberlik etmek.""",
6    backstory="""8 yıllık iOS deneyimine sahip, 50+ App Store uygulaması
7    ship etmiş bir mühendis. WWDC speaker'ı olarak Apple best practices
8    konusunda otorite. Memory management, concurrency ve SwiftUI
9    performance optimizasyonu özel uzmanlık alanları.
10    Kod review'larında acımasız ama yapıcı.
11    "It works" yetmez, "it works correctly and efficiently" ister.""",
12    tools=[CodeInterpreterTool(), FileReadTool()],
13    allow_delegation=False,
14    max_iter=8
15)

1# YANLIŞ: allow_delegation=True olan agent'ların sonsuz döngüye girme riski
2agent_a = Agent(role="Manager", allow_delegation=True)
3agent_b = Agent(role="Worker", allow_delegation=True)  # İkisi birbirine delege edebilir!
4 
5# DOĞRU: Hiyerarşik delegasyon tasarımı
6manager = Agent(role="Project Manager", allow_delegation=True)
7worker_a = Agent(role="Frontend Dev", allow_delegation=False)
8worker_b = Agent(role="Backend Dev", allow_delegation=False)

1from crewai import Task
2from pydantic import BaseModel
3 
4# Structured output için Pydantic model
5class ResearchReport(BaseModel):
6    topic: str
7    key_findings: list[str]
8    sources: list[str]
9    confidence_score: float
10 
11research_task = Task(
12    description="""
13    {topic} hakkında kapsamlı bir araştırma yap.
14 
15    Araştırman şunları kapsamalı:
16    1. Güncel gelişmeler (son 6 ay)
17    2. Key players ve market share
18    3. Technical advantages ve limitations
19    4. Production use cases
20 
21    En az 5 güvenilir kaynak kullan.
22    """,
23    expected_output="Yapılandırılmış araştırma raporu, key findings ve kaynaklar",
24    output_pydantic=ResearchReport,  # Structured output
25    agent=researcher,
26    async_execution=False,
27    context=[]  # Bu task bağımsız, önceki task çıktısına bağlı değil
28)
29 
30write_task = Task(
31    description="""
32    Araştırma raporunu teknik bir blog yazısına dönüştür.
33 
34    Blog özellikleri:
35    - 1500-2000 kelime
36    - Başlangıç seviyesi okuyucuya hitap et
37    - Kod örnekleri ekle (varsa)
38    - SEO-friendly başlıklar kullan
39    """,
40    expected_output="Markdown formatında, publication-ready blog yazısı",
41    agent=writer,
42    context=[research_task]  # Araştırma task'ının çıktısını input olarak alır
43)
44 
45review_task = Task(
46    description="""
47    Blog yazısını teknik doğruluk ve kalite için incele.
48 
49    Kontrol listesi:
50    - Teknik claim'ler araştırma verisiyle uyuşuyor mu?
51    - Grammar ve stil tutarlı mı?
52    - Kod örnekleri çalışır mı?
53 
54    Sorun varsa writer'a düzeltme için delege et.
55    """,
56    expected_output="Final, onaylanmış blog yazısı veya düzeltme talepleri",
57    agent=reviewer,
58    context=[research_task, write_task]  # Her iki task'ın context'i var
59)

1# Task output'unu sınırlamak için output sınırı
2analysis_task = Task(
3    description="Veriyi analiz et, en önemli 5 bulguyu listele",
4    expected_output="Maksimum 500 kelime, sadece bullet points",  # Açıkça boyut sınırla
5    agent=analyst
6)

1sequential_crew = Crew(
2    agents=[researcher, writer, reviewer],
3    tasks=[research_task, write_task, review_task],
4    process=Process.sequential,
5    verbose=True,
6    memory=True
7)
8 
9# Çalıştırma
10result = sequential_crew.kickoff(inputs={"topic": "CrewAI Production Patterns"})
11print(result.raw)

1from langchain_openai import ChatOpenAI
2 
3# Manager için güçlü model zorunlu
4manager_llm = ChatOpenAI(model="gpt-4o", temperature=0)
5 
6hierarchical_crew = Crew(
7    agents=[researcher, writer, reviewer],
8    tasks=[research_task, write_task, review_task],
9    process=Process.hierarchical,
10    manager_llm=manager_llm,
11    verbose=True,
12    memory=True,
13    planning=True  # Pre-execution planning aktif
14)

1from crewai.tools import BaseTool
2from pydantic import BaseModel, Field
3import requests
4from typing import Type
5 
6# Input schema tanımı
7class DatabaseQueryInput(BaseModel):
8    query: str = Field(description="SQL query to execute")
9    database: str = Field(description="Database name (analytics/production/staging)")
10    limit: int = Field(default=100, description="Max rows to return")
11 
12# Tool implementasyonu
13class DatabaseQueryTool(BaseTool):
14    name: str = "Database Query Tool"
15    description: str = """
16    Internal database'e SQL query çalıştırır.
17    SELECT query'leri destekler, DML operasyonları (INSERT/UPDATE/DELETE) yasak.
18    Sonuçları JSON formatında döner.
19    """
20    args_schema: Type[BaseModel] = DatabaseQueryInput
21 
22    def _run(self, query: str, database: str, limit: int = 100) -> str:
23        # Güvenlik kontrolü
24        forbidden_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER"]
25        if any(keyword in query.upper() for keyword in forbidden_keywords):
26            return "HATA: Sadece SELECT query'leri destekleniyor."
27 
28        try:
29            # Gerçek DB bağlantısı burada
30            # conn = get_db_connection(database)
31            # results = conn.execute(f"{query} LIMIT {limit}")
32 
33            # Demo için mock data
34            return f"Query çalıştırıldı: {query[:50]}... - 42 satır döndü"
35        except Exception as e:
36            return f"Database hatası: {str(e)}"
37 
38# Decorator tabanlı alternatif (daha basit)
39from crewai.tools import tool
40 
41@tool("Web Scraper Tool")
42def web_scraper(url: str) -> str:
43    """
44    Verilen URL'den içerik çeker ve metin formatında döner.
45    JavaScript-heavy siteler için Selenium gerekebilir.
46    """
47    try:
48        response = requests.get(url, timeout=10, headers={
49            "User-Agent": "Mozilla/5.0 (compatible; CrewAI Bot)"
50        })
51        response.raise_for_status()
52        # Basit HTML stripping
53        from html.parser import HTMLParser
54        class TextExtractor(HTMLParser):
55            def __init__(self):
56                super().__init__()
57                self.text = []
58            def handle_data(self, data):
59                self.text.append(data)
60 
61        parser = TextExtractor()
62        parser.feed(response.text)
63        return " ".join(parser.text)[:3000]  # Token limiti için
64    except Exception as e:
65        return f"Scraping hatası: {str(e)}"
66 
67# Agent'a tool atama
68researcher_with_tools = Agent(
69    role="Data Research Analyst",
70    goal="Veri topla ve analiz et",
71    backstory="Database ve web kaynaklarına erişimi olan analist.",
72    tools=[DatabaseQueryTool(), web_scraper],
73    max_iter=5
74)

1class StrictAPICallInput(BaseModel):
2    endpoint: str = Field(
3        description="API endpoint'i - sadece /users, /orders, /products",
4        pattern=r"^/(users|orders|products)$"  # Regex validation
5    )
6    method: str = Field(
7        description="HTTP metodu",
8        pattern=r"^(GET|POST)$"
9    )
10 
11class StrictAPITool(BaseTool):
12    name: str = "API Call Tool"
13    description: str = "Internal API'ye call yapar. Sadece /users, /orders, /products endpoint'leri desteklenir."
14    args_schema: Type[BaseModel] = StrictAPICallInput
15 
16    def _run(self, endpoint: str, method: str) -> str:
17        # Pydantic zaten validate etti, ek kontrol gerek yok
18        return f"{method} {endpoint} - 200 OK"

1from crewai import Crew, Process
2from crewai.memory import LongTermMemory, ShortTermMemory, EntityMemory
3from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage
4 
5# Memory-enabled Crew
6memory_crew = Crew(
7    agents=[researcher, writer],
8    tasks=[research_task, write_task],
9    process=Process.sequential,
10 
11    # Short-term: Mevcut crew run içinde context
12    memory=True,
13 
14    # Long-term: Çalışmalar arası persistent storage
15    long_term_memory=LongTermMemory(
16        storage=LTMSQLiteStorage(db_path="./crew_memory.db")
17    ),
18 
19    # Entity memory: Kişi, şirket, ürün gibi entity'leri takip eder
20    entity_memory=EntityMemory(),
21 
22    verbose=True
23)
24 
25# Custom embedding için memory yapılandırma
26import os
27os.environ["OPENAI_API_KEY"] = "your-key"
28 
29# ChromaDB ile long-term memory
30from crewai.memory.storage.chroma_db_storage import ChromaDBStorage
31 
32chroma_crew = Crew(
33    agents=[researcher],
34    tasks=[research_task],
35    memory=True,
36    long_term_memory=LongTermMemory(
37        storage=ChromaDBStorage(
38            collection_name="crew_research_memory",
39            persist_directory="./chroma_db"
40        )
41    )
42)

1import asyncio
2from crewai import Task, Crew, Process
3 
4# Paralel task'lar - birbirine bağımlı değil
5market_research_task = Task(
6    description="Rakip analizi yap",
7    expected_output="Rakip listesi ve analizi",
8    agent=researcher,
9    async_execution=True  # Paralel çalışır
10)
11 
12technical_research_task = Task(
13    description="Teknik stack analizi yap",
14    expected_output="Teknik değerlendirme",
15    agent=researcher,
16    async_execution=True  # Paralel çalışır
17)
18 
19# Bu task ikisinin de bitmesini bekler
20synthesis_task = Task(
21    description="Market ve teknik araştırmayı birleştir",
22    expected_output="Kapsamlı rapor",
23    agent=writer,
24    context=[market_research_task, technical_research_task],  # Her ikisini de bekler
25    async_execution=False
26)
27 
28# Async crew kickoff
29async def run_crew():
30    crew = Crew(
31        agents=[researcher, writer],
32        tasks=[market_research_task, technical_research_task, synthesis_task],
33        process=Process.sequential,
34        verbose=True
35    )
36    result = await crew.kickoff_async(inputs={"topic": "Vector Databases 2026"})
37    return result
38 
39# Çalıştırma
40result = asyncio.run(run_crew())

1# Aynı crew ile birden fazla input'u paralel işle
2async def batch_process():
3    crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
4 
5    inputs_list = [
6        {"topic": "CrewAI Production Patterns"},
7        {"topic": "LangGraph vs CrewAI"},
8        {"topic": "Vector Databases 2026"},
9    ]
10 
11    # Hepsini paralel çalıştır
12    results = await crew.kickoff_for_each_async(inputs=inputs_list)
13    return results

1from crewai import Crew
2import json
3 
4# Detaylı callback sistemi
5def task_callback(task_output):
6    """Her task tamamlandığında çağrılır"""
7    print(f"[TASK COMPLETE] Agent: {task_output.agent}")
8    print(f"[TASK COMPLETE] Output (ilk 200 char): {str(task_output.raw)[:200]}")
9 
10    # Loglama
11    log_entry = {
12        "timestamp": "2026-04-18T10:00:00Z",
13        "agent": task_output.agent,
14        "task_output_length": len(str(task_output.raw)),
15    }
16    print(json.dumps(log_entry))
17 
18def step_callback(step_output):
19    """Her agent adımında çağrılır - tool calls dahil"""
20    print(f"[STEP] {step_output}")
21 
22monitored_crew = Crew(
23    agents=[researcher, writer],
24    tasks=[research_task, write_task],
25    process=Process.sequential,
26    verbose=True,
27    task_callback=task_callback,
28    step_callback=step_callback,
29 
30    # Token kullanım takibi
31    full_output=True  # Tüm intermediate output'ları yakala
32)
33 
34result = monitored_crew.kickoff()
35 
36# Token kullanım özeti
37if hasattr(result, "token_usage"):
38    print(f"Toplam token: {result.token_usage}")

1from langfuse.callback import CallbackHandler
2 
3langfuse_handler = CallbackHandler(
4    public_key="lf_pk_...",
5    secret_key="lf_sk_...",
6    host="https://cloud.langfuse.com"
7)
8 
9# CrewAI agent'larına Langfuse callback ekle
10researcher_monitored = Agent(
11    role="Senior Research Analyst",
12    goal="Kapsamlı araştırma yap",
13    backstory="Deneyimli analist",
14    callbacks=[langfuse_handler],
15    llm="gpt-4o"
16)

1from langchain_openai import ChatOpenAI
2from langchain_anthropic import ChatAnthropic
3 
4# Maliyet optimizasyonu: doğru göreve doğru model
5cheap_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)  # ~$0.15/1M token
6standard_llm = ChatOpenAI(model="gpt-4o", temperature=0)    # ~$2.50/1M token
7premium_llm = ChatAnthropic(model="claude-opus-4-7")         # En güçlü, en pahalı
8 
9# Göreve göre model atama
10data_collector = Agent(
11    role="Data Collector",
12    goal="Ham veri topla",
13    backstory="Veri toplama uzmanı",
14    llm=cheap_llm,  # Basit görev, ucuz model
15    max_iter=3
16)
17 
18senior_analyst = Agent(
19    role="Senior Data Analyst",
20    goal="Karmaşık pattern analizi",
21    backstory="İleri düzey analitik uzmanı",
22    llm=premium_llm,  # Karmaşık görev, güçlü model
23    max_iter=8
24)
25 
26# Rate limiting koruması
27from langchain_core.rate_limiters import InMemoryRateLimiter
28 
29rate_limiter = InMemoryRateLimiter(
30    requests_per_second=0.5,  # Dakikada 30 istek
31    check_every_n_seconds=0.1,
32    max_bucket_size=10,
33)
34 
35rate_limited_llm = ChatOpenAI(
36    model="gpt-4o",
37    rate_limiter=rate_limiter
38)

1# crew_config.py
2from pydantic_settings import BaseSettings
3 
4class CrewSettings(BaseSettings):
5    openai_api_key: str
6    anthropic_api_key: str
7    serper_api_key: str
8 
9    # Güvenlik limitleri
10    max_iterations_per_agent: int = 10
11    max_execution_time_seconds: int = 300
12    max_rpm: int = 30
13 
14    # Memory
15    memory_enabled: bool = True
16    memory_db_path: str = "./data/crew_memory.db"
17 
18    class Config:
19        env_file = ".env"
20 
21settings = CrewSettings()
22 
23production_crew = Crew(
24    agents=[researcher, writer],
25    tasks=[research_task, write_task],
26    process=Process.sequential,
27    max_rpm=settings.max_rpm,
28    memory=settings.memory_enabled,
29    verbose=False  # Production'da verbose=False
30)

1import signal
2from contextlib import contextmanager
3 
4class TimeoutError(Exception):
5    pass
6 
7@contextmanager
8def timeout(seconds):
9    def handler(signum, frame):
10        raise TimeoutError(f"Crew execution {seconds}s limitini aştı!")
11 
12    signal.signal(signal.SIGALRM, handler)
13    signal.alarm(seconds)
14    try:
15        yield
16    finally:
17        signal.alarm(0)
18 
19# Güvenli crew çalıştırma
20def safe_crew_kickoff(crew, inputs, timeout_seconds=300):
21    try:
22        with timeout(timeout_seconds):
23            result = crew.kickoff(inputs=inputs)
24            return {"success": True, "result": result.raw}
25    except TimeoutError as e:
26        return {"success": False, "error": str(e)}
27    except Exception as e:
28        return {"success": False, "error": f"Beklenmeyen hata: {str(e)}"}

1# LangGraph yaklaşımı - daha fazla control, daha fazla boilerplate
2from langgraph.graph import StateGraph, END
3from typing import TypedDict
4 
5class WorkflowState(TypedDict):
6    topic: str
7    research: str
8    article: str
9    approved: bool
10 
11def research_node(state: WorkflowState) -> WorkflowState:
12    # Research logic
13    return {**state, "research": f"Research on {state['topic']}"}
14 
15def write_node(state: WorkflowState) -> WorkflowState:
16    # Write logic
17    return {**state, "article": f"Article based on {state['research']}"}
18 
19def should_revise(state: WorkflowState) -> str:
20    return "end" if state["approved"] else "write"
21 
22graph = StateGraph(WorkflowState)
23graph.add_node("research", research_node)
24graph.add_node("write", write_node)
25graph.add_conditional_edges("write", should_revise, {"end": END, "write": "write"})
26graph.set_entry_point("research")
27workflow = graph.compile()
28 
29# CrewAI yaklaşımı - aynı workflow, çok daha az kod
30crew = Crew(
31    agents=[researcher, writer, reviewer],
32    tasks=[research_task, write_task, review_task],
33    process=Process.sequential
34)