chendandan_homework/app/graph/route_nodes.py

from __future__ import annotations
"""路线评估节点：组合候选航班与酒店结果，调用高德距离工具评估酒店与景点的实际交通距离。"""

import json
from typing import Any, Literal

from langchain_core.messages import ToolMessage
from langchain_core.tools import BaseTool

from app.graph.state import TravelState
from app.schemas.route import (
    RouteEvaluationResult,
    RouteLeg,
)
from app.services.route_evaluation_service import (
    RouteEvaluationService,
)


def _to_mapping(
    value: Any,
) -> dict[str, Any] | None:
    """把dict或Pydantic模型统一转换为dict，方便取structuredContent。"""

    if isinstance(value, dict):
        return value

    model_dump = getattr(
        value,
        "model_dump",
        None,
    )

    if callable(model_dump):
        dumped = model_dump()

        if isinstance(dumped, dict):
            return dumped

    return None


def extract_mcp_payload(
    result: Any,
) -> dict[str, Any]:
    """提取直接调用MCP工具后的结构化内容。

    LangChain Adapter可能把structuredContent
    放入ToolMessage.artifact。
    """

    if isinstance(result, ToolMessage):
        artifact = getattr(
            result,
            "artifact",
            None,
        )

        artifact_mapping = _to_mapping(
            artifact
        )

        if artifact_mapping:
            for key in (
                "structured_content",
                "structuredContent",
            ):
                structured = (
                    artifact_mapping.get(key)
                )

                if isinstance(
                    structured,
                    dict,
                ):
                    return structured

            return artifact_mapping

        structured_attr = getattr(
            artifact,
            "structured_content",
            None,
        )

        if isinstance(
            structured_attr,
            dict,
        ):
            return structured_attr

        result = result.content

    mapping = _to_mapping(result)

    if mapping:
        for key in (
            "structured_content",
            "structuredContent",
        ):
            structured = mapping.get(key)

            if isinstance(structured, dict):
                return structured

        return mapping

    if isinstance(result, str):
        parsed = json.loads(result)

        if isinstance(parsed, dict):
            return parsed

    if isinstance(result, list):
        for block in result:
            if not isinstance(block, dict):
                continue

            text = block.get("text")

            if not isinstance(text, str):
                continue

            try:
                parsed = json.loads(text)
            except json.JSONDecodeError:
                continue

            if isinstance(parsed, dict):
                return parsed

    raise RuntimeError(
        "无法从高德MCP结果中提取结构化数据。"
    )

def find_numeric_values(
    value: Any,
    target_key: str,
) -> list[float]:
    """递归提取指定字段中的数字。"""

    results: list[float] = []

    if isinstance(value, dict):
        for key, item in value.items():
            if key == target_key:
                if (
                    isinstance(
                        item,
                        (int, float, str),
                    )
                    and not isinstance(item, bool)
                ):
                    try:
                        results.append(
                            float(item)
                        )
                    except ValueError:
                        pass

            results.extend(
                find_numeric_values(
                    item,
                    target_key,
                )
            )

    elif isinstance(value, list):
        for item in value:
            results.extend(
                find_numeric_values(
                    item,
                    target_key,
                )
            )

    return results


def get_hotel_location(
    hotel: dict[str, Any],
) -> str | None:
    """从标准化酒店结果提取高德坐标字符串。"""

    coordinates = hotel.get("coordinates")

    if not isinstance(coordinates, dict):
        return None

    longitude = coordinates.get("longitude")
    latitude = coordinates.get("latitude")

    if longitude is None or latitude is None:
        return None

    return f"{longitude},{latitude}"


def route_after_selection(
    state: TravelState,
) -> Literal[
    "research_destination",
    "end",
]:
    """筛选完成后路由：有结果进入地图研究，无结果结束。"""

    if state.get("errors"):
        return "end"

    if not state.get(
        "candidate_selection_result"
    ):
        return "end"

    return "research_destination"


def route_after_map_research(
    state: TravelState,
) -> Literal[
    "evaluate_routes",
    "end",
]:
    """地图研究完成后路由：有结果进入路线评估，无结果结束。"""

    if state.get("errors"):
        return "end"

    if not state.get(
        "map_research_result"
    ):
        return "end"

    if not state.get(
        "candidate_selection_result"
    ):
        return "end"

    return "evaluate_routes"

