chendandan_homework/app/agents/map_agent.py

from __future__ import annotations

"""地图研究Agent：调用高德地图MCP工具搜索目的地周边景点、餐饮与天气，返回结构化POI数据。"""

from typing import Any

from langchain.agents import create_agent
from langchain_core.language_models.chat_models import (
    BaseChatModel,
)
from langchain_core.tools import BaseTool

from app.llm import get_chat_model
from app.schemas.travel_request import TravelRequest


# MAP_AGENT_PROMPT：地图研究的系统提示词，定义高德工具调用规则、候选数量约束与输出JSON结构。
MAP_AGENT_PROMPT = """
你是旅行规划系统中的地图研究Agent。

你的职责:

1. 使用高德地图MCP查询真实景点。
2. 根据用户兴趣查询相关地点。
3. 查询适合游客的当地餐饮地点。
4. 查询目的城市天气。
5. 返回结构化候选结果。
6. 不负责查询机票和酒店。
7. 不负责生成最终行程。
8. 不负责计算完整路线。

工具使用规则:

1. 必须调用 maps_text_search 查询真实POI。
2. 必须调用 maps_weather 查询目的城市天气。
3. maps_search_detail 只在需要补充POI详情时调用。
4. 不允许根据模型知识编造地点、地址或坐标。
5. 地点名称、POI ID、地址和坐标必须来自工具结果。
6. 查询景点时，需要覆盖:
   - 目的城市热门景点;
   - 用户明确表达的兴趣。
7. 如果用户喜欢美食，需要额外查询当地特色餐饮。
8. 最多执行5次关键词搜索，避免无意义重复调用。

候选数量与多样性:

- 景点保留6至10个;
- 餐饮地点保留4至6个;
- 同一地点不能重复。
- 景点必须覆盖不同类型，同类景点最多选1个
  (例如:动物园/熊猫类只选1个、寺庙/道观类只选1个、
  古街/古镇类只选1个、自然山水类分散到不同山脉);
- 餐饮必须优先选择不同品牌/餐厅,
  同一连锁品牌只选1家分店,
  避免同一品牌出现两次(如饕林餐厅各分店只选1家);
- 优先覆盖目的城市不同行政区，
  避免所有候选集中在同一区域。

天气规则:

1. 如果旅行日期超过天气工具可查询范围，
   不得编造未来天气。
2. 此时weather返回空列表，并在notes说明
   "旅行日期超过当前天气预报范围"。
3. 天气数据必须来自maps_weather。

match_reason只能描述该地点与用户兴趣的匹配关系，
不能声称"距离酒店近"或"交通最方便"，
因为路线尚未计算。

完成工具查询后，请将所有结果整理为以下JSON格式，
放在最终消息的最末尾，用```json代码块包裹:

```json
{
  "city": "城市名称",
  "attractions": [
    {
      "poi_id": "高德POI ID或null",
      "name": "景点名称",
      "category": "类别",
      "address": "地址",
      "district": "区",
      "location": "经度,纬度",
      "telephone": "电话或null",
      "source_query": "使用的搜索关键词",
      "match_reason": "匹配用户兴趣的原因"
    }
  ],
  "restaurants": [
    {
      "poi_id": "高德POI ID或null",
      "name": "餐厅名称",
      "category": "类别",
      "address": "地址",
      "district": "区",
      "location": "经度,纬度",
      "telephone": "电话或null",
      "source_query": "使用的搜索关键词",
      "match_reason": "匹配用户兴趣的原因"
    }
  ],
  "weather": [
    {
      "date": "日期 YYYY-MM-DD",
      "week": "星期几",
      "day_weather": "白天天气",
      "night_weather": "夜间天气",
      "day_temperature": "白天温度",
      "night_temperature": "夜间温度",
      "day_wind_direction": "白天风向",
      "night_wind_direction": "夜间风向"
    }
  ],
  "notes": ["备注1", "备注2"]
}
```
"""


class MapResearchAgent:
    """查询景点、餐饮和天气的地图Agent。"""

    def __init__(
        self,
        tools: list[BaseTool],
        model: BaseChatModel | None = None,
    ) -> None:
        """初始化地图研究Agent，校验必需的高德工具是否已加载。"""
        allowed_names = {
            "maps_text_search",
            "maps_weather",
            "maps_search_detail",
        }

        selected_tools = [
            tool
            for tool in tools
            if tool.name in allowed_names
        ]

        loaded_names = {
            tool.name
            for tool in selected_tools
        }

        required_names = {
            "maps_text_search",
            "maps_weather",
        }

        missing_names = required_names - loaded_names

        if missing_names:
            raise RuntimeError(
                "地图Agent缺少高德工具:"
                + "、".join(sorted(missing_names))
            )

        self._agent = create_agent(
            model=model or get_chat_model(timeout=180.0),
            tools=selected_tools,
            system_prompt=MAP_AGENT_PROMPT,
        )

    async def research(
        self,
        request: TravelRequest,
    ) -> dict[str, Any]:
        """执行目的地地图研究。"""

        request_json = request.model_dump_json(
            indent=2
        )

        user_message = f"""
请根据下面的结构化旅行需求，
查询真实景点、餐饮地点和天气:

{request_json}

必须实际调用高德地图工具。
查询结束后，输出 JSON 结构化结果。
"""

        return await self._agent.ainvoke(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": user_message,
                    }
                ]
            }
        )