chendandan_homework/app/agents/itinerary_agent.py

from __future__ import annotations

"""行程规划Agent：基于候选资源（航班/酒店/景点）与地图研究结果，生成包含每日活动安排的结构化多日行程。"""

import json
from typing import Any

from langchain_core.language_models.chat_models import (
    BaseChatModel,
)
from langchain_core.messages import (
    HumanMessage,
    SystemMessage,
)

from app.llm import get_chat_model
from app.schemas.itinerary import ItineraryPlan
from app.schemas.place import MapResearchResult
from app.schemas.route import (
    RouteEvaluationResult,
)
from app.schemas.selection import (
    CandidateSelectionResult,
)
from app.schemas.travel_request import (
    TravelRequest,
)


# ITINERARY_PLANNER_PROMPT：行程规划的系统提示词，定义每日行程的JSON输出结构与修改要求。
ITINERARY_PLANNER_PROMPT = """
你是多角色旅行规划系统中的行程规划专家。

上游角色已经完成：

1. Requirement Analyst：解析用户需求。
2. Resource Search Agent：查询真实航班和酒店。
3. Destination Research Agent：查询真实景点、
   餐厅和天气。
4. Candidate Selector：筛选航班和酒店。
5. Route Evaluator：评估酒店到景点的距离。

你的职责是基于上游结果，选择一个航班方案、
一家酒店，并生成逐日旅行安排。

严格规则：

一、真实性

1. 只能选择输入候选中存在的航班、酒店、
   景点和餐厅。
2. 不得编造候选中不存在的名称、ID、地址、
   坐标、航班号或价格。
3. 景点和餐厅的reference_id使用输入中的poi_id。
4. 如果poi_id为空，可以保留reference_id=null，
   但名称必须和输入候选完全一致。
5. 不确定的信息放入assumptions或warnings，
   不得伪装为确定事实。

二、航班

1. 往返旅行必须选择一个完整round_trip_options。
2. combination_id、outbound_option_id和
   return_option_id必须复制候选中的真实ID。
3. quoted_price采用候选中的接口报价。
4. 不得把去程和返程价格再次相加。
5. 到达日需要考虑航班到达时间。
6. 返程日必须为前往机场预留合理时间。

三、酒店

1. 优先使用Route Evaluator排名第一的酒店。
2. 如选择其他酒店，必须给出明确原因。
3. check_in_date必须等于用户的departure_date
   （不是航班实际到达日期）。
4. check_out_date必须等于用户的return_date。
5. nights必须等于
   (check_out_date - check_in_date)的天数。
   例如：8月10日入住、8月14日退房
   → nights = 4。
6. price_per_night从酒店候选复制。
7. estimated_total_price = price_per_night × nights；
   该值后续还会由Validator重新计算。

四、每日行程

1. 必须覆盖旅行期间的每一个自然日。
2. 日期必须连续。
3. 到达日和返程日不能安排过满。
4. 同一个景点不得重复安排。
5. 单个活动的结束时间必须晚于开始时间。
6. 景点尽量按照位置和主题组合。
7. 仅可使用输入中存在的餐厅。
8. 活动名称必须与候选名称完全一致，
   不得添加"早餐："、"午餐："、"晚餐："等前缀。
9. 不需要强行为每一餐指定餐厅。
10. 未查询到真实交通时间时，
   不得编造精确交通时间。

五、节奏

- relaxed：
  每天原则上不超过2个主要景点，
  保留休息和自由活动时间。
- normal：
  每天原则上2至3个主要景点。
- intensive：
  每天可安排3至4个主要景点，
  但仍需考虑交通和开放时间风险。

六、天气

1. 有真实天气数据时，可以据此安排室内外活动。
2. 没有天气数据时，不得编造天气。
3. 在warnings中提示用户出发前再次确认天气。

七、费用

1. 只能使用已有的航班和酒店报价。
2. 未查询门票或餐饮价格时，
   estimated_cost应为null。
3. 总预算是否超支由后续Validator负责。

八、输出格式（必须严格遵守字段名）

你必须返回一个 JSON 对象，顶层字段如下：

{
  "title": "行程标题（字符串）",
  "overview": "行程概览（字符串）",
  "selected_flight": {
    "flight_type": "round_trip 或 one_way",
    "combination_id": "来自 round_trip_options 的组合ID（仅往返）",
    "outbound_option_id": "去程航班ID（必填）",
    "return_option_id": "返程航班ID（往返必填，单程为null）",
    "quoted_price": 接口报价数字,
    "currency": "CNY",
    "outbound_flight_numbers": ["MU001"],
    "return_flight_numbers": ["MU002"],
    "outbound_departure_time": "去程出发时间字符串",
    "outbound_arrival_time": "去程到达时间字符串",
    "return_departure_time": "返程出发时间字符串",
    "return_arrival_time": "返程到达时间字符串",
    "selection_reason": "选择理由"
  },
  "selected_hotel": {
    "hotel_id": "来自候选的酒店ID",
    "name": "酒店名称",
    "check_in_date": "入住日期 YYYY-MM-DD（必须等于 departure_date）",
    "check_out_date": "退房日期 YYYY-MM-DD（必须等于 return_date）",
    "nights": "check_out_date - check_in_date 的天数，如 8/10→8/14 为 4",
    "price_per_night": 每晚价格,
    "estimated_total_price": price_per_night × nights,
    "currency": "CNY",
    "address": "地址文本（可选）",
    "location": "经纬度字符串如 \"104.07,30.66\"（可选，不是对象）",
    "selection_reason": "选择理由"
  },
  "days": [
    {
      "day_index": 从1开始的整数,
      "date": "YYYY-MM-DD",
      "theme": "当日主题",
      "activities": [
        {
          "sequence": 从1开始递增的整数,
          "activity_type": "flight|hotel_check_in|hotel_check_out|attraction|restaurant|transport|free_time|other",
          "reference_id": "景点或餐厅的poi_id（航班/入住等为null）",
          "name": "活动名称",
          "start_time": "HH:MM",
          "end_time": "HH:MM",
          "address": "地址（可选）",
          "location": "经纬度（可选）",
          "transport_mode": "walking|public_transit|driving|taxi|unknown（可选）",
          "estimated_transport_minutes": 预估交通分钟（可选）,
          "estimated_cost": 预估费用（可选）,
          "notes": "备注（可选）"
        }
      ],
      "daily_notes": ["当日备注（可选）"]
    }
  ],
  "highlights": ["行程亮点"],
  "assumptions": ["不确定的假设"],
  "warnings": ["需要用户注意的点"]
}

注意：
- 不要在最外层包裹 {"itineraryPlan": {...}}，直接输出上述 JSON。
- 所有日期格式为 YYYY-MM-DD，时间格式为 HH:MM。
- 必须从输入候选中复制真实数据，不得编造。
"""


