Que sont les flux agents à Crewai? - Analytique Vidhya

Rationalisez vos flux de travail IA avec des flux de Crewai! Ce framework puissant fournit des modèles structurés pour l'orchestration des interactions entre les agents d'IA, permettant aux développeurs d'intégrer de manière transparente des tâches de codage et des équipages pour une automatisation IA robuste. Les flux agentiques de Crewai offrent des workflows axés sur les événements, la simplification de la coordination des tâches, la gestion de l'État et le contrôle de l'exécution dans vos applications d'IA.

Table des matières

• Que sont les équipages?
• Comprendre les flux
  • Contrôle du flux de travail et séquençage
  • Gestion efficace de l'État
  • Manipulation des entrées flexibles
  • Architecture et ajustement dynamique axé sur les événements
  • Routage des tâches et exécution conditionnelle
• Flows in Practice: un exemple de recommandation de film
  • Configuration et installation
  • Gestion des avertissements
  • Variables d'environnement de chargement
  • Importation de modules nécessaires
  • Définir l'agent
  • Définition des tâches
  • Créer des équipages pour chaque genre
  • Définir les genres et le genre
  • Construire le MovieRecommendationFlow
  • Visualiser le flux
  • Initier le flux
• Conclusion
• Questions fréquemment posées

Que sont les équipages?

Les équipages de Crewai facilitent l'orchestration des agents de l'IA pour l'achèvement des tâches automatisées. Ils permettent une collaboration en douceur entre les agents pour résoudre des problèmes complexes. Mais pourquoi "coule"? Parce que les flux de Crewai fournissent des modèles structurés pour gérer ces interactions d'agent, définissant comment les agents communiquent et travaillent ensemble pour atteindre des objectifs spécifiques. Les flux sont essentiellement des séquences de tâches, où la sortie d'une tâche peut déclencher la suivante. Le système propose des mécanismes flexibles pour gérer l'exécution de l'état et conditionnel.

Comprendre les flux

Les flux fonctionnent sur un modèle axé sur les événements, réagissant à des déclencheurs et des conditions spécifiques. Cela permet des ajustements de flux de travail dynamiques en fonction des résultats de l'exécution des tâches, rationalisant les processus complexes d'IA.

Contrôle du flux de travail et séquençage

Les flux Crewai permettent aux développeurs de structurer les séquences de tâches et de contrôler le flux d'informations entre les tâches. Les tâches peuvent être enchaînées, créant un ordre d'opérations logiques. L'exécution conditionnelle des tâches basée sur les sorties de tâches antérieures est également prise en charge.

Gestion efficace de l'État

La gestion structurée de l'État, à l'aide BaseModel de Pydontic, garantit la cohérence et la structure des données entre les tâches. Cela offre une sécurité, une validation et une gestion plus faciles des états de données complexes.

Manipulation des entrées flexibles

Les flux acceptent les entrées pour initialiser ou mettre à jour leur état à tout moment pendant l'exécution. Les entrées peuvent être fournies au début, pendant ou après l'exécution, en fonction des besoins du flux de travail.

Architecture et ajustement dynamique axé sur les événements

Les flux de Crewai s'ajustent dynamiquement en fonction des résultats de la tâche. Les tâches peuvent écouter les sorties des étapes précédentes, créant un système réactif où de nouvelles tâches sont déclenchées en fonction des sorties précédentes. Les décorateurs @listen() et @router() offrent cette flexibilité, permettant une liaison de tâches conditionnelle et dynamique. Le décorateur @start() marque le point de départ du flux.

{{Table_placeholder 21}} `ou<td>Triggers a listener when any specified method emits an output.</td> <tr><td></td></tr> et_`

Décorateurs et logique conditionnelle	Description
@listen()	Crée des méthodes d'auditeur déclenchées par des événements spécifiques ou des sorties de tâches.
@router()	Permet le routage conditionnel, permettant différents chemins d'exécution en fonction des sorties de pas antérieures. Utile pour gérer les résultats de réussite / d'échec.
Déclenche un auditeur uniquement lorsque toutes les méthodes spécifiées émettent des sorties.

Routage des tâches et exécution conditionnelle

Les flux utilisent le routage pour contrôler l'exécution en fonction des conditions. Le décorateur @router() permet aux méthodes de sélectionner des chemins d'exécution en fonction des résultats de la tâche antérieurs. Par exemple, une méthode peut vérifier la sortie d'une tâche précédente et choisir un chemin en fonction de la satisfaction des conditions spécifiques.

Flows in Practice: un exemple de recommandation de film

Créons un système agentique utilisant des flux Crewai pour recommander des films basés sur le genre.

Configuration et installation

 <code>!pip install crewai -U !pip install crewai-tools</code>

