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中文網其他相關文章！