def _compact_flight(
    flight: dict[str, Any],
) -> dict[str, Any]:
    """保留Planner真正需要的航班字段。"""

    return {
        "option_id": flight.get("option_id"),
        "flight_numbers": flight.get(
            "flight_numbers",
            [],
        ),
        "airlines": flight.get("airlines", []),
        "departure_airport": flight.get(
            "departure_airport"
        ),
        "arrival_airport": flight.get(
            "arrival_airport"
        ),
        "departure_time": flight.get(
            "departure_time"
        ),
        "arrival_time": flight.get(
            "arrival_time"
        ),
        "duration_minutes": flight.get(
            "duration_minutes"
        ),
        "stop_count": flight.get("stop_count"),
        "price": flight.get("price"),
        "currency": flight.get("currency"),
    }


def build_planner_context(
    request: TravelRequest,
    selection: CandidateSelectionResult,
    map_result: MapResearchResult,
    route_result: RouteEvaluationResult,
    revision_feedback: list[str] | None = None,
) -> dict[str, Any]:
    """把共享State压缩为Planner需要的上下文。

    避免将几十条原始航班和酒店全部放入Prompt。
    """

    round_trip_options: list[
        dict[str, Any]
    ] = []

    for option in selection.round_trip_options[:5]:
        round_trip_options.append(
            {
                "combination_id": (
                    option.combination_id
                ),
                "score": option.score,
                "quoted_price": option.quoted_price,
                "currency": option.currency,
                "outbound": _compact_flight(
                    option.outbound
                ),
                "return_flight": _compact_flight(
                    option.return_flight
                ),
                "reasons": option.reasons,
                "warnings": option.warnings,
            }
        )

    one_way_options: list[
        dict[str, Any]
    ] = []

    for candidate in selection.one_way_options[:5]:
        one_way_options.append(
            {
                "score": candidate.score,
                "flight": _compact_flight(
                    candidate.flight
                ),
                "reasons": candidate.reasons,
                "warnings": candidate.warnings,
            }
        )

    evaluated_hotels: list[
        dict[str, Any]
    ] = []

    for evaluated in (
        route_result.evaluated_hotels[:3]
    ):
        hotel = evaluated.hotel

        evaluated_hotels.append(
            {
                "hotel_id": evaluated.hotel_id,
                "name": evaluated.hotel_name,
                "final_score": (
                    evaluated.final_score
                ),
                "base_hotel_score": (
                    evaluated.base_hotel_score
                ),
                "average_distance_meters": (
                    evaluated
                    .average_distance_meters
                ),
                "average_duration_seconds": (
                    evaluated
                    .average_duration_seconds
                ),
                "nearest_subway_distance_meters": (
                    evaluated
                    .nearest_subway_distance_meters
                ),
                "price_per_night": hotel.get(
                    "price_per_night"
                ),
                "total_price": hotel.get(
                    "total_price"
                ),
                "currency": hotel.get("currency"),
                "coordinates": hotel.get(
                    "coordinates"
                ),
                "reasons": evaluated.reasons,
                "warnings": evaluated.warnings,
                "route_legs": [
                    {
                        "destination_name": (
                            leg.destination_name
                        ),
                        "distance_meters": (
                            leg.distance_meters
                        ),
                        "duration_seconds": (
                            leg.duration_seconds
                        ),
                        "source": leg.source,
                    }
                    for leg in evaluated.route_legs
                ],
            }
        )

    attractions = [
        place.model_dump(mode="json")
        for place in map_result.attractions[:10]
    ]

    restaurants = [
        place.model_dump(mode="json")
        for place in map_result.restaurants[:6]
    ]

    weather = [
        item.model_dump(mode="json")
        for item in map_result.weather
    ]

    return {
        "travel_request": request.model_dump(
            mode="json"
        ),
        "flight_candidates": {
            "flight_type": selection.flight_type,
            "round_trip_options": (
                round_trip_options
            ),
            "one_way_options": one_way_options,
            "warnings": selection.warnings,
        },
        "evaluated_hotels": evaluated_hotels,
        "selected_hotel_id_from_route_evaluator": (
            route_result.selected_hotel_id
        ),
        "attractions": attractions,
        "restaurants": restaurants,
        "weather": weather,
        "map_notes": map_result.notes,
        "route_warnings": route_result.warnings,
        "revision_feedback": (
            revision_feedback or []
        ),
    }


