CrewAI多智能體框架的實操教程-旅行規劃-2

1、創建一個新的 CrewAI 項目 surprise_trip

crewai create crew surprise_trip

選擇模型廠商和模型

生成.env

MODEL=gpt-4o
OPENAI_API_KEY=你的api_keySERPER_API_KEY=你的SERPER api_key

2、探索項目結構

3、配置代理

修改?agents.yaml文件。

# 個性化活動規劃師 Agent
personalized_activity_planner:# 角色名稱：用于識別 agent 的職責（對模型來說類似標題）role: >活動規劃師# 目標描述：告訴 agent 它要完成的任務目標goal: >調研并尋找目的地的有趣活動和事件，確保這些內容符合旅行者的興趣和年齡段# 背景故事：為 agent 賦予人設，增強語言模型的代入感，提升生成質量backstory: >你擅長根據旅行者的具體偏好和人群特點設計個性化的行程安排，你能快速找到最適合他們的活動組合，打造難忘的旅程。# 餐廳探索者 Agent
restaurant_scout:role: >餐廳偵查員goal: >尋找目的地評分高的餐廳和用餐體驗，并推薦風景優美和有趣的場所backstory: >你是一個熱愛美食的人，熟悉本地最棒的用餐地點，你總能找到既美味又有氛圍的餐廳，同時也能發現風景優美和適合游玩的好地方。# 行程整合者 Agent
itinerary_compiler:role: >行程整合師goal: >整合所有調研到的信息，編排成一個完整的逐日行程安排，確保包含航班與酒店信息；請用中文輸出；backstory: >你擅長組織和呈現信息，注重細節，能夠把零散內容整合為一個條理清晰、愉快易行的旅行計劃。

4、配置任務

修改這個?tasks.yaml文件。

# 個性化活動規劃任務
personalized_activity_planning_task:# 任務描述：為旅行者尋找符合興趣和年齡段的活動與事件description: >調研并發現適合在 {destination} 進行的有趣活動。重點關注符合旅行者興趣和年齡段的活動和事件。利用互聯網搜索工具和推薦引擎獲取信息。旅行者信息如下：- 出發地: {origin}- 目的地: {destination}- 年齡: {age}- 酒店位置: {hotel_location}- 航班信息: {flight_information}- 旅行時長: {trip_duration}# 預期輸出：旅行期間每天推薦的活動清單expected_output: >一份按天列出的活動與事件推薦清單。每一項應包含活動名稱、地點、簡要描述、適合旅行者的理由，并盡可能附上評論和評分信息。# 餐廳與風景地點探索任務
restaurant_scenic_location_scout_task:# 任務描述：尋找優質餐廳與適合游玩的風景地description: >在 {destination} 尋找高評分的餐廳和獨特的用餐體驗。推薦風景優美的地點和有趣的活動，符合旅行者偏好。使用互聯網搜索工具、餐廳評論網站和旅行指南。提供多樣化選擇，滿足不同口味與預算，并提供評分信息。旅行者信息如下：- 出發地: {origin}- 目的地: {destination}- 年齡: {age}- 酒店位置: {hotel_location}- 航班信息: {flight_information}- 旅行時長: {trip_duration}# 預期輸出：每天推薦的餐廳與景點清單expected_output: >一份按天列出的推薦餐廳、風景地點和活動清單。每一項應包括名稱、地址、菜系或活動類型、簡要描述及評分信息。# 行程整合任務
itinerary_compilation_task:# 任務描述：整合所有信息為完整的行程規劃文檔description: >整合所有調研信息，編排出 {destination} 的完整日程安排。行程應包含航班、酒店信息、計劃的每日活動與用餐安排。使用文本格式化與文檔生成工具進行內容組織與美化。# 預期輸出：結構化的日程表expected_output: >一份詳細的旅行行程文檔。內容應包含逐日安排，并整合航班、酒店、活動、餐廳和風景地點等信息。格式為 Markdown，不帶“```”

5、配置團隊

