打造AI智能旅行規劃器：基于LLM和Crew AI的Agent實踐

引言

今天來學習大佬開發的一個AI驅動的旅行規劃應用程序，它能夠自動處理旅行規劃的復雜性——尋jni找航班、預訂酒店以及優化行程。傳統上，這個過程需要手動搜索多個平臺，常常導致決策效率低下。

通過利用**代理型人工智能（Agentic AI）**的力量，這個應用程序展示了AI代理是如何協作以簡化規劃過程的——檢索實時旅行數據、分析選項，并使用大型語言模型（LLM）生成AI驅動的建議。

什么是代理型人工智能（Agentic AI）？

代理型人工智能（Agentic AI）指的是能夠主動運行、獨立做決策，并且無需持續人工干預即可執行復雜任務的自主人工智能系統。

關鍵特性：

自主決策 — 通過允許AI自主思考、推理并采取行動，減少人工勞動。
多代理協作 — 讓不同的AI代理專注于特定任務，提高準確性和效率。
可擴展性與效率 — AI代理并行執行任務，與傳統工作流程相比減少了處理時間。
增強用戶體驗 — 為復雜任務提供更快、更智能、更個性化的解決方案。

與需要逐步指令的傳統AI模型不同，代理型人工智能（Agentic AI）能夠動態工作，實時做決策，與其他代理協作，并根據上下文數據優化工作流程。

這個代理型人工智能（Agentic AI）系統的關鍵特性

以下是使這個AI旅行規劃器強大的最重要的特性的分解：
在這里插入圖片描述

1. 航班搜索自動化

通過SerpAPI從Google Flights檢索實時航班數據
根據價格、轉機次數和旅行時間篩選航班。
AI根據性價比和便利性推薦最佳航班。

2. 酒店推薦

從Google Hotels搜索實時酒店可用性
根據位置、預算、設施和用戶評分進行篩選。
AI通過分析與關鍵地點的接近程度等因素，建議最佳酒店。

3. AI驅動的分析與建議

使用Gemini LLM的AI代理評估旅行選項。
使用Crew AI協調多個AI代理以做出更好的決策。
AI解釋其對航班和酒店的推薦，提供見解。

4. 動態行程生成

AI根據航班和酒店預訂構建結構化的旅行計劃。
生成按天劃分的行程，包含必游景點、餐廳推薦以及當地交通選項。

5. 易于使用的API集成

API端點允許用戶搜索航班、酒店并請求AI驅動的建議。
啟用與前端UI（Streamlit）的無縫集成。

實現指南

讓我們一步步了解構建這個AI驅動的旅行規劃系統的步驟：

先決條件

在開始編碼之前，請確保在本地計算機上滿足以下先決條件：

安裝Python 3.8+。
擁有用于獲取實時航班和酒店數據的 SerpAPI 密鑰。
擁有用于AI建議的 Google Gemini API 密鑰。
使用 CrewAI 來協調AI驅動的代理。

安裝所需的庫：

pip install fastapi uvicorn pydantic serpapi crewai streamlit

步驟1：設置環境并加載API密鑰

首先，讓我們設置環境并配置必要的API密鑰：

import os
import uvicorn
import asyncio
import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from serpapi import GoogleSearch
from crewai import Agent, Task, Crew, Process, LLM
from datetime import datetime
from functools import lru_cache# 加載API密鑰
GEMINI_API_KEY = os.getenv("GOOGLE_API_KEY", "gemini_api_key_here")
SERP_API_KEY = os.getenv("SERPER_API_KEY", "serpapi_key_here")# 初始化日志記錄器
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

步驟2：初始化Google Gemini AI（LLM）

我們將使用CrewAI的LLM，為簡單起見，實現LRU緩存以僅初始化一次：

@lru_cache(maxsize=1)
def initialize_llm():"""初始化并緩存LLM實例，避免重復初始化。"""return LLM(model="gemini/gemini-2.0-flash",provider="google",api_key=GEMINI_API_KEY)

步驟3：定義Pydantic模型

我們將使用Pydantic模型來強制執行請求和響應的嚴格數據驗證：

class FlightRequest(BaseModel):origin: strdestination: stroutbound_date: strreturn_date: strclass HotelRequest(BaseModel):location: strcheck_in_date: strcheck_out_date: strclass ItineraryRequest(BaseModel):destination: strcheck_in_date: strcheck_out_date: strflights: strhotels: strclass FlightInfo(BaseModel):airline: strprice: strduration: strstops: strdeparture: strarrival: strtravel_class: strreturn_date: strairline_logo: strclass HotelInfo(BaseModel):name: strprice: strrating: floatlocation: strlink: strclass AIResponse(BaseModel):flights: List[FlightInfo] = []hotels: List[HotelInfo] = []ai_flight_recommendation: str = ""ai_hotel_recommendation: str = ""itinerary: str = ""

