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