Crewai的代理流动是什么？ - 分析Vidhya

用Crewai流简化您的AI工作流！这个强大的框架为协调AI代理之间的相互作用提供了结构化模式，使开发人员能够无缝整合编码任务和乘员组以进行鲁棒的AI自动化。 Crewai的代理流提供了事件驱动的工作流程，简化了您的AI应用程序中的任务协调，状态管理和执行控制。

目录

• 什么是船员？
• 了解流
  • 工作流控制和测序
  • 有效的国家管理
  • 灵活的输入处理
  • 事件驱动的体系结构和动态调整
  • 任务路由和有条件执行
• 在实践中流：电影推荐示例
  • 设置和安装
  • 处理警告
  • 加载环境变量
  • 导入必要的模块
  • 定义代理
  • 定义任务
  • 为每种类型创建船员
  • 定义流派和基因列
  • 构建MovieRecommendationFlow
  • 可视化流程
  • 启动流程
• 结论
• 常见问题

什么是船员？

Crewai的工作人员促进了AI代理的编排，以完成自动化任务完成。它们使代理商之间的平稳协作能够解决复杂的问题。但是为什么“流”呢？因为Crewai流提供了用于管理这些代理相互作用的结构化模式，从而定义了代理如何进行交流和共同努力以实现特定目标。流量本质上是任务序列，其中一个任务的输出可以触发下一个任务。该系统提供了用于管理状态和有条件执行的灵活机制。

了解流

流在事件驱动的模型上运行，对特定的触发器和条件做出反应。这允许根据任务执行结果进行动态工作流调整，并简化复杂的AI进程。

工作流控制和测序

Crewai流使开发人员能够构建任务序列并控制任务之间的信息流。可以将任务束缚在一起，创建逻辑运营顺序。还支持基于先前任务输出的任务执行。

有效的国家管理

结构化状态管理通常使用Pydantic的BaseModel ，可确保任务之间的数据一致性和结构。这提供了复杂数据状态的类型安全性，验证和更轻松的管理。

灵活的输入处理

Flow接受输入以在执行过程中的任何时候初始化或更新其状态。根据工作流程需求，可以在开始，期间或执行后提供输入。

事件驱动的体系结构和动态调整

CREWAI根据任务结果动态调整。任务可以从先前的步骤中侦听输出，从而创建一个反应性系统，该系统根据先前的输出触发新任务。 @listen()和@router()装饰器提供了这种灵活性，可以启用条件和动态任务链接。 @start()装饰器标记了流的起点。

{{table_placeholder 21}}'或<td>Triggers a listener when any specified method emits an output.</td> <tr><td></td></tr>和_`

装饰者和有条件的逻辑	描述
@listen()	创建由特定事件或任务输出触发的侦听方法。
@router()	启用有条件的路由，允许基于先前步骤输出的不同执行路径。对于管理成功/失败成果有用。
仅当所有指定方法发出输出时，才会触发侦听器。

任务路由和有条件执行

流利用路由来基于条件控制执行。 @router()装饰允许方法基于先前的任务结果选择执行路径。例如，方法可能会检查上一个任务的输出，并根据是否满足特定条件选择路径。

在实践中流：电影推荐示例

让我们使用crewai流量创建一个代理系统，以基于流派推荐电影。

设置和安装

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

处理警告

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

加载环境变量（用您的实际密钥替换占位符）

 <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>

导入必要的模块

<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>

定义代理

单个代理将用于所有任务。该代理使用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>

定义任务

<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>

为每种类型创建船员

<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>

定义流派和基因列

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

构建MovieRecommendationFlow

该类从Flow类继承并使用状态功能。

 <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>

@listen ， @router ， or_和@start装饰器管理流的执行。

可视化流程

<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>

启动流程

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

结论

Crewai的事件驱动的工作流简化了AI任务编排。 Crewai流的灵活和自适应性质，结合@listen() ， @router()和状态管理等功能，使其成为构建高效和动态AI应用程序的强大工具。

常见问题

Q1。如何将输入传递到流量？使用flow.kickoff(inputs={"counter": 10}) 。

Q2。 @start()和@listen()有什么区别？ @start()标记流动起点； @listen()标记由任务完成触发的方法。

Q3。如何可视化我的流程？使用flow.plot() 。

Q4。我可以纳入人类反馈吗？是的，crewai流支持人类的反馈。

以上是Crewai的代理流动是什么？ - 分析Vidhya的详细内容。更多信息请关注PHP中文网其他相关文章！