打造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/"
...