AI 综合实战:旅行规划助手

"""综合实战: 旅行规划助手架构：- 主控 Agent: 协调整个规划流程- 航班专家: 搜索和推荐航班- 酒店专家: 搜索和推荐酒店- 活动专家: 推荐当地活动- 预算专家: 估算总成本使用 Subgraphs 实现每个专家"""from typing import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.checkpoint.memory import MemorySaverfrom langchain_core.messages import HumanMessagefrom langchain_openai import ChatOpenAIimport jsonimport osfrom datetime import datetimefrom dotenv import load_dotenvimport reload_dotenv()model = ChatOpenAI(    model="deepseek-chat",    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),    openai_api_base="https://api.deepseek.com/v1",    temperature=0,)# =========== State ===========class TravelPlanState(TypedDict):    destination: str    start_date: str    end_date: str    num_travelers: int    budget: float    flight_recommendations: list    hotel_recommendations: list    activity_recommendations: list    budget_breakdown: dict    final_plan: str    total_cost: floatclass FlightSearchState(TypedDict):    destination: str    start_date: str    end_date: str    num_travelers: int    search_results: list    recommendations: listclass HotelSearchState(TypedDict):    destination: str    start_date: str    end_date: str    num_travelers: int    search_results: list    recommendations: listclass ActivitySearchState(TypedDict):    destination: str    start_date: str    end_date: str    num_travelers: int    search_results: list    recommendations: list# ============= 模拟 API =============def search_flight_api(destination: str, date: str, travelers: int) -> list:    return [        {"airline": "航空公司 A", "departure": "08:00", "arrival": "12:00", "price": 1200, "duration": "4h"},        {"airline": "航空公司 B", "departure": "14:00", "arrival": "18:00", "price": 950, "duration": "4h"}    ]def search_hotels_api(destination: str, checkin: str, checkout: str) -> list:    return [        {"name": "豪华酒店", "rating": 5, "price_per_night": 800, "amenities": ["WIFI","游泳池","健身房","早餐"]},        {"name": "经济酒店", "rating": 3, "price_per_night": 300, "amenities": ["WIFI", "早餐"]}    ]def search_activities_api(destination: str) -> list:    return [        {"name":"城市观光", "duration": "4h", "price": 200},        {"name":"博物馆参观", "duration": "3h", "price": 150},        {"name":"美食之旅", "duration": "3h", "price": 300}    ]# =============== 工具函数：安全解析 LLM JSON ===============def safe_parse_json(text: str) -> dict | list | None:    """尝试从文本中提取 JSON，处理 markdown 代码块"""    # 去除可能的 markdown 标记    cleaned = re.sub(r'```json\s*|```\s*', '', text).strip()    try:        return json.loads(cleaned)    except:        # 尝试查找第一个 { 或 [ 到最后一个 } 或 ]        match = re.search(r'(\[.*\]|\{.*\})', cleaned, re.DOTALL)        if match:            try:                return json.loads(match.group(0))            except:                pass    return None# =============== 子图1：航班 ===============def flight_search_node(state: FlightSearchState) -> dict:    print(" 搜索航班。。。")    results = search_flight_api(state["destination"], state["start_date"], state["num_travelers"])    return {"search_results": results}def flight_recommend_node(state: FlightSearchState) -> dict:    print(" AI 分析航班...")    results = state.get("search_results", [])    if not results:        return {"recommendations": []}    prompt = f"""    从以下航班中选择最适合的航班，并给出理由。考虑价格、时间等因素。    航班列表：{json.dumps(results, ensure_ascii=False, indent=2)}    请按 JSON 格式返回（只返回JSON，不要其他内容）：    {{"index": 最佳航班在列表中的索引（整数）, "reason": "选择理由"}}    """    try:        response = model.invoke([HumanMessage(content=prompt)])        data = safe_parse_json(response.content)        if data and "index" in data:            idx = data["index"]            if isinstance(idx, int) and 0 <= idx < len(results):                selected = results[idx]                reason = data.get("reason", "综合推荐")                return {"recommendations": [{"flight": selected, "reason": reason}]}    except Exception as e:        print(f"  LLM解析失败: {e}")    # 回退：选价格最低    selected = min(results, key=lambda x: x["price"])    return {"recommendations": [{"flight": selected, "reason": "自动选择价格最低航班"}]}flight_subgraph = StateGraph(FlightSearchState)flight_subgraph.add_node("search", flight_search_node)flight_subgraph.add_node("recommend", flight_recommend_node)flight_subgraph.add_edge(START, "search")flight_subgraph.add_edge("search", "recommend")flight_subgraph.add_edge("recommend", END)flight_agent = flight_subgraph.compile()# =============== 子图2：酒店 ===============def hotels_search_node(state: HotelSearchState) -> dict:    print(" 搜索酒店。。。")    results = search_hotels_api(state["destination"], state["start_date"], state["end_date"])    return {"search_results": results}def hotels_recommend_node(state: HotelSearchState) -> dict:    print(" AI 分析酒店。。。")    results = state.get("search_results", [])    if not results:        return {"recommendations": []}    prompt = f"""    从以下酒店中选择最适合的酒店，并给出理由。考虑价格、评分、设施等。    酒店列表：{json.dumps(results, ensure_ascii=False, indent=2)}    请按 JSON 格式返回：{{"index": 最佳酒店索引, "reason": "理由"}}    """    try:        response = model.invoke([HumanMessage(content=prompt)])        data = safe_parse_json(response.content)        if data and "index" in data:            idx = data["index"]            if isinstance(idx, int) and 0 <= idx < len(results):                selected = results[idx]                reason = data.get("reason", "综合推荐")                return {"recommendations": [{"hotel": selected, "reason": reason}]}    except Exception as e:        print(f"  LLM解析失败: {e}")    # 回退：评分最高    selected = max(results, key=lambda x: (x["rating"], -x["price_per_night"]))    return {"recommendations": [{"hotel": selected, "reason": "自动选择评分最高酒店"}]}hotel_subgraph = StateGraph(HotelSearchState)hotel_subgraph.add_node("search", hotels_search_node)hotel_subgraph.add_node("recommend", hotels_recommend_node)hotel_subgraph.add_edge(START, "search")hotel_subgraph.add_edge("search", "recommend")hotel_subgraph.add_edge("recommend", END)hotel_agent = hotel_subgraph.compile()# =============== 子图3：活动 ===============def activity_search_node(state: ActivitySearchState) -> dict:    print(" 搜索活动。。。")    activities = search_activities_api(state["destination"])    return {"search_results": activities}def activity_filter_node(state: ActivitySearchState) -> dict:    print(" AI 筛选活动...")    activities = state.get("search_results", [])    if not activities:        return {"recommendations": []}    prompt = f"""    根据以下信息，从活动列表中选择最适合的 2 项活动：    目的地：{state['destination']}    日期：{state['start_date']} 至 {state['end_date']}    人数：{state['num_travelers']}    活动列表：{json.dumps(activities, ensure_ascii=False, indent=2)}    请按 JSON 数组格式返回选中的活动（只返回JSON数组，不要其他内容）：    例如：[{{"name": "活动名称", "reason": "推荐理由"}}, ...]    """    try:        response = model.invoke([HumanMessage(content=prompt)])        data = safe_parse_json(response.content)        if isinstance(data, list) and len(data) > 0:            selected_activities = []            for item in data:                found = next((a for a in activities if a["name"] == item.get("name")), None)                if found:                    found_copy = found.copy()                    found_copy["reason"] = item.get("reason", "")                    selected_activities.append(found_copy)            if selected_activities:                return {"recommendations": selected_activities[:2]}    except Exception as e:        print(f"  LLM筛选失败: {e}")    # 回退：前两项    return {"recommendations": activities[:2]}activity_subgraph = StateGraph(ActivitySearchState)activity_subgraph.add_node("search", activity_search_node)activity_subgraph.add_node("filter", activity_filter_node)activity_subgraph.add_edge(START, "search")activity_subgraph.add_edge("search", "filter")activity_subgraph.add_edge("filter", END)activity_agent = activity_subgraph.compile()# =============== 主图 ===============def call_flight_expert(state: TravelPlanState) -> dict:    print(" 调用航班专家。。。")    result = flight_agent.invoke({        "destination": state["destination"],        "start_date": state["start_date"],        "end_date": state["end_date"],        "num_travelers": state["num_travelers"],        "search_results": [],        "recommendations": []    })    return {"flight_recommendations": result.get("recommendations", [])}def call_hotel_expert(state: TravelPlanState) -> dict:    print(" 调用酒店专家。。。")    result = hotel_agent.invoke({        "destination": state["destination"],        "start_date": state["start_date"],        "end_date": state["end_date"],        "num_travelers": state["num_travelers"],        "search_results": [],        "recommendations": []    })    return {"hotel_recommendations": result.get("recommendations", [])}def call_activity_expert(state: TravelPlanState) -> dict:    print(" 调用活动专家。。。")    result = activity_agent.invoke({        "destination": state["destination"],        "start_date": state["start_date"],        "end_date": state["end_date"],        "num_travelers": state["num_travelers"],        "search_results": [],        "recommendations": []    })    return {"activity_recommendations": result.get("recommendations", [])}def calculate_budget(state: TravelPlanState) -> dict:    flight_recs = state.get("flight_recommendations", [])    hotel_recs = state.get("hotel_recommendations", [])    activity_recs = state.get("activity_recommendations", [])    if flight_recs:        flight_price = flight_recs[0]["flight"]["price"]        flight_cost = flight_price * state["num_travelers"]    else:        flight_cost = 2000 * state["num_travelers"]    if hotel_recs:        hotel_price = hotel_recs[0]["hotel"]["price_per_night"]        start = datetime.strptime(state["start_date"], "%Y-%m-%d")        end = datetime.strptime(state["end_date"], "%Y-%m-%d")        nights = (end - start).days        if nights <= 0:            nights = 1        hotel_cost = hotel_price * nights    else:        hotel_cost = 500    if activity_recs:        activity_cost = sum(a.get("price", 0) for a in activity_recs) * state["num_travelers"]    else:        activity_cost = 0    total = flight_cost + hotel_cost + activity_cost    breakdown = {"flight": flight_cost, "hotel": hotel_cost, "activities": activity_cost, "total": total}    return {"budget_breakdown": breakdown, "total_cost": total}# =============== 核心修改：最终计划生成（增加保底方案） ===============def generate_final_plan(state: TravelPlanState) -> dict:    print(" 生成最终方案...")    # 先构造一个“保底计划”，如果模型返回空则使用这个    flight_info = state["flight_recommendations"][0] if state.get("flight_recommendations") else {"flight": {"airline": "无", "price": 0}, "reason": "无"}    hotel_info = state["hotel_recommendations"][0] if state.get("hotel_recommendations") else {"hotel": {"name": "无", "price_per_night": 0}, "reason": "无"}    activities = state.get("activity_recommendations", [])    breakdown = state.get("budget_breakdown", {})    # 构建备用计划（纯文本）    fallback_plan = f"""【备用旅行计划】（由于 AI 生成失败，此为模板）目的地：{state['destination']}日期：{state['start_date']} 至 {state['end_date']}人数：{state['num_travelers']}航班：{flight_info['flight']['airline']}，价格 {flight_info['flight']['price']} 元/人酒店：{hotel_info['hotel']['name']}，每晚 {hotel_info['hotel']['price_per_night']} 元活动：{', '.join([a['name'] for a in activities]) if activities else'无'}费用估算：- 机票：{breakdown.get('flight', 0)} 元- 住宿：{breakdown.get('hotel', 0)} 元- 活动：{breakdown.get('activities', 0)} 元- 总计：{breakdown.get('total', 0)} 元"""    try:        prompt = f"""        生成一份详细的旅行计划，包含每日行程安排和费用明细。        目的地: {state["destination"]}        日期：{state['start_date']} 至 {state['end_date']}        人数: {state['num_travelers']}        航班: {json.dumps(flight_info, ensure_ascii=False)}        酒店: {json.dumps(hotel_info, ensure_ascii=False)}        活动: {json.dumps(activities, ensure_ascii=False)}        预算分解: {json.dumps(breakdown, ensure_ascii=False)}        总预算: {state['budget']}        """        response = model.invoke([HumanMessage(content=prompt)])        plan = response.content.strip()        if not plan:            plan = fallback_plan    except Exception as e:        print(f"  模型调用异常: {e}")        plan = fallback_plan    # 确保最终计划不为空    if not plan:        plan = "无法生成计划，请检查网络或API配置。"    return {"final_plan": plan}# 构建主图main_graph = StateGraph(TravelPlanState)main_graph.add_node("flight_expert", call_flight_expert)main_graph.add_node("hotel_expert", call_hotel_expert)main_graph.add_node("activity_expert", call_activity_expert)main_graph.add_node("budget_calc", calculate_budget)main_graph.add_node("generate_plan", generate_final_plan)main_graph.add_edge(START, "flight_expert")main_graph.add_edge(START, "hotel_expert")main_graph.add_edge(START, "activity_expert")main_graph.add_edge("flight_expert", "budget_calc")main_graph.add_edge("hotel_expert", "budget_calc")main_graph.add_edge("activity_expert", "budget_calc")main_graph.add_edge("budget_calc", "generate_plan")main_graph.add_edge("generate_plan", END)# memory = MemorySaver() checkpointer=memorytravel_planner_app = main_graph.compile()# ============= 运行演示 =============def run_travel_planner_demo():    print("\n" + "="*60)    print("多智能体旅行规划助手（修正版）")    print("="*60 + "\n")    request = {        "destination": "日本东京",        "start_date": "2026-08-01",        "end_date": "2026-08-05",        "num_travelers": 2,        "budget": 20000.0,        "flight_recommendations": [],        "hotel_recommendations": [],        "activity_recommendations": [],        "budget_breakdown": {},        "final_plan": "",        "total_cost": 0.0    }    print(f" 规划需求：")    print(f" 目的地：{request['destination']}")    print(f" 日期：{request['start_date']} ~ {request['end_date']}")    print(f" 人数：{request['num_travelers']}")    print(f" 预算：{request['budget']}\n")    print("开始规划...\n")    result = travel_planner_app.invoke(request)    print("\n" + "="*60)    print("规划结果")    print("="*60 + "\n")    # 航班推荐    print(" 航班推荐：")    if result.get("flight_recommendations"):        f = result["flight_recommendations"][0]        flight = f["flight"]        print(f"  {flight['airline']}  -  {flight['departure']}~{flight['arrival']}")        print(f"  价格：{flight['price']} x {request['num_travelers']} = {flight['price'] * request['num_travelers']}")        print(f"  推荐理由：{f.get('reason', '')}\n")    else:        print("  无航班推荐\n")    # 酒店推荐    print(" 酒店推荐：")    if result.get("hotel_recommendations"):        h = result["hotel_recommendations"][0]        hotel = h["hotel"]        print(f"  {hotel['name']} ({hotel['rating']}星)")        print(f"  价格：{hotel['price_per_night']}/晚")        print(f"  推荐理由：{h.get('reason', '')}\n")    else:        print("  无酒店推荐\n")    # 活动推荐    print(" 活动推荐：")    if result.get("activity_recommendations"):        for act in result["activity_recommendations"]:            print(f"  {act['name']} ({act['duration']}) - {act['price']}")            if act.get("reason"):                print(f"    推荐理由：{act['reason']}")        print()    else:        print("  无活动推荐\n")    # 预算    print(" 预算分解：")    bd = result.get("budget_breakdown", {})    print(f"  航班：{bd.get('flight', 0)}")    print(f"  酒店：{bd.get('hotel', 0)}")    print(f"  活动：{bd.get('activities', 0)}")    print(f"  总计：{bd.get('total', 0)}\n")    if bd.get("total", 0) <= request["budget"]:        print(f"  ✅ 在预算内(预算：{request['budget']})\n")    else:        print(f"  ⚠️ 超出预算 {bd['total'] - request['budget']}\n")    # ====== 关键：打印最终计划 ======    print("="*60)    print(" 完整方案 ")    print("="*60)    final = result.get("final_plan", "")    #print(final)          # 直接打印     # 分行打印，避免截断    for line in final.splitlines():        print(line)    # ====== 同时保存到文件（方案一） ======    with open("travel_plan.txt", "w", encoding="utf-8") as f:        f.write(final)    print("（完整计划已保存到 travel_plan.txt）")        print("\n" + "="*60)    print(" 规划完成 ")    print("="*60 + "\n")if __name__ == "__main__":    run_travel_planner_demo()

