¿Qué son los flujos agentes en CrewAI?

¿Quiere simplificar la creación y trámite de flujos de trabajo de IA? Los flujos de CrewAI ofrecen patrones estructurados para orquestar las interacciones de los agentes de IA. Permiten a los desarrolladores combinar eficazmente tareas de codificación y equipos, ofreciendo un entorno potente para desarrollar la automatización de la IA. Con Agentic Flows en CrewAI, puede diseñar flujos de trabajo estructurados basados ​​en eventos que agilizan la coordinación de tareas, administran el estado y controlan la ejecución internamente de sus aplicaciones de IA.

¿Qué son las tripulaciones?

Los equipos de crewAI permiten la orquestación de agentes de IA para la automatización de tareas. Facilita la colaboración fluida entre agentes para resolver problemas complejos. Entonces, ¿por qué «fluye», preguntas? Los flujos de CrewAI son patrones estructurados para orquestar las interacciones de los agentes de IA. Definen cómo los agentes colaboran y se comunican para realizar tareas. Estos flujos constan de una secuencia de tareas donde el resultado de una tarea puede desencadenar la posterior, y el sistema proporciona mecanismos flexibles para la trámite del estado y el control de ejecución condicional.

¿Qué son los flujos?

Los flujos están impulsados ​​por eventos, lo que significa que reaccionan a condiciones o desencadenantes específicos. Puede diseñar flujos de trabajo que se ajusten dinámicamente en función de los resultados de la ejecución de diferentes tareas, lo que permite una ejecución perfecta de procesos complejos de IA.

Control del flujo de trabajo

En CrewAI Flows, los desarrolladores pueden distribuir el flujo de tareas y controlar cómo pasa la información entre ellas. Las tareas se pueden encadenar, creando un flujo inductivo de operaciones. Esta estructura permite a los desarrolladores explicar una secuencia de operaciones donde ciertas tareas se ejecutan condicionalmente en función del resultado de tareas anteriores.

Trámite del Estado

Podemos usar Structured State Management, que utiliza un esquema predefinido, generalmente a través del BaseModel de Pydantic, para certificar que los datos pasados ​​entre tareas sigan una estructura específica. Proporciona los beneficios de seguridad de tipos, subsistencia y trámite más sencilla de estados de datos complejos.

Flexibilidad de entrada

Los flujos pueden aceptar entradas para inicializar o poner al día su estado en cualquier punto de la ejecución. Dependiendo de los requisitos del flujo de trabajo, estas entradas se pueden proporcionar al principio, durante o posteriormente de la ejecución.

Obra basada en eventos

CrewAI Flows permite que los flujos de trabajo se ajusten dinámicamente en función de los resultados de diferentes tareas. Las tareas pueden escuchar los resultados de pasos anteriores, lo que permite un sistema reactivo donde se pueden activar nuevas tareas en función de los resultados de las anteriores. Los decoradores @listen() y @router() permiten este nivel de flexibilidad, permitiendo a los desarrolladores vincular tareas de forma condicional y dinámica en función de sus resultados. Recuerde que el interiorista @start() se utiliza para marcar el punto de inicio de un Flujo.

Decoradores y método condicional	Descripción
@escuchar()	Crea métodos de audición activados por eventos específicos o resultados de tareas.
@enrutador()	Habilita el enrutamiento condicional internamente del flujo, lo que permite diferentes rutas de ejecución basadas en los resultados de los pasos anteriores. Útil para diligenciar resultados de éxito o fracaso.
o_	Activa un asistente cuando cualquiera de los métodos especificados emite una salida. Ideal para continuar posteriormente de completar cualquier tarea.
y_	Garantiza que un asistente se active solo cuando todos los métodos especificados emitan resultados. Útil para esperar hasta que se completen múltiples condiciones o tareas ayer de continuar.

Enrutamiento de tareas

Los flujos asimismo permiten el uso de enrutamiento para controlar cómo procede la ejecución en función de condiciones específicas. El @enrutador() Decorator facilita esto al permitir que los métodos elijan su ruta de ejecución en función de los resultados de tareas anteriores. Por ejemplo, un método podría comprobar el resultado de una tarea inicial y lanzarse si continúa por una ruta u otra, dependiendo de si se cumplen ciertas condiciones.

