CrewAI

Intégrez CrewAI à Weave pour surveiller et tracer les applications multi-agents

CrewAI est un framework Python léger et ultra-rapide, entièrement développé from scratch, totalement indépendant de LangChain et des autres frameworks d’agents. CrewAI offre aux développeurs à la fois la simplicité d’abstractions de haut niveau (Crews) et un contrôle précis à bas niveau (Flows), ce qui en fait une solution idéale pour créer des agents d’IA autonomes adaptés à tous les cas d’usage. Pour en savoir plus, consultez la documentation de CrewAI.

Lorsque vous travaillez avec des agents d’IA, il est crucial de déboguer et de surveiller leurs interactions. Les applications CrewAI se composent souvent de plusieurs agents qui collaborent entre eux ; il est donc essentiel de comprendre comment ils coopèrent et communiquent. Weave simplifie ce processus en capturant automatiquement les traces de vos applications CrewAI, afin que vous puissiez surveiller et analyser les performances et les interactions de vos agents.

L’intégration prend en charge Crews et Flows.

​

Premiers pas avec Crew

Vous devez installer CrewAI (plus de détails) et weave pour exécuter cet exemple :

pip install crewai weave

Nous allons maintenant créer un Crew CrewAI et suivre l’exécution à l’aide de Weave. Pour commencer, appelez simplement weave.init() au début de votre script. L’argument de weave.init() est le nom du projet dans lequel les traces seront enregistrées.

import weave
from crewai import Agent, Task, Crew, LLM, Process

# Initialiser Weave avec le nom de votre projet
weave.init(project_name="crewai_demo")

# Créer un LLM avec une température de 0 pour garantir des sorties déterministes
llm = LLM(model="gpt-4o-mini", temperature=0)

# Créer des agents
researcher = Agent(
    role='Research Analyst',
    goal='Find and analyze the best investment opportunities',
    backstory='Expert in financial analysis and market research',
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role='Report Writer',
    goal='Write clear and concise investment reports',
    backstory='Experienced in creating detailed financial reports',
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

# Créer des tâches
research_task = Task(
    description='Deep research on the {topic}',
    expected_output='Comprehensive market data including key players, market size, and growth trends.',
    agent=researcher
)

writing_task = Task(
    description='Write a detailed report based on the research',
    expected_output='The report should be easy to read and understand. Use bullet points where applicable.',
    agent=writer
)

# Créer un crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True,
    process=Process.sequential,
)

# Exécuter le crew
result = crew.kickoff(inputs={"topic": "AI in material science"})
print(result)

Weave suit et consigne tous les appels effectués via la bibliothèque CrewAI, y compris les interactions entre agents, les exécutions de tâches et les appels LLM. Vous pouvez consulter les traces dans l’interface web de Weave.

CrewAI fournit plusieurs méthodes pour mieux contrôler le processus de kickoff : kickoff(), kickoff_for_each(), kickoff_async() et kickoff_for_each_async(). L’intégration prend en charge la journalisation des traces pour toutes ces méthodes.

​

Suivi des outils

Les outils CrewAI offrent aux agents des capacités allant de la recherche sur le Web et de l’analyse de données à la collaboration et à la délégation de tâches entre collègues. L’intégration peut également en assurer le traçage.

Nous allons améliorer la qualité du rapport généré dans l’exemple ci-dessus en lui donnant accès à un outil capable d’effectuer des recherches sur Internet et de renvoyer les résultats les plus pertinents.

Commençons par installer la dépendance supplémentaire.

pip install 'crewai[tools]'

Dans cet exemple, nous utilisons SerperDevTool pour permettre à notre agent ‘Research Analyst’ d’effectuer des recherches d’informations pertinentes sur Internet. Pour en savoir plus sur cet outil et sur les exigences de l’API, cliquez ici.

# .... existing imports ....
from crewai_tools import SerperDevTool

# Nous fournissons l'outil à l'agent.
researcher = Agent(
    role='Research Analyst',
    goal='Find and analyze the best investment opportunities',
    backstory='Expert in financial analysis and market research',
    llm=llm,
    verbose=True,
    allow_delegation=False,
    tools=[SerperDevTool()],
)

# .... existing code ....

Exécuter ce Crew avec un agent ayant accès à Internet produit de meilleurs résultats, plus pertinents. Nous traçons automatiquement l’utilisation des outils, comme illustré dans l’image ci-dessous.

L’intégration applique automatiquement un patch à tous les outils disponibles dans le dépôt crewAI-tools.

​

Prise en main de Flow

import weave
# Initialiser Weave avec le nom de votre projet
weave.init("crewai_demo")

