khalilbenaz/claude-skills-collection/agent-crewai-expert

pip install crewai==0.100.0 crewai-tools==0.20.0
# Avec CLI officielle (recommandé pour nouveaux projets)
pip install crewai[tools]
crewai create crew mon_projet

mon_projet/
├── src/mon_projet/
│   ├── crew.py          # définition du Crew principal
│   ├── agents.py        # factory d'agents
│   ├── tasks.py         # définition des tâches
│   └── tools/
│       └── custom_tool.py
├── .env                 # OPENAI_API_KEY, SERPER_API_KEY
├── pyproject.toml
└── README.md

OPENAI_API_KEY=sk-...
OPENAI_MODEL_NAME=gpt-4o
SERPER_API_KEY=...        # si SerperDevTool

from crewai import Agent
researcher = Agent(
    role="Analyste Financier Senior",
    goal="Extraire les indicateurs clés de {company} à partir de rapports publics",
    backstory=(
        "Vous êtes analyste financier avec 12 ans d'expérience chez un cabinet de conseil Tier-1. "
        "Expert en lecture de bilans IFRS, vous détectez les signaux faibles que d'autres ignorent. "
        "Vous citez toujours vos sources avec l'URL et la date."
    ),
    llm="gpt-4o",
    tools=[search_tool, scrape_tool],
    allow_delegation=False,   # évite les délégations involontaires en sequential
    max_iter=5,               # stoppe les boucles runaway
    verbose=True,
)

from crewai.tools import tool
@tool("Calculateur de marge brute")
def calculate_margin(revenue: float, cost: float) -> str:
    """Calcule la marge brute en pourcentage à partir du CA et du coût."""
    if revenue == 0:
        return "Erreur : CA ne peut pas être zéro."
    margin = ((revenue - cost) / revenue) * 100
    return f"Marge brute : {margin:.2f}%"

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
class SQLQueryInput(BaseModel):
    query: str = Field(description="Requête SQL SELECT à exécuter")
class SQLTool(BaseTool):
    name: str = "Outil SQL"
    description: str = "Exécute une requête SQL en lecture seule sur la base de données."
    args_schema: type[BaseModel] = SQLQueryInput
    def _run(self, query: str) -> str:
        # connexion + exécution
        return results_as_str

from crewai import Task
from pydantic import BaseModel
class AnalysisOutput(BaseModel):
    company: str
    revenue_2024: float
    growth_rate: float
    key_risks: list[str]
analysis_task = Task(
    description=(
        "Analysez les données financières publiques de {company} pour l'exercice 2024. "
        "Cherchez les rapports annuels, communiqués de presse et données Boursorama. "
        "Identifiez le CA, la croissance YoY, et les 3 principaux risques."
    ),
    expected_output=(
        "Un rapport JSON structuré avec : company, revenue_2024 (en M€), "
        "growth_rate (en %), key_risks (liste de 3 phrases max chacune)."
    ),
    agent=researcher,
    output_pydantic=AnalysisOutput,   # force la structure de sortie
    output_file="analysis.json",
)

synthesis_task = Task(
    description="Rédigez un executive summary à partir des analyses fournies pour {company}.",
    expected_output="Executive summary de 300 mots max en français.",
    agent=writer,
    context=[analysis_task],    # attend la complétion de analysis_task
)

from crewai import Crew, Process
# Process séquentiel (défaut, déterministe)
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[analysis_task, writing_task, review_task],
    process=Process.sequential,
    verbose=True,
    memory=True,
    embedder={"provider": "openai", "config": {"model": "text-embedding-3-small"}},
    max_rpm=20,           # limite globale appels/minute
)
# Exécution standard
result = crew.kickoff(inputs={"company": "Société Générale", "year": "2024"})
print(result.raw)
print(f"Tokens : {crew.usage_metrics}")
# Exécution sur une liste (batch)
results = crew.kickoff_for_each(inputs=[
    {"company": "BNP Paribas"},
    {"company": "Crédit Agricole"},
])
# Exécution asynchrone
import asyncio
result = asyncio.run(crew.kickoff_async(inputs={"company": "AXA"}))

hierarchical_crew = Crew(
    agents=[analyst, researcher, writer],  # sans agent dans les Task
    tasks=[task1, task2, task3],           # le manager choisit qui fait quoi
    process=Process.hierarchical,
    manager_llm="gpt-4o",   # ou manager_agent=Agent(...)
    verbose=True,
)

from crewai import Pipeline
pipeline = Pipeline(
    stages=[
        data_collection_crew,   # output → input du suivant automatiquement
        analysis_crew,
        reporting_crew,
    ]
)
result = pipeline.kickoff(inputs={"topic": "march IA 2026"})

from crewai.agents.agent_builder.base_agent_executor_mixin import CrewAgentExecutorMixin
def on_step(step_output):
    print(f"[STEP] {step_output}")
def on_task(task_output):
    print(f"[TASK DONE] {task_output.description[:60]}")
crew = Crew(
    ...,
    step_callback=on_step,
    task_callback=on_task,
)

pip install langfuse

import os
os.environ["LANGFUSE_SECRET_KEY"] = "..."
os.environ["LANGFUSE_PUBLIC_KEY"] = "..."
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"
# CrewAI détecte automatiquement Langfuse via OpenTelemetry

import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["SERPER_API_KEY"] = "..."
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
researcher = Agent(
    role="Chercheur Expert en IA",
    goal="Trouver des informations précises et récentes sur {topic}",
    backstory=(
        "Chercheur senior spécialisé en IA avec 10 ans d'expérience. "
        "Identifie les sources fiables, cite les URLs, synthétise efficacement."
    ),
    tools=[search_tool, scrape_tool],
    llm="gpt-4o", max_iter=5, verbose=True,
)
writer = Agent(
    role="Rédacteur Technique",
    goal="Rédiger un article de blog engageant sur {topic}",
    backstory="Rédacteur tech ciblant un public de développeurs. Ton clair, exemples concrets.",
    llm="gpt-4o", max_iter=3, verbose=True,
)
reviewer = Agent(
    role="Éditeur",
    goal="Vérifier qualité, exactitude et clarté de l'article",
    backstory="Éditeur expérimenté : structure, cohérence technique, engagement.",
    llm="gpt-4o-mini", max_iter=3, verbose=True,
)
research_task = Task(
    description=(
        "Recherche approfondie sur {topic}. "
        "5 tendances, acteurs clés, chiffres, cas d'usage. Cite les URLs."
    ),
    expected_output="Rapport structuré : résumé (200 mots), 5 tendances, acteurs, chiffres, URLs.",
    agent=researcher, output_file="research.md",
)
writing_task = Task(
    description="Article de blog 800-1000 mots sur {topic}. Titre, intro, 3-4 H2, conclusion CTA.",
    expected_output="Article Markdown complet entre 800 et 1000 mots.",
    agent=writer, context=[research_task], output_file="draft.md",
)
review_task = Task(
    description="Révise l'article : exactitude, clarté, SEO. Produis la version finale.",
    expected_output="Version finale Markdown + rapport de révision.",
    agent=reviewer, context=[writing_task], output_file="final.md",
)
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,
    verbose=True, memory=True, max_rpm=20,
    embedder={"provider": "openai", "config": {"model": "text-embedding-3-small"}},
)
if __name__ == "__main__":
    result = crew.kickoff(inputs={"topic": "Agents IA en 2026"})
    print(result.raw)
    print(f"Tokens : {crew.usage_metrics}")