class ItineraryPlannerAgent:
    """负责综合上游角色结果生成逐日行程。"""

    def __init__(
        self,
        model: BaseChatModel | None = None,
    ) -> None:
        """初始化行程规划Agent，支持注入自定义模型。"""
        base_model = model or get_chat_model(
            timeout=180.0
        )

        # json_mode 通过 response_format 约束模型输出 JSON，
        # 对 DeepSeek 等推理模型的兼容性更好。
        self._structured_model = (
            base_model.with_structured_output(
                ItineraryPlan,
                method="json_mode",
            )
        )

    async def plan(
        self,
        request: TravelRequest,
        selection: CandidateSelectionResult,
        map_result: MapResearchResult,
        route_result: RouteEvaluationResult,
        revision_feedback: list[str] | None = None,
    ) -> dict[str, Any]:
        """生成结构化旅行方案。"""

        context = build_planner_context(
            request=request,
            selection=selection,
            map_result=map_result,
            route_result=route_result,
            revision_feedback=revision_feedback,
        )

        context_json = json.dumps(
            context,
            ensure_ascii=False,
            indent=2,
        )

        feedback_instruction = ""

        if revision_feedback:
            feedback_instruction = (
                "\n这是审查角色上一轮提出的修改要求，"
                "新方案必须逐项处理：\n"
                + "\n".join(
                    f"- {item}"
                    for item in revision_feedback
                )
            )

        user_message = f"""
请根据以下上游角色提供的真实数据生成旅行方案。

{context_json}

{feedback_instruction}

最终必须以 JSON 格式返回 ItineraryPlan 结构化结果。
不要输出结构化结果之外的解释。
"""

        response = await self._structured_model.ainvoke(
            [
                SystemMessage(
                    content=(
                        ITINERARY_PLANNER_PROMPT
                    )
                ),
                HumanMessage(
                    content=user_message
                ),
            ]
        )

        if isinstance(response, ItineraryPlan):
            structured = response
        else:
            # 兼容 LLM 可能包裹在 {"itineraryPlan": {...}} 中的情况。
            if (
                isinstance(response, dict)
                and "itineraryPlan" in response
                and len(response) == 1
            ):
                response = response["itineraryPlan"]

            # 兼容 LLM 将 coordinates 对象误填到
            # selected_hotel.location 的情况。
            if isinstance(response, dict):
                hotel = response.get("selected_hotel")
                if isinstance(hotel, dict):
                    loc = hotel.get("location")
                    if isinstance(loc, dict):
                        lat = loc.get("latitude")
                        lng = loc.get("longitude")
                        if lat is not None and lng is not None:
                            hotel["location"] = (
                                f"{lng},{lat}"
                            )

            structured = ItineraryPlan.model_validate(response)

        return {
            "structured_response": structured,
            "messages": [],
        }