Gestion des avertissements

 <code>import warnings warnings.filterwarnings('ignore')</code>

Variables de chargement de l'environnement (remplacer les espaces réservés par vos clés réelles)

 <code>import os os.environ["OPENAI_API_KEY"] = 'YOUR_OPENAI_API_KEY' os.environ['OPENAI_MODEL_NAME'] = 'gpt-4o-mini-2024-07-18' os.environ["SERPER_API_KEY"]='YOUR_SERPER_API_KEY'</code>

Importation de modules nécessaires

 <code>from crewai import Agent, Task, Crew from crewai.flow.flow import listen, start, and_, or_, router from crewai_tools import SerperDevTool from crewai import Flow from pydantic import BaseModel</code>

Définir l'agent

Un seul agent sera utilisé pour toutes les tâches. Cet agent utilise un outil de recherche Google.

 <code>movie_agent = Agent( role="Recommend popular movie specific to the genre", goal="Provide a list of movies based on user preferences", backstory="You are a cinephile, " "you recommend good movies to your friends, " "the movies should be of the same genre", tools=[SerperDevTool()], verbose=True )</code>

Définition des tâches

 <code>action_task = Task(name="ActionTask", description="Recommends a popular action movie", expected_output="A list of 10 popular movies", agent=movie_agent) comedy_task = Task(name="ComedyTask", description="Recommends a popular comedy movie", expected_output="A list of 10 popular movies", agent=movie_agent) drama_task = Task(name="DramaTask", description="Recommends a popular drama movie", expected_output="A list of 10 popular movies", agent=movie_agent) sci_fi_task = Task(name="SciFiTask", description="Recommends a sci-fi movie", expected_output="A list of 10 popular movies", agent=movie_agent)</code>

Créer des équipages pour chaque genre

 <code>action_crew = Crew(agents=[movie_agent], tasks=[action_task], verbose=True) comedy_crew = Crew(agents=[movie_agent], tasks=[comedy_task], verbose=True) drama_crew = Crew(agents=[movie_agent], tasks=[drama_task], verbose=True) sci_fi_crew = Crew(agents=[movie_agent], tasks=[sci_fi_task], verbose=True)</code>

Définir les genres et le genre

 <code>GENRES = ["action", "comedy", "drama", "sci-fi"] class GenreState(BaseModel): genre: str = ""</code>

Construire le MovieRecommendationFlow

Cette classe hérite de la classe Flow et utilise la fonctionnalité d'état.

 <code>class MovieRecommendationFlow(Flow[GenreState]): @start() def input_genre(self): genre = input("Enter a genre: ") print(f"Genre input received: {genre}") self.state.genre = genre return genre @router(input_genre) def route_to_crew(self): genre = self.state.genre if genre not in GENRES: raise ValueError(f"Invalid genre: {genre}") if genre == "action": return "action" elif genre == "comedy": return "comedy" elif genre == "drama": return "drama" elif genre == "sci-fi": return "sci-fi" @listen("action") def action_movies(self, genre): recommendations = action_crew.kickoff() return recommendations @listen("comedy") def comedy_movies(self, genre): recommendations = comedy_crew.kickoff() return recommendations @listen("drama") def drama_movies(self, genre): recommendations = drama_crew.kickoff() return recommendations @listen("sci-fi") def sci_fi_movies(self, genre): recommendations = sci_fi_crew.kickoff() return recommendations @listen(or_("action_movies", "comedy_movies", "drama_movies", "sci_fi_movies")) def finalize_recommendation(self, recommendations): print("Final movie recommendations:") return recommendations</code>

Les décorateurs @listen , @router , or_ et @start gèrent l'exécution du flux.

Visualiser le flux

 <code>flow = MovieRecommendationFlow() flow.plot() #This will generate a file, you'll need to display it separately (eg, using an image display function in your environment)</code>

Initier le flux

 <code>recommendations = await flow.kickoff_async()</code>

Conclusion

Les workflows axés sur les événements de Crewai simplifient l'orchestration des tâches AI. La nature flexible et adaptative des flux de Crewai, combinées à des fonctionnalités comme @listen() , @router() et à la gestion de l'État, en fait de puissants outils pour construire des applications d'IA efficaces et dynamiques.

Questions fréquemment posées

Q1. Comment transmettre des entrées à un flux? Utilisez flow.kickoff(inputs={"counter": 10}) .

Q2. Quelle est la différence entre @start() et @listen() ? @start() marque les points de départ de flux; @listen() Marque les méthodes déclenchées par l'achèvement de la tâche.

Q3. Comment visualiser mon flux? Utilisez flow.plot() .

Q4. Puis-je intégrer les commentaires humains? Oui, les flux de Crewai soutiennent les commentaires humains en boucle.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!