diff --git a/backend/app/tools/__init__.py b/backend/app/tools/__init__.py new file mode 100644 index 0000000..43e5fc6 --- /dev/null +++ b/backend/app/tools/__init__.py @@ -0,0 +1,13 @@ +"""Agent tool functions with deterministic, testable behavior.""" + +from app.tools.route_sort_day import normalize_stop_role, route_sort_day +from app.tools.schemas import RouteSortResult, RouteSortWarning, RouteStop, ToolPOI + +__all__ = [ + "normalize_stop_role", + "RouteSortResult", + "RouteSortWarning", + "RouteStop", + "ToolPOI", + "route_sort_day", +] diff --git a/backend/app/tools/route_sort_day.py b/backend/app/tools/route_sort_day.py new file mode 100644 index 0000000..cfdc237 --- /dev/null +++ b/backend/app/tools/route_sort_day.py @@ -0,0 +1,211 @@ +import math +from collections.abc import Sequence + +from app.tools.schemas import RouteSortResult, RouteSortWarning, RouteStop, StopRole + + +def _has_coord(stop: RouteStop) -> bool: + poi = stop.poi + return poi.lng is not None and poi.lat is not None + + +def _haversine_m(a_lng: float, a_lat: float, b_lng: float, b_lat: float) -> int: + radius = 6371000.0 + p1, p2 = math.radians(a_lat), math.radians(b_lat) + dphi = math.radians(b_lat - a_lat) + dlmb = math.radians(b_lng - a_lng) + h = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2 + return int(2 * radius * math.asin(math.sqrt(h))) + + +_ROLE_ALIASES: dict[str, StopRole] = { + "breakfast": "breakfast", + "morning_meal": "breakfast", + "早餐": "breakfast", + "早饭": "breakfast", + "早茶": "breakfast", + "lunch": "lunch", + "noon_meal": "lunch", + "午餐": "lunch", + "午饭": "lunch", + "dinner": "dinner", + "evening_meal": "dinner", + "晚餐": "dinner", + "晚饭": "dinner", + "attraction": "attraction", + "sight": "attraction", + "sightseeing": "attraction", + "poi": "attraction", + "play": "attraction", + "景点": "attraction", + "游玩": "attraction", + "体验": "attraction", + "hotel": "hotel", + "lodging": "hotel", + "stay": "hotel", + "住宿": "hotel", + "酒店": "hotel", +} + + +def normalize_stop_role(stop: RouteStop) -> tuple[StopRole, RouteSortWarning | None]: + """Normalize Chinese/English stop slots into canonical route roles. + + Example: + >>> stop = RouteStop(slot="景点", poi=ToolPOI(name="宽窄巷子", category="play")) + >>> normalize_stop_role(stop)[0] + 'attraction' + """ + slot_key = (stop.slot or "").strip().lower() + category = stop.poi.category + if slot_key in _ROLE_ALIASES: + return _ROLE_ALIASES[slot_key], None + + if category == "play": + return "attraction", RouteSortWarning( + code="category_role_fallback", + severity="info", + stop_name=stop.poi.name, + slot=stop.slot, + category=category, + message="Unrecognized slot; inferred attraction role from POI category.", + ) + if category == "stay": + return "hotel", RouteSortWarning( + code="category_role_fallback", + severity="info", + stop_name=stop.poi.name, + slot=stop.slot, + category=category, + message="Unrecognized slot; inferred hotel role from POI category.", + ) + if category == "eat": + return "unknown", RouteSortWarning( + code="ambiguous_meal_role", + stop_name=stop.poi.name, + slot=stop.slot, + category=category, + message="Meal category does not identify breakfast, lunch, or dinner.", + ) + + return "unknown", RouteSortWarning( + code="unknown_role", + severity="error", + stop_name=stop.poi.name, + slot=stop.slot, + category=category, + message="Could not infer a route sorting role.", + ) + + +def _nearest_attractions( + attractions: Sequence[RouteStop], + start_lng: float | None, + start_lat: float | None, +) -> list[RouteStop]: + with_coords = [ + (stop, float(stop.poi.lng), float(stop.poi.lat)) + for stop in attractions + if _has_coord(stop) + ] + without_coords = [stop for stop in attractions if not _has_coord(stop)] + if len(with_coords) <= 1: + return [item[0] for item in with_coords] + without_coords + + remaining = with_coords[:] + ordered: list[RouteStop] = [] + + if start_lng is not None and start_lat is not None: + current_lng, current_lat = start_lng, start_lat + else: + first, current_lng, current_lat = remaining.pop(0) + ordered.append(first) + + while remaining: + next_stop, next_lng, next_lat = min( + remaining, + key=lambda item: _haversine_m( + current_lng, + current_lat, + item[1], + item[2], + ), + ) + ordered.append(next_stop) + remaining.remove((next_stop, next_lng, next_lat)) + current_lng, current_lat = next_lng, next_lat + + return ordered + without_coords + + +def _split_attractions( + attractions: list[RouteStop], + has_lunch: bool, + has_evening_anchor: bool, +) -> tuple[list[RouteStop], list[RouteStop]]: + if not has_lunch or not has_evening_anchor: + return attractions, [] + half = math.ceil(len(attractions) / 2) + return attractions[:half], attractions[half:] + + +def route_sort_day( + stops: Sequence[RouteStop], + start_lng: float | None = None, + start_lat: float | None = None, +) -> RouteSortResult: + """Return a deterministic day timeline sorted by slot semantics and distance. + + Input slots may use canonical English values or common Chinese aliases. + The algorithm works on normalized roles and returns warnings for fallback + decisions that an agent harness may want to inspect. It does not assign + meal times; callers should provide breakfast/lunch/dinner roles explicitly. + """ + original = list(stops) + warnings: list[RouteSortWarning] = [] + grouped: dict[StopRole, list[RouteStop]] = { + "breakfast": [], + "lunch": [], + "dinner": [], + "attraction": [], + "hotel": [], + "unknown": [], + } + for stop in original: + role, warning = normalize_stop_role(stop) + grouped[role].append(stop) + if warning is not None: + warnings.append(warning) + if not _has_coord(stop): + warnings.append( + RouteSortWarning( + code="missing_coordinates", + stop_name=stop.poi.name, + slot=stop.slot, + category=stop.poi.category, + message="Stop has no complete coordinate pair.", + ) + ) + + attractions = grouped["attraction"] + sorted_attractions = _nearest_attractions(attractions, start_lng, start_lat) + morning, afternoon = _split_attractions( + sorted_attractions, + has_lunch=bool(grouped["lunch"]), + has_evening_anchor=bool(grouped["dinner"] or grouped["hotel"]), + ) + + ordered: list[RouteStop] = [] + ordered.extend(grouped["breakfast"]) + ordered.extend(morning) + ordered.extend(grouped["lunch"]) + ordered.extend(afternoon) + ordered.extend(grouped["dinner"]) + ordered.extend(grouped["hotel"]) + ordered.extend(grouped["unknown"]) + + return RouteSortResult( + stops=ordered, + degraded=any(w.severity in ("warning", "error") for w in warnings), + warnings=warnings, + ) diff --git a/backend/app/tools/schemas.py b/backend/app/tools/schemas.py new file mode 100644 index 0000000..6556141 --- /dev/null +++ b/backend/app/tools/schemas.py @@ -0,0 +1,44 @@ +from typing import Literal + +from pydantic import BaseModel, Field + +StopRole = Literal["breakfast", "lunch", "dinner", "attraction", "hotel", "unknown"] +ToolPOICategory = Literal["eat", "play", "stay", "other"] +WarningSeverity = Literal["info", "warning", "error"] +RouteSortWarningCode = Literal[ + "missing_coordinates", + "unknown_role", + "category_role_fallback", + "ambiguous_meal_role", +] + + +class ToolPOI(BaseModel): + name: str + category: ToolPOICategory + lng: float | None = None + lat: float | None = None + address: str | None = None + amap_id: str | None = None + + +class RouteStop(BaseModel): + slot: str = "" + poi: ToolPOI + arrive_time: str | None = None + stay_minutes: int | None = Field(default=None, ge=1) + + +class RouteSortWarning(BaseModel): + code: RouteSortWarningCode + severity: WarningSeverity = "warning" + stop_name: str + slot: str = "" + category: str | None = None + message: str = "" + + +class RouteSortResult(BaseModel): + stops: list[RouteStop] + degraded: bool = False + warnings: list[RouteSortWarning] = Field(default_factory=list) diff --git a/backend/tests/test_route_sort_day_tool.py b/backend/tests/test_route_sort_day_tool.py new file mode 100644 index 0000000..cffe40e --- /dev/null +++ b/backend/tests/test_route_sort_day_tool.py @@ -0,0 +1,155 @@ +from app.tools.schemas import RouteStop, ToolPOI +from app.tools.route_sort_day import normalize_stop_role, route_sort_day + + +def _stop(name: str, slot: str, category: str, lng=None, lat=None) -> RouteStop: + return RouteStop( + slot=slot, + poi=ToolPOI(name=name, category=category, lng=lng, lat=lat), + ) + + +def test_route_sort_day_keeps_timeline_slots_and_sorts_attractions_by_distance(): + stops = [ + _stop("酒店", "hotel", "stay", 104.08, 30.65), + _stop("午餐", "lunch", "eat", 104.06, 30.65), + _stop("远景点", "attraction", "play", 104.20, 30.65), + _stop("早餐", "breakfast", "eat", 104.00, 30.65), + _stop("近景点", "attraction", "play", 104.02, 30.65), + _stop("中景点", "attraction", "play", 104.04, 30.65), + _stop("晚餐", "dinner", "eat", 104.07, 30.65), + ] + + ordered = route_sort_day(stops, start_lng=104.0, start_lat=30.65) + + assert ordered.degraded is False + assert ordered.warnings == [] + assert [s.poi.name for s in ordered.stops] == [ + "早餐", + "近景点", + "中景点", + "午餐", + "远景点", + "晚餐", + "酒店", + ] + + +def test_route_sort_day_puts_missing_coordinates_after_sortable_attractions(): + stops = [ + _stop("早餐", "breakfast", "eat", 104.00, 30.65), + _stop("未知坐标景点", "attraction", "play"), + _stop("远景点", "attraction", "play", 104.20, 30.65), + _stop("近景点", "attraction", "play", 104.02, 30.65), + _stop("午餐", "lunch", "eat", 104.06, 30.65), + ] + + ordered = route_sort_day(stops, start_lng=104.0, start_lat=30.65) + + assert ordered.degraded is True + assert [(w.code, w.severity) for w in ordered.warnings] == [ + ("missing_coordinates", "warning") + ] + assert [s.poi.name for s in ordered.stops] == [ + "早餐", + "近景点", + "远景点", + "未知坐标景点", + "午餐", + ] + + +def test_route_sort_day_does_not_mutate_input_stops(): + stops = [ + _stop("午餐", "lunch", "eat", 104.06, 30.65), + _stop("近景点", "attraction", "play", 104.02, 30.65), + _stop("早餐", "breakfast", "eat", 104.00, 30.65), + ] + + route_sort_day(stops, start_lng=104.0, start_lat=30.65) + + assert [s.poi.name for s in stops] == ["午餐", "近景点", "早餐"] + + +def test_route_sort_day_accepts_chinese_slots_and_preserves_original_values(): + stops = [ + _stop("酒店", "住宿", "stay", 104.08, 30.65), + _stop("午餐", "午餐", "eat", 104.06, 30.65), + _stop("远景点", "景点", "play", 104.20, 30.65), + _stop("早餐", "早餐", "eat", 104.00, 30.65), + _stop("近景点", "游玩", "play", 104.02, 30.65), + _stop("晚餐", "晚饭", "eat", 104.07, 30.65), + ] + + ordered = route_sort_day(stops, start_lng=104.0, start_lat=30.65) + + assert ordered.degraded is False + assert [s.poi.name for s in ordered.stops] == [ + "早餐", + "近景点", + "午餐", + "远景点", + "晚餐", + "酒店", + ] + assert ordered.stops[0].slot == "早餐" + + +def test_route_sort_day_falls_back_from_category_without_guessing_meal_slots(): + stops = [ + _stop("空 slot 景点", "", "play", 104.02, 30.65), + _stop("空 slot 酒店", "", "stay", 104.08, 30.65), + _stop("空 slot 餐厅", "", "eat", 104.06, 30.65), + ] + + ordered = route_sort_day(stops, start_lng=104.0, start_lat=30.65) + + assert ordered.degraded is True + assert [s.poi.name for s in ordered.stops] == [ + "空 slot 景点", + "空 slot 酒店", + "空 slot 餐厅", + ] + assert [(w.code, w.severity) for w in ordered.warnings] == [ + ("category_role_fallback", "info"), + ("category_role_fallback", "info"), + ("ambiguous_meal_role", "warning"), + ] + + +def test_route_sort_day_warns_when_inferring_role_from_unknown_slot(): + stops = [ + _stop("早餐", "breakfast", "eat", 104.00, 30.65), + _stop("未知活动", "自由活动", "play", 104.03, 30.65), + _stop("景点", "attraction", "play", 104.02, 30.65), + ] + + ordered = route_sort_day(stops, start_lng=104.0, start_lat=30.65) + + assert ordered.degraded is False + assert [s.poi.name for s in ordered.stops] == ["早餐", "景点", "未知活动"] + assert [(w.code, w.severity, w.stop_name) for w in ordered.warnings] == [ + ("category_role_fallback", "info", "未知活动") + ] + + +def test_normalize_stop_role_is_public_and_returns_warning_severity(): + role, warning = normalize_stop_role(_stop("随便逛", "自由活动", "play")) + + assert role == "attraction" + assert warning is not None + assert warning.code == "category_role_fallback" + assert warning.severity == "info" + + +def test_route_sort_day_marks_unknown_role_as_error(): + unknown = RouteStop( + slot="自由活动", + poi=ToolPOI(name="未知地点", category="other", lng=104.03, lat=30.65), + ) + + ordered = route_sort_day([unknown], start_lng=104.0, start_lat=30.65) + + assert ordered.degraded is True + assert ordered.warnings[0].code == "unknown_role" + assert ordered.warnings[0].severity == "error"