def make_route_evaluation_node(
    distance_tool: BaseTool,
    around_search_tool: BaseTool | None,
    *,
    service: RouteEvaluationService
    | None = None,
    max_hotels: int = 3,
    max_attractions: int = 4,
):
    """创建酒店位置与路线评估节点。

    为控制调用量，仅评估前3家酒店和前4个景点。
    """

    route_service = (
        service
        or RouteEvaluationService()
    )

    async def route_evaluation_node(
        state: TravelState,
    ) -> dict[str, Any]:
        """路线评估节点：调用RouteEvaluationService评估酒店位置并计算景点间路线距离。"""

        request = state["travel_request"]

        selection = state[
            "candidate_selection_result"
        ]

        map_result = state[
            "map_research_result"
        ]

        hotel_candidates = (
            selection.hotels[:max_hotels]
        )

        attractions = [
            attraction
            for attraction
            in map_result.attractions
            if attraction.location
        ][:max_attractions]

        warnings: list[str] = []

        if not hotel_candidates:
            message = "没有可用于路线评估的酒店。"

            return {
                "errors": [message],
                "route_warnings": [message],
                "final_answer": message,
            }

        if not attractions:
            message = (
                "没有带有效坐标的景点候选。"
            )

            return {
                "errors": [message],
                "route_warnings": [message],
                "final_answer": message,
            }

        legs_by_hotel: dict[
            str,
            list[RouteLeg],
        ] = {}

        subway_distance_by_hotel: dict[
            str,
            float | None,
        ] = {}

        distance_call_count = 0
        around_search_call_count = 0

        for hotel_candidate in hotel_candidates:
            hotel = hotel_candidate.hotel

            hotel_id = hotel_candidate.hotel_id

            hotel_name = str(
                hotel.get("name")
                or hotel_id
            )

            hotel_location = (
                get_hotel_location(hotel)
            )

            hotel_legs: list[RouteLeg] = []

            if hotel_location is None:
                warnings.append(
                    f"{hotel_name}缺少经纬度，"
                    "无法执行真实距离查询。"
                )

                legs_by_hotel[hotel_id] = []
                subway_distance_by_hotel[
                    hotel_id
                ] = None

                continue

            for attraction in attractions:
                attraction_location = (
                    attraction.location
                )

                if attraction_location is None:
                    continue

                try:
                    distance_call_count += 1

                    tool_result = (
                        await distance_tool.ainvoke(
                            {
                                "origins": (
                                    hotel_location
                                ),
                                "destination": (
                                    attraction_location
                                ),
                                # 1代表驾车距离。
                                "type": "1",
                            }
                        )
                    )

                    payload = extract_mcp_payload(
                        tool_result
                    )

                    distances = (
                        find_numeric_values(
                            payload,
                            "distance",
                        )
                    )

                    durations = (
                        find_numeric_values(
                            payload,
                            "duration",
                        )
                    )

                    if not distances:
                        raise RuntimeError(
                            "距离结果中没有distance字段"
                        )

                    hotel_legs.append(
                        RouteLeg(
                            origin_name=hotel_name,
                            origin_location=(
                                hotel_location
                            ),
                            destination_name=(
                                attraction.name
                            ),
                            destination_location=(
                                attraction_location
                            ),
                            distance_meters=(
                                distances[0]
                            ),
                            duration_seconds=(
                                durations[0]
                                if durations
                                else None
                            ),
                            source="amap",
                        )
                    )

                except Exception as exc:
                    warnings.append(
                        f"{hotel_name} → "
                        f"{attraction.name}："
                        "高德距离查询失败，"
                        "已使用直线距离降级。"
                    )

                    try:
                        fallback_leg = (
                            route_service
                            .build_haversine_leg(
                                origin_name=(
                                    hotel_name
                                ),
                                origin_location=(
                                    hotel_location
                                ),
                                destination_name=(
                                    attraction.name
                                ),
                                destination_location=(
                                    attraction_location
                                ),
                            )
                        )

                        hotel_legs.append(
                            fallback_leg
                        )

                    except Exception as fallback_exc:
                        warnings.append(
                            "直线距离降级也失败："
                            f"{type(fallback_exc).__name__}: "
                            f"{fallback_exc}"
                        )

            legs_by_hotel[
                hotel_id
            ] = hotel_legs

            subway_distance: float | None = None

            if (
                request
                .hotel_preferences
                .near_subway
                and around_search_tool
                is not None
            ):
                try:
                    around_search_call_count += 1

                    tool_result = (
                        await around_search_tool.ainvoke(
                            {
                                "location": (
                                    hotel_location
                                ),
                                "radius": "1500",
                                "keywords": "地铁站",
                            }
                        )
                    )

                    payload = extract_mcp_payload(
                        tool_result
                    )

                    subway_distances = (
                        find_numeric_values(
                            payload,
                            "distance",
                        )
                    )

                    if subway_distances:
                        subway_distance = min(
                            subway_distances
                        )

                    else:
                        warnings.append(
                            f"{hotel_name}周边搜索"
                            "没有返回可解析的地铁距离。"
                        )

                except Exception as exc:
                    warnings.append(
                        f"{hotel_name}周边地铁查询失败："
                        f"{type(exc).__name__}: {exc}"
                    )

            subway_distance_by_hotel[
                hotel_id
            ] = subway_distance

        evaluated_hotels = (
            route_service.rank_hotels(
                hotel_candidates,
                legs_by_hotel,
                subway_distance_by_hotel,
                require_near_subway=bool(
                    request
                    .hotel_preferences
                    .near_subway
                ),
                limit=max_hotels,
            )
        )

        selected_hotel_id = (
            evaluated_hotels[0].hotel_id
            if evaluated_hotels
            else None
        )

        result = RouteEvaluationResult(
            evaluated_hotels=evaluated_hotels,
            reference_attractions=[
                attraction.name
                for attraction in attractions
            ],
            selected_hotel_id=selected_hotel_id,
            distance_call_count=(
                distance_call_count
            ),
            around_search_call_count=(
                around_search_call_count
            ),
            warnings=warnings,
        )

        if not evaluated_hotels:
            message = (
                "路线评估完成，但没有得到"
                "有效酒店排序结果。"
            )

            return {
                "route_evaluation_result": result,
                "route_warnings": warnings,
                "errors": [message],
                "final_answer": message,
            }

        selected_hotel = evaluated_hotels[0]

        return {
            "route_evaluation_result": result,
            "route_warnings": warnings,
            "final_answer": (
                "酒店路线评估完成："
                f"评估{len(evaluated_hotels)}家酒店，"
                f"参考{len(attractions)}个景点；"
                "当前综合排名第一为"
                f"“{selected_hotel.hotel_name}”，"
                f"综合得分"
                f"{selected_hotel.final_score:.2f}。"
            ),
        }

    return route_evaluation_node