from crewai.flow.flow import Flow, listen, router, start
from litellm import completion


class CustomerFeedbackFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def fetch_feedback(self):
        print("Fetching customer feedback")
        # Dans un scénario réel, ceci pourrait être remplacé par un appel API.
        # Pour cet exemple, nous simulons des retours clients.
        feedback = (
            "I had a terrible experience with the product. "
            "It broke after one use and customer service was unhelpful."
        )
        self.state["feedback"] = feedback
        return feedback

    @router(fetch_feedback)
    def analyze_feedback(self, feedback):
        # Utiliser le modèle de langage pour analyser le sentiment
        prompt = (
            f"Analyze the sentiment of this customer feedback and "
            "return only 'positive' or 'negative':\n\n"
            f"Feedback: {feedback}"
        )
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        sentiment = response["choices"][0]["message"]["content"].strip().lower()
        # Si la réponse est ambiguë, utiliser « negative » par défaut
        if sentiment not in ["positive", "negative"]:
            sentiment = "negative"
        return sentiment

    @listen("positive")
    def handle_positive_feedback(self):
        # Générer un message de remerciement pour un retour positif
        prompt = "Generate a thank you message for a customer who provided positive feedback."
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        thank_you_message = response["choices"][0]["message"]["content"].strip()
        self.state["response"] = thank_you_message
        return thank_you_message

    @listen("negative")
    def handle_negative_feedback(self):
        # Générer un message d'excuse avec une promesse d'amélioration du service pour un retour négatif
        prompt = (
            "Generate an apology message to a customer who provided negative feedback and offer assistance or a solution."
        )
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        apology_message = response["choices"][0]["message"]["content"].strip()
        self.state["response"] = apology_message
        return apology_message

# Instancier et démarrer le flux
flow = CustomerFeedbackFlow()
result = flow.kickoff()
print(result)

L’intégration applique automatiquement un patch au point d’entrée Flow.kickoff, ainsi que tous les décorateurs disponibles — @start, @listen, @router, @or_ et @and_.

​

Garde-fou Crew - Suivez vos propres ops

Les garde-fous de tâche permettent de valider et de transformer les résultats d’une tâche avant de les transmettre à la suivante. Vous pouvez utiliser une simple fonction Python pour valider l’exécution de l’agent à la volée.

En décorant cette fonction avec @weave.op, vous commencez à capturer les entrées, les sorties et la logique de l’application afin de pouvoir déboguer la manière dont les données sont validées par vos agents. Cela lance également automatiquement la gestion des versions du code pendant que vous expérimentez, afin de capturer des détails ad hoc qui n’ont pas encore été validés dans Git.

Prenons l’exemple d’un analyste de recherche et d’un rédacteur. Nous ajoutons un garde-fou pour valider la longueur du rapport généré.

# .... imports existants et initialisation de weave ....

# Décorez votre fonction guardrail avec `@weave.op()`
@weave.op(name="guardrail-validate_blog_content")
def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    # Récupérer le résultat brut sous forme de chaîne
    result = result.raw

    """Valider que le contenu du blog répond aux exigences."""
    try:
        # Vérifier le nombre de mots
        word_count = len(result.split())

        if word_count > 200:
            return (False, {
                "error": "Le contenu du blog dépasse 200 mots",
                "code": "WORD_COUNT_ERROR",
                "context": {"word_count": word_count}
            })

        # Logique de validation supplémentaire ici
        return (True, result.strip())
    except Exception as e:
        return (False, {
            "error": "Erreur inattendue lors de la validation",
            "code": "SYSTEM_ERROR"
        })


# .... agents existants et tâche de l'analyste de recherche ....

writing_task = Task(
    description='Rédiger un rapport détaillé basé sur la recherche en moins de 200 mots',
    expected_output='Le rapport doit être facile à lire et à comprendre. Utilisez des puces le cas échéant.',
    agent=writer,
    guardrail=validate_blog_content,
)

# .... code existant pour exécuter le crew ....

En décorant simplement la fonction de garde-fou avec @weave.op, nous pouvons suivre les entrées et les sorties de cette fonction, ainsi que le temps d’exécution, les informations sur les tokens si un LLM est utilisé en arrière-plan, la version du code, et bien plus encore.

​

Conclusion

N’hésitez pas à nous faire savoir ce que nous devrions améliorer dans cette intégration. Si vous rencontrez un problème, veuillez ouvrir une issue ici.

Pour en savoir plus sur la création de puissants systèmes multi-agents avec CrewAI, consultez leurs nombreux exemples et leur documentation.