Lea asimismo: Creación de agentes de IA colaborativos con CrewAI

Flujos en acto

Construyamos un sistema agente utilizando flujos de CrewAI que recomienda películas según el tipo. Este será un sistema agente simple que nos ayudará a comprender qué funciona detrás de él.

Instalaciones

!pip install crewai -U
!pip install crewai-tools

Control de advertencia

import warnings
warnings.filterwarnings('ignore')

Cargar variables de entorno

Ir a Serpiente y obtenga su esencia API de Serper para la búsqueda en Google. Usaremos el maniquí 4o-mini de OpenAI.

import os
os.environ("OPENAI_API_KEY") = ‘’
os.environ('OPENAI_MODEL_NAME') = 'gpt-4o-mini-2024-07-18'
os.environ("SERPER_API_KEY")=''

Importar módulos necesarios

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

explicar agente

Para simplificar las cosas, definiremos un único agente que utilizarán todas las tareas. Este agente tendrá una aparejo de búsqueda de Google que todas las tareas podrán utilizar.

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
)

Delimitar tareas

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
)

Delimitar equipos para cada tipo.

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
)

Delimitar géneros y GenreState

GENRES = ("action", "comedy", "drama", "sci-fi")
class GenreState(BaseModel):
    genre: str = ""

Delimitar flujo de recomendación de película

Definimos una clase que hereda de la clase Flow y opcionalmente podemos usar la funcionalidad de estado para sumar o modificar atributos de estado, aquí ya hemos definido GenreState con un atributo de tipo de tipo condena. (Observe que nuestro maniquí pydantic “GenreState” se pasa entre corchetes)

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

Deberíamos especificar el flujo de ejecución por @listen interiorista pasando el método inicial y la salida del método inicial se puede advenir como argumento en el posterior método. (Nota: el flujo debe comenzar con el interiorista @start)

El enrutador se utiliza aquí para lanzarse qué método debe ejecutarse en función de la salida del método internamente del interiorista @router.

or_ especifica que el método debe ejecutarse tan pronto como cualquiera de los métodos del método or_( ) devuelva una salida.

(Nota: use and_( ) si desea que el método espere hasta que se reciban todas las expectativas).

Trazar el flujo

flow = MovieRecommendationFlow()
flow.plot()

Mostrar el diagrama de flujo

from IPython.core.display import display, HTML
with open('/content/crewai_flow.html', 'r') as f:
    html_content = f.read()
display(HTML(html_content))

Este diagrama debería darle una mejor comprensión de lo que está sucediendo. Según el «tipo de entrada», solo se ejecuta una de las cuatro tareas, y cuando una de ellas devuelve un resultado, se activa la «Recomendación de finalización».

Inicia el flujo

recommendations = await flow.kickoff_async()

Ingresé ciencia ficción como entrada cuando me preguntaron.

Posteriormente de toda la ejecución de Flow, este es el resultado que obtuve para las recomendaciones de películas de ciencia ficción que quería.

Lea asimismo: Sistema multiagente CrewAI para escribir artículos a partir de vídeos de YouTube

Conclusión

En este artículo, hemos explorado cómo TripulaciónAI Los flujos de trabajo basados ​​en eventos simplifican la creación y trámite de la orquestación de tareas de IA. Al disfrutar los flujos estructurados, los desarrolladores pueden diseñar sistemas de IA dinámicos y eficientes que permitan una coordinación perfecta entre tareas y agentes. Con potentes herramientas como @listen(), @router() y mecanismos de trámite de estado, CrewAI permite flujos de trabajo flexibles y adaptables. El ejemplo práctico de un sistema de recomendación de películas destaca cómo Agentic Flows en CrewAI se puede utilizar de forma efectiva para personalizar aplicaciones de IA, con tareas que reaccionan a las entradas del sucesor y se ajustan dinámicamente a las condiciones.

Por otra parte, si está buscando un curso de agente de IA en vírgula, explore: Software pionero de IA agente.

Pregunta frecuente