步驟4：Fast API初始化用于后端處理

現在，我們將創建一個REST API來處理旅行規劃請求：

from fastapi import FastAPI, HTTPExceptionapp = FastAPI(title="Travel Planning API", version="1.0.1")

步驟5：使用SerpAPI實時檢索航班和酒店數據

讓我們實現輔助函數，使用SerpAPI檢索航班和酒店數據：

None

async def run_search(params):"""通用函數，異步運行SerpAPI搜索。"""try:return await asyncio.to_thread(lambda: GoogleSearch(params).get_dict())except Exception as e:logger.exception(f"SerpAPI搜索錯誤：{str(e)}")raise HTTPException(status_code=500, detail=f"搜索API錯誤：{str(e)}")async def search_flights(flight_request: FlightRequest):"""使用SerpAPI從Google Flights獲取實時航班詳情。"""logger.info(f"搜索航班：從{flight_request.origin}到{flight_request.destination}")params = {"api_key": SERP_API_KEY,"engine": "google_flights","hl": "en","gl": "us","departure_id": flight_request.origin.strip().upper(),"arrival_id": flight_request.destination.strip().upper(),"outbound_date": flight_request.outbound_date,"return_date": flight_request.return_date,"currency": "USD"}search_results = await run_search(params)flights = search_results.get("flights")return flights
async def search_hotels(hotel_request: HotelRequest):"""從SerpAPI獲取酒店信息。"""logger.info(f"搜索酒店：{hotel_request.location}")params = {"api_key": SERP_API_KEY,"engine": "google_hotels","q": hotel_request.location,"hl": "en","gl": "us","check_in_date": hotel_request.check_in_date,"check_out_date": hotel_request.check_out_date,"currency": "USD","sort_by": 3,"rating": 8}search_results = await run_search(params)hotels = search_results.get("properties")return hotels

步驟6：定義多代理任務、團隊和流程

現在讓我們使用CrewAI設置我們的多代理系統：

代理1：航班分析：

航班分析師代理接收航班選項
Gemini LLM分析價格、時長、轉機次數和便利性
AI推薦最佳航班并提供詳細理由

代理2：酒店分析：

酒店分析師代理接收住宿選項
Gemini LLM比較價格、評分、位置和設施
AI推薦最佳酒店并提供詳細理由

代理3：行程生成：

行程代理使用航班和酒店推薦
創建按天劃分的計劃，包含景點、餐廳和交通安排
根據地理位置優化日程安排
包含時間估算和實用旅行提示

None

async def get_ai_recommendation(data_type, formatted_data):logger.info(f"從AI獲取{data_type}分析")llm_model = initialize_llm()# 根據數據類型配置代理if data_type == "flights":role = "AI航班分析師"goal = "分析航班選項并推薦最佳選擇，考慮價格、時長、轉機次數和整體便利性。"backstory = f"AI專家，提供基于多個因素的航班選項深入分析。"description = """根據以下詳細信息，推薦最佳航班：**推薦理由**：- **價格**：詳細說明為什么此航班與其他航班相比最具性價比。- **時長**：解釋為什么此航班與其他航班相比具有最佳時長。- **轉機次數**：討論為什么此航班的轉機次數最少或最合理。- **旅行艙位**：描述為什么此航班提供最佳舒適度和設施。使用提供的航班數據作為推薦依據。確保在回答中明確說明每個屬性的選擇理由，不要重復航班詳細信息。"""elif data_type == "hotels":role = "AI酒店分析師"goal = "分析酒店選項并推薦最佳選擇，考慮價格、評分、位置和設施。"backstory = f"AI專家，提供基于多個因素的酒店選項深入分析。"description = """根據以下分析，生成最佳酒店的詳細推薦。回答應基于價格、評分、位置和設施等因素提供清晰的理由。**AI酒店推薦**推薦最佳酒店，基于以下分析：**推薦理由**：- **價格**：與其他選項相比，推薦的酒店在價格上最具性價比，提供的設施和服務物有所值。解釋為什么這是最佳選擇。- **評分**：與替代選項相比，評分更高，確保更好的整體客戶體驗。解釋為什么這使其成為最佳選擇。- **位置**：酒店位于優越位置，靠近重要景點，方便旅行者出行。- **設施**：酒店提供如Wi-Fi、游泳池、健身中心、免費早餐等設施。討論這些設施如何提升體驗，使其適合不同類型的旅行者。**理由要求**：- 確保每個部分清晰解釋為什么此酒店是基于價格、評分、位置和設施等因素的最佳選擇。- 與其他選項進行比較，解釋為什么此選項脫穎而出。- 提供簡潔、結構良好的理由，使推薦對旅行者清晰明了。- 推薦應幫助旅行者基于多個因素做出明智的決策，而不僅僅是單一因素。"""else:raise ValueError("無效的AI推薦數據類型")# 創建代理和任務analyze_agent = Agent(role=role,goal=goal,backstory=backstory,llm=llm_model,verbose=False)analyze_task = Task(description=f"{description}\n\n要分析的數據：\n{formatted_data}",agent=analyze_agent,expected_output=f"基于對所提供詳細信息的分析，解釋最佳{data_type}選擇的結構化推薦。")# 定義CrewAI工作流程analyst_crew = Crew(agents=[analyze_agent],tasks=[analyze_task],process=Process.sequential,verbose=False)# 執行CrewAI流程crew_results = await asyncio.to_thread(analyst_crew.kickoff)return str(crew_results)

