CrewAI

Crew minimal : Researcher -> Writer, processus séquentiel

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

search_tool = SerperDevTool()

researcher = Agent(
    role="Senior Research Analyst",
    goal="Find and summarise the latest developments on {topic}",
    backstory="Expert at distilling complex information into clear insights.",
    tools=[search_tool],
    verbose=True,
)

writer = Agent(
    role="Content Writer",
    goal="Write a concise, accurate report based on the research findings",
    backstory="Skilled at turning research into readable prose.",
    verbose=True,
)

research_task = Task(
    description="Research the current state of {topic}. Identify key trends.",
    expected_output="A bullet-point summary of 5-10 key findings with sources.",
    agent=researcher,
)

write_task = Task(
    description="Write a 3-paragraph report based on the research findings.",
    expected_output="A polished 3-paragraph report, ready to publish.",
    agent=writer,
    context=[research_task],  # output of research_task feeds this task
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "CrewAI multi-agent frameworks"})

context=[research_task] est le branchement clé : le Writer reçoit l'output du Researcher comme contexte. Sans cela, les agents tournent en isolation et le Writer n'a aucune recherche sur laquelle s'appuyer.

Crew hiérarchique avec manager LLM et mémoire

from crewai import Agent, Task, Crew, Process
import os

manager_llm = "claude-sonnet-4-5"  # or any LLM identifier

analyst = Agent(
    role="Data Analyst",
    goal="Extract structured insights from raw data",
    backstory="Precise, detail-oriented data specialist.",
)

validator = Agent(
    role="Quality Validator",
    goal="Verify accuracy and completeness of analysis outputs",
    backstory="Critical thinker who catches errors before they ship.",
)

writer = Agent(
    role="Report Writer",
    goal="Produce a clear, stakeholder-ready report",
    backstory="Expert at translating technical findings into business language.",
)

analyse_task = Task(
    description="Analyse the dataset at {data_path} and extract key metrics.",
    expected_output="Structured JSON with metrics: totals, trends, anomalies.",
)

validate_task = Task(
    description="Check the analysis output for accuracy and completeness.",
    expected_output="Validation report: pass/fail per metric with notes.",
)

report_task = Task(
    description="Write an executive summary based on validated analysis.",
    expected_output="2-page executive summary with charts described in text.",
)

crew = Crew(
    agents=[analyst, validator, writer],
    tasks=[analyse_task, validate_task, report_task],
    process=Process.hierarchical,
    manager_llm=manager_llm,
    memory=True,  # LanceDB-backed memory across calls
    verbose=True,
)

result = crew.kickoff(inputs={"data_path": "/data/q2-report.csv"})

En mode hiérarchique, le manager LLM distribue les tâches et valide les outputs avant de continuer. memory=True active la mémoire adossée à LanceDB : les agents rappellent le contexte des appels précédents. Surveiller le coût de l'appel LLM supplémentaire par délégation de tâche.

Flow : aiguillage événementiel entre Crews avec état typé

from crewai.flow.flow import Flow, listen, start, router
from crewai import Crew, Process
from pydantic import BaseModel

class ReportState(BaseModel):
    topic: str = ""
    research_summary: str = ""
    needs_deep_dive: bool = False
    final_report: str = ""

class ReportFlow(Flow[ReportState]):

    @start()
    def kick_off_research(self):
        research_crew = build_research_crew()
        result = research_crew.kickoff(inputs={"topic": self.state.topic})
        self.state.research_summary = str(result)

    @router(kick_off_research)
    def route_by_depth(self):
        # Send to deep dive if summary flagged gaps
        if "insufficient data" in self.state.research_summary.lower():
            return "deep_dive"
        return "write"

    @listen("deep_dive")
    def run_deep_dive(self):
        deep_crew = build_deep_dive_crew()
        result = deep_crew.kickoff(
            inputs={"summary": self.state.research_summary}
        )
        self.state.research_summary += " " + str(result)

    @listen("write")
    @listen("run_deep_dive")
    def write_report(self):
        write_crew = build_write_crew()
        result = write_crew.kickoff(
            inputs={"research": self.state.research_summary}
        )
        self.state.final_report = str(result)

flow = ReportFlow()
flow.kickoff(inputs={"topic": "AI agent orchestration trends 2026"})

@router retourne un label qui détermine quelle branche @listen s'exécute ensuite, permettant un branchement conditionnel sans réécrire les Crews. Passer aux Flows quand la logique de coordination (branchement, transmission d'état) dépasse ce qu'un seul processus Crew peut exprimer.