修改這個?crew.py文件

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task# 如果你有自定義工具，可以取消以下注釋導入自定義工具
# from surprise_travel.tools.custom_tool import MyCustomTool# 以下是 CrewAI 提供的內置工具，用于網頁搜索與抓取
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from pydantic import BaseModel, Field
from typing import List, Optional# 活動模型類：表示一天中安排的某一項活動
class Activity(BaseModel):name: str = Field(..., description="活動名稱")location: str = Field(..., description="活動地點")description: str = Field(..., description="活動描述")date: str = Field(..., description="活動日期")cousine: str = Field(..., description="餐廳的菜系類型")why_its_suitable: str = Field(..., description="為什么這項活動適合旅行者")reviews: Optional[List[str]] = Field(..., description="活動的評論列表")rating: Optional[float] = Field(..., description="活動評分")# 每日旅行計劃模型類
class DayPlan(BaseModel):date: str = Field(..., description="當天的日期")activities: List[Activity] = Field(..., description="當天的活動列表")restaurants: List[str] = Field(..., description="當天推薦的餐廳列表")flight: Optional[str] = Field(None, description="當天的航班信息（如有）")# 整體行程模型類
class Itinerary(BaseModel):name: str = Field(..., description="行程名稱，可以有趣一點")day_plans: List[DayPlan] = Field(..., description="每天的行程計劃列表")hotel: str = Field(..., description="酒店信息")# 使用 CrewBase 裝飾器標記為 Crew 項目的入口
@CrewBase
class SurpriseTravelCrew():"""SurpriseTravel 旅行策劃團隊（crew）"""# 代理配置文件路徑（YAML 格式）agents_config = 'config/agents.yaml'# 任務配置文件路徑（YAML 格式）tasks_config = 'config/tasks.yaml'# 個性化活動規劃 agent@agentdef personalized_activity_planner(self) -> Agent:return Agent(config=self.agents_config['personalized_activity_planner'],  # 從配置文件加載 agent 設置tools=[SerperDevTool(), ScrapeWebsiteTool()],  # 使用搜索和網頁抓取工具verbose=True,  # 輸出詳細執行日志allow_delegation=False,  # 禁止任務委托給其他代理)# 餐廳與景點推薦 agent@agentdef restaurant_scout(self) -> Agent:return Agent(config=self.agents_config['restaurant_scout'],tools=[SerperDevTool(), ScrapeWebsiteTool()],verbose=True,allow_delegation=False,)# 整合所有信息并輸出最終行程的 agent@agentdef itinerary_compiler(self) -> Agent:return Agent(config=self.agents_config['itinerary_compiler'],tools=[SerperDevTool()],verbose=True,allow_delegation=False,)# 個性化活動規劃任務@taskdef personalized_activity_planning_task(self) -> Task:return Task(config=self.tasks_config['personalized_activity_planning_task'],agent=self.personalized_activity_planner()  # 指定任務負責的 agent)# 餐廳與風景位置探索任務@taskdef restaurant_scenic_location_scout_task(self) -> Task:return Task(config=self.tasks_config['restaurant_scenic_location_scout_task'],agent=self.restaurant_scout())# 編譯最終行程任務，并指定輸出格式為 Itinerary 數據模型@taskdef itinerary_compilation_task(self) -> Task:return Task(config=self.tasks_config['itinerary_compilation_task'],agent=self.itinerary_compiler(),output_json=Itinerary  # 指定結構化輸出格式)@taskdef reporting_task(self) -> Task:return Task(config=self.tasks_config['itinerary_compilation_task'],  # type: ignore[index]agent=self.itinerary_compiler(),output_file='report-01.md')# 將 agents 與 tasks 組裝成一個完整的 crew 執行流@crewdef crew(self) -> Crew:"""創建 SurpriseTravel 團隊，定義其執行流程"""return Crew(agents=self.agents,  # 所有注冊的代理（通過 @agent 自動生成）tasks=self.tasks,  # 所有注冊的任務（通過 @task 自動生成）process=Process.sequential,  # 順序執行所有任務verbose=True,# 如需使用分層流程，也可使用以下配置：# process=Process.hierarchical, 參考：https://docs.crewai.com/how-to/Hierarchical/)

6、配置主函數

修改這個?main.py文件

# 指定使用 Python 解釋器運行該腳本import sys
import warnings  # 導入警告處理模塊
from datetime import datetime  # 導入處理日期和時間的模塊from surprise_trip.crew import SurpriseTravelCrew  # 從項目中導入 SurpriseTravelCrew 類# 關閉特定模塊中的語法警告（此處為 pysbd 模塊）
warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")# 此主程序文件是為了在本地運行 crew 而設計的，請勿在此文件中添加不必要的邏輯。
# 你可以替換下面的 inputs 變量，它會自動填充任務和代理的信息。def run():"""運行 crew 執行流程。"""# 設置輸入參數，可以根據需要修改。它會自動被各個任務和 agent 使用。inputs = {'origin': '上海虹橋機場（SHA）',  # 出發地'destination': '北京大興機場（PKX）',  # 目的地'age': 31,  # 用戶年齡'hotel_location': '北京朝陽區',  # 希望住的酒店位置'flight_information': '國航 CA1234，起飛時間為 2025年6月30日 上午10:00',  # 航班信息'trip_duration': '14 天'  # 旅行時長}try:# 實例化 SurpriseTravelCrew 并運行 crew 的 kickoff 方法（啟動執行流程）SurpriseTravelCrew().crew().kickoff(inputs=inputs)except Exception as e:# 捕獲并拋出異常，便于調試raise Exception(f"運行 crew 時發生錯誤: {e}")def train():"""訓練 crew，執行指定次數的迭代。"""inputs = {'origin': '上海虹橋機場（SHA）',  # 出發地'destination': '北京大興機場（PKX）',  # 目的地'age': 31,  # 用戶年齡'hotel_location': '北京朝陽區',  # 希望住的酒店位置'flight_information': '國航 CA1234，起飛時間為 2025年6月30日 上午10:00',  # 航班信息'trip_duration': '14 天'  # 旅行時長}try:# 通過命令行參數獲取迭代次數和保存的模型文件名# 使用 crew 的 train 方法進行訓練SurpriseTravelCrew().crew().train(n_iterations=int(sys.argv[1]), filename=sys.argv[2], inputs=inputs)except Exception as e:# 捕獲并拋出異常，便于調試raise Exception(f"訓練 crew 時發生錯誤: {e}")def replay():"""從指定的任務 ID 開始回放 crew 執行過程。"""try:# 從命令行參數中獲取任務 ID，并調用 replay 方法回放SurpriseTravelCrew().crew().replay(task_id=sys.argv[1])except Exception as e:# 捕獲并拋出異常，便于調試raise Exception(f"回放 crew 時發生錯誤: {e}")def test():"""測試 crew 的執行邏輯，并返回評估結果。"""# 設置測試輸入inputs = {"topic": "AI LLMs",  # 測試主題，例如大語言模型"current_year": str(datetime.now().year)  # 當前年份}try:# 從命令行參數中獲取迭代次數和用于評估的 LLM 名稱，調用 test 方法進行測試SurpriseTravelCrew().crew().test(n_iterations=int(sys.argv[1]), eval_llm=sys.argv[2], inputs=inputs)except Exception as e:# 捕獲并拋出異常，便于調試raise Exception(f"測試 crew 時發生錯誤: {e}")