現在，讓我們實現行程規劃代理：

async def generate_itinerary(destination, flights_text, hotels_text, check_in_date, check_out_date):"""根據航班和酒店信息生成詳細的旅行行程。"""try:# 將字符串日期轉換為datetime對象check_in = datetime.strptime(check_in_date, "%Y-%m-%d")check_out = datetime.strptime(check_out_date, "%Y-%m-%d")# 計算天數days = (check_out - check_in).daysllm_model = initialize_llm()analyze_agent = Agent(role="AI旅行規劃師",goal="根據航班和酒店信息為用戶創建詳細的行程",backstory="AI旅行專家，生成包含航班詳情、酒店住宿和目的地必游景點的按天劃分的行程。",llm=llm_model,verbose=False)analyze_task = Task(description=f"""根據以下詳細信息，為用戶創建一個{days}天的行程：**航班詳情**：{flights_text}**酒店詳情**：{hotels_text}**目的地**：{destination}**旅行日期**：{check_in_date}至{check_out_date}（{days}天）行程應包括：- 航班到達和出發信息- 酒店入住和退房詳情- 按天劃分的活動- 必游景點及預計參觀時間- 餐廳推薦，用于用餐- 當地交通提示**格式要求**：- 使用markdown格式，使用清晰的標題（#用于主標題，##用于天數，###用于部分）- 使用emoji表示不同類型的活動（地標用，餐廳用🍽?等）- 使用項目符號列出活動- 包括每個活動的預計時間- 格式化行程，使其視覺上美觀且易于閱讀""",agent=analyze_agent,expected_output="一個結構良好、視覺上美觀的markdown格式行程，包括航班、酒店和按天劃分的活動，帶有emoji、標題和項目符號。")
itinerary_planner_crew = Crew(agents=[analyze_agent],tasks=[analyze_task],process=Process.sequential,verbose=False)crew_results = await asyncio.to_thread(itinerary_planner_crew.kickoff)
return str(crew_results)

步驟7：航班和酒店搜索的API端點

現在讓我們實現我們的API端點：

獲取航班推薦

@app.post ("/search_flights/", response_model=AIResponse)
async def get_flight_recommendations(flight_request: FlightRequest):flights = await search_flights(flight_request)flights_text = format_travel_data("flights", flights)ai_recommendation = await get_ai_recommendation("flights", flights_text)return AIResponse(flights=flights, ai_flight_recommendation=ai_recommendation);

2. 獲取酒店推薦

@app.post ("/search_hotels/", response_model=AIResponse)
async def get_hotel_recommendations(hotel_request: HotelRequest):hotels = await search_hotels(hotel_request)hotels_text = format_travel_data("hotels", hotels)ai_recommendation = await get_ai_recommendation("hotels", hotels_text)return AIResponse(hotels=hotels, ai_hotel_recommendation=ai_recommendation);

步驟8：生成AI處理的行程

讓我們實現我們的行程生成端點：

@app.post("/generate_itinerary/", response_model=AIResponse)
async def get_itinerary(itinerary_request: ItineraryRequest):itinerary = await generate_itinerary(itinerary_request.destination,itinerary_request.flights,itinerary_request.hotels,itinerary_request.check_in_date,itinerary_request.check_out_date)return AIResponse(itinerary=itinerary)

最后，讓我們添加服務器啟動代碼：


# 運行FastAPI服務器
if __name__ == "__main__":logger.info("啟動旅行規劃API服務器")uvicorn.run(app, host="0.0.0.0", port=8000)

步驟9：構建Streamlit前端界面

現在讓我們使用Streamlit創建一個用戶友好的界面：

import streamlit as st
import requests
from datetime import datetime, timedelta# API URL
API_BASE_URL = "http://localhost:8000"
API_URL_FLIGHTS = f"{API_BASE_URL}/search_flights/"
API_URL_HOTELS = f"{API_BASE_URL}/search_hotels/"
API_URL_COMPLETE = f"{API_BASE_URL}/complete_search/"
API_URL_ITINERARY = f"{API_BASE_URL}/generate_itinerary/"
...