============================================================多智能体旅行规划助手（修正版）============================================================ 规划需求： 目的地：日本东京 日期：2026-08-01 ~ 2026-08-05 人数：2 预算：20000.0开始规划... 调用活动专家。。。 搜索活动。。。 AI 筛选活动... 调用航班专家。。。 搜索航班。。。 AI 分析航班... 调用酒店专家。。。 搜索酒店。。。 AI 分析酒店。。。 生成最终方案...============================================================...============================================================ 规划完成 ============================================================

### 日本东京5日深度旅行计划 (2026-08-01 至 2026-08-05)**旅行人数：** 2人  **总预算：** 20,000元人民币  **已选航班：** 航空公司B（14:00-18:00，950元/人）  **已选酒店：** 豪华酒店（5星，800元/晚，含早餐）  **已选活动：** 城市观光（4小时，200元/人）、美食之旅（3小时，300元/人）---## 每日行程安排### Day 1（8月1日，周六）：抵达东京，入住休整| 时间 | 安排 | 备注 ||------|------|------|| 14:00 | 国内出发，乘坐航空公司B航班 | 建议提前2小时到机场 || 18:00 | 抵达东京成田/羽田机场 | 出关、取行李约1小时 || 19:00 | 乘坐机场快线/利木津巴士前往市区 | 费用约100元/人（预算外，可刷交通卡） || 20:30 | 抵达豪华酒店，办理入住 | 酒店位于新宿/银座等核心区 || 21:00 | 酒店周边散步，便利店采购 | 熟悉环境，购买水、零食等 || 22:00 | 回酒店休息 | 调整时差，养精蓄锐 |**餐饮：** 晚餐可在机场或酒店附近简餐（预算外，约100元/人）  **住宿：** 豪华酒店（含次日早餐）---### Day 2（8月2日，周日）：经典城市观光| 时间 | 安排 | 备注 ||------|------|------|| 08:00 | 酒店享用自助早餐 | 已含在房费中 || 09:00 | **城市观光活动（4小时）** | 专业导游带领，覆盖皇居、浅草寺、东京塔等主要地标 || 13:00 | 活动结束，午餐 | 推荐浅草附近品尝天妇罗或寿司（预算外，约150元/人） || 14:30 | 自由活动：银座/涩谷购物 | 可逛百货公司、药妆店 || 18:00 | 晚餐：新宿居酒屋体验 | 预算外，约200元/人 || 20:00 | 东京都厅展望台看夜景 | 免费，俯瞰东京全景 || 21:30 | 返回酒店休息 | |**餐饮：** 早餐（含）、午餐+晚餐（预算外，约350元/人）  **住宿：** 豪华酒店---### Day 3（8月3日，周一）：美食之旅+自由探索| 时间 | 安排 | 备注 ||------|------|------|| 08:00 | 酒店早餐 | 已含 || 09:30 | **美食之旅（3小时）** | 走访筑地市场、品尝寿司、和果子等特色美食 || 12:30 | 午餐（美食之旅已含部分试吃） | 若未饱可加餐（预算外，约50元/人） || 14:00 | 自由活动：明治神宫+原宿 | 感受传统文化与潮流文化碰撞 || 17:00 | 涩谷十字路口打卡 | 世界最繁忙路口 || 18:30 | 晚餐：烤肉或涮涮锅 | 预算外，约250元/人 || 20:00 | 六本木之丘展望台 | 门票约120元/人（预算外） || 22:00 | 返回酒店 | |**餐饮：** 早餐（含）、午餐（部分含）、晚餐（预算外，约250元/人）  **住宿：** 豪华酒店---### Day 4（8月4日，周二）：富士山/箱根一日游（可选）| 时间 | 安排 | 备注 ||------|------|------|| 07:00 | 酒店早餐 | 已含 || 08:00 | 参加富士山/箱根一日游旅行团 | 费用约600元/人（预算外，可提前预订） || 10:00 | 抵达富士山五合目 | 欣赏壮丽山景 || 12:00 | 午餐（团餐或自理） | 约100元/人 || 14:00 | 箱根海盗船+大涌谷 | 体验火山地貌，品尝黑鸡蛋 || 17:00 | 返回东京市区 | || 19:30 | 晚餐：拉面或乌冬面 | 预算外，约80元/人 || 20:30 | 秋叶原电器街/动漫店 | 自由活动 || 22:00 | 返回酒店 | |**备选方案：** 若不想去富士山，可改为台场海滨公园+teamLab数字艺术馆（门票约200元/人）  **餐饮：** 早餐（含）、午餐+晚餐（预算外，约180元/人）  **住宿：** 豪华酒店---### Day 5（8月5日，周三）：返程| 时间 | 安排 | 备注 ||------|------|------|| 08:00 | 酒店早餐 | 已含 || 09:00 | 退房，行李寄存 | 可寄存酒店前台 || 09:30 | 上野公园+阿美横町购物 | 购买伴手礼、药妆 || 12:00 | 午餐：鳗鱼饭或寿司 | 预算外，约150元/人 || 13:30 | 返回酒店取行李 | || 14:00 | 乘坐机场快线前往机场 | 费用约100元/人 || 16:00 | 抵达机场，办理登机 | 建议提前2小时 || 18:00 | 航班起飞，返回国内 | |**餐饮：** 早餐（含）、午餐（预算外，约150元/人）---## 费用明细### 已包含费用（已预订）| 项目 | 单价 | 数量 | 小计（元） ||------|------|------|------------|| 机票（航空公司B） | 950元/人 | 2人 | **1,900** || 酒店（豪华酒店，5晚） | 800元/晚 | 4晚 | **3,200** || 城市观光活动 | 200元/人 | 2人 | **400** || 美食之旅活动 | 300元/人 | 2人 | **600** || **已包含小计** | | | **6,100** |### 预算外预估费用（现场支付）| 项目 | 预估费用（元/人） | 2人合计（元） ||------|-------------------|---------------|| 机场交通（往返） | 200 | 400 || 市内交通（地铁/公交） | 300 | 600 || 餐饮（除早餐外） | 1,000 | 2,000 || 景点门票（六本木、富士山等） | 800 | 1,600 || 购物及其他 | 1,000 | 2,000 || **预算外预估小计** | | **6,600** |### 总费用预估| 项目 | 金额（元） ||------|------------|| 已包含费用 | 6,100 || 预算外预估 | 6,600 || **总预估** | **12,700** || **总预算** | **20,000** || **剩余可支配** | **7,300** |---## 温馨提示1. **交通卡：** 建议购买Suica或Pasmo卡，方便乘坐地铁和便利店消费。2. **语言：** 东京大部分地方可用英语沟通，建议下载翻译APP。3. **天气：** 8月东京炎热潮湿，建议携带防晒用品、雨具和轻薄衣物。4. **网络：** 建议提前购买日本SIM卡或租用移动WiFi。5. **现金：** 部分小店和景点只收现金，建议携带适量日元。6. **购物退税：** 在贴有“Tax Free”标识的商店消费满5000日元可退税。7. **保险：** 建议购买旅行保险，保障意外和医疗。祝您旅途愉快！如有任何调整需求，可随时修改计划。