wehub-resource-sync
610 lines
25 KiB
Python
610 lines
25 KiB
Python
"""
|
||
Map task definitions.
|
||
"""
|
||
# -- Task Index (auto-generated, do not edit) --
|
||
# 16 tasks | L1×1 L2×9 L3×5 L4×1
|
||
#
|
||
# [L1] CheckDriveRoute 帮我查一下到{place}的驾车路线
|
||
# [L2] CheckHighestRatedPlace 附近{radius}内评分最高的{category}是哪家，优先告诉我离我最近的
|
||
# [L2] CheckNearestPlaceAddress 离我最近的{category}在什么地址
|
||
# [L3] SetMapNorthUp 把地图设置成始终上北下南
|
||
# [L2] QueryDrivingDistance {place}离这儿开车有多远
|
||
# [L4] CheckRouteSuccess 从{origin}开车去{destination}怎么走，前几步告诉我
|
||
# [L3] FindBestRatedAndRoute 附近{radius}内评分最高且最近的{category}是哪家，开车过去大概多远
|
||
# [L2] ModifyMultiSettings 把地图停车位置通知设为{parking_pref}，并将保存近期搜索设为{save_recent_searches}
|
||
# [L2] DarkModeSettings 把地图主题设为{theme}
|
||
# [L3] FindNearestWithRating 最近的{category}叫什么、评分多少
|
||
# [L2] CompareRouteDuration 查一下去{place}步行和开车哪个更快，各要多久
|
||
# [L2] FindNearestAndRoute 帮我找最近的{category}，看看开车过去怎么走
|
||
# [L2] EstimateDrivingCost 帮我算一下开车去{place}的油费，按每公里{rate}元算
|
||
# [L2] NearestInRadiusRatingRank 最近的有评分的{category}在附近{radius}内同类评分里排第几
|
||
# [L3] BestRatedWithWalkRoute 帮我找附近{radius}内的{category}里评分最高且最近的，看看走过去多远
|
||
# [L3] NearestDetailAndWalkRoute 最近的有评分的{category}叫什么、评分多少，走过去要多久
|
||
# -- End Task Index --
|
||
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from bench_env.task.base import BaseTask
|
||
from bench_env.task.common_tasks import AnswerTask, CriteriaTask
|
||
from bench_env.task.judge import JudgeInput
|
||
from bench_env.task.map.app import (
|
||
CATEGORY_PARAM,
|
||
DRIVING_OD_PAIRS,
|
||
MAP_SEARCH_CHANGES,
|
||
PLACE_PARAM,
|
||
RADIUS_PARAM,
|
||
Map,
|
||
)
|
||
from bench_env.task.utils import check_alternatives
|
||
|
||
|
||
# =============================================================================
|
||
# L1 — Atomic queries
|
||
# =============================================================================
|
||
# =============================================================================
|
||
# L2 — Core search and route tasks
|
||
# =============================================================================
|
||
|
||
class CheckDriveRoute(BaseTask):
|
||
templates = ["帮我查一下到{place}的驾车路线"]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "atomic"
|
||
difficulty = "L1"
|
||
capabilities = ["search", "nav"]
|
||
parameters = {"place": PLACE_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
return [
|
||
Map(input.apps["map"]).check_route(
|
||
mode="DRIVING",
|
||
destination_hint=self.p.place,
|
||
)
|
||
]
|
||
|
||
|
||
class CheckHighestRatedPlace(AnswerTask):
|
||
templates = ["附近{radius}内评分最高的{category}是哪家，优先告诉我离我最近的"]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
capabilities = ["search", "extract"]
|
||
parameters = {"category": CATEGORY_PARAM, "radius": RADIUS_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [{"type": "text", "label": "地点名称", "hint": "如：海底捞"}]
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
Map.require_rated_in_radius(self.p.category, self.p.radius)
|
||
|
||
def get_answer(self, input: JudgeInput) -> str:
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
best = Map.best_rated_from_results(results, max_distance_meters=self.p.radius)
|
||
return str(best["name"])
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
return [sc, {"field": "answer.name", "passed": False, "expected": "最高评分地点名称", "actual": "前置搜索未完成"}]
|
||
return [sc, Map.check_answer_match(input.answer, self.get_answer(input), field="answer.name")]
|
||
|
||
|
||
class CheckNearestPlaceAddress(AnswerTask):
|
||
templates = ["离我最近的{category}在什么地址"]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
max_steps = 30
|
||
capabilities = ["search", "extract"]
|
||
parameters = {"category": CATEGORY_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [{"type": "text", "label": "地址", "hint": "如：北京市海淀区学院路28号"}]
|
||
|
||
def get_answer(self, input: JudgeInput) -> str:
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
nearest = Map.nearest_from_results(results)
|
||
return Map.extract_address(nearest)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
return [sc, {"field": "answer.address", "passed": False, "expected": "最近地点地址", "actual": "前置搜索未完成"}]
|
||
return [sc, Map.check_answer_match(input.answer, self.get_answer(input), field="answer.address")]
|
||
|
||
|
||
class SetMapNorthUp(CriteriaTask):
|
||
templates = [
|
||
"把地图设置成始终上北下南",
|
||
"Set the map to always show north at the top",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
max_steps = 30
|
||
capabilities = ["settings"]
|
||
criteria = {
|
||
"settings.navigation.keepMapNorthUp": True,
|
||
}
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
await self._invert_criteria(env)
|
||
class QueryDrivingDistance(AnswerTask):
|
||
templates = ["{place}离这儿开车有多远"]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
capabilities = ["nav", "extract"]
|
||
parameters = {"place": PLACE_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [{"type": "text", "label": "驾车距离", "hint": "如：3.7公里"}]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
routes = Map.resolve_routes_from_current(self.p.place, "DRIVING")
|
||
return check_alternatives(
|
||
[Map.check_geo_distance(input.answer, str(r["distance"]), field="answer.drive_distance") for _, r in routes],
|
||
)
|
||
|
||
|
||
# =============================================================================
|
||
# L3 — Complex single-app tasks
|
||
# =============================================================================
|
||
|
||
class CheckRouteSuccess(BaseTask):
|
||
templates = [
|
||
"从{origin}开车去{destination}怎么走，前几步告诉我",
|
||
"帮我看下从{origin}开车到{destination}怎么走，把前几步路线说一下",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "sequential"
|
||
difficulty = "L4"
|
||
max_steps = 45
|
||
capabilities = ["nav", "extract"]
|
||
parameters = {
|
||
"origin": {
|
||
"type": "string",
|
||
"default": "故宫",
|
||
"description": "出发地（须与离线 routes.json 中已存在的驾车段一致）",
|
||
},
|
||
"destination": {
|
||
"type": "string",
|
||
"default": "天安门广场",
|
||
"description": "目的地（须与离线 routes.json 中已存在的驾车段一致）",
|
||
},
|
||
"_check_route_od": {
|
||
"sampler": Map.sample_driving_od,
|
||
"fields": {"origin": "origin", "destination": "destination"},
|
||
},
|
||
}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [{"type": "text", "label": "前几步路线", "hint": "如：向北走200米，左转进入平安大道"}]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"])
|
||
oh, dh = self.p.origin, self.p.destination
|
||
if (oh, dh) not in DRIVING_OD_PAIRS:
|
||
raise ValueError(
|
||
"CheckRouteSuccess: unknown origin/destination pair for offline route: "
|
||
f"{oh!r} -> {dh!r}"
|
||
)
|
||
# 从离线 routes.json 读取期望步骤作为 ground truth，
|
||
# 避免从 active_route（Agent 行为结果）读取导致自引用。
|
||
triples = Map.resolve_route_pairs(oh, dh, "DRIVING")
|
||
return check_alternatives(
|
||
[m.check_route(mode="DRIVING", origin_hint=oh, destination_hint=dh, field="route_generated") for _ in triples],
|
||
[Map.check_geo_steps(input.answer, Map.route_step_texts_from_api_route(route), max_steps=3) for _, _, route in triples],
|
||
)
|
||
|
||
|
||
class FindBestRatedAndRoute(BaseTask):
|
||
templates = [
|
||
"附近{radius}内评分最高且最近的{category}是哪家，开车过去大概多远",
|
||
"帮我找一下{radius}内评分最高且最近的{category}，再看看开车过去有多远",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
capabilities = ["search", "nav", "extract"]
|
||
parameters = {"category": CATEGORY_PARAM, "radius": RADIUS_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [
|
||
{"type": "text", "label": "地点名称", "hint": "如：便利店"},
|
||
{"type": "text", "label": "驾车距离", "hint": "如：2.8公里"},
|
||
]
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
Map.require_rated_in_radius(self.p.category, self.p.radius)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
_skip = "前置搜索未完成"
|
||
return [
|
||
sc,
|
||
{"field": "route_to_best_rated", "passed": False, "expected": "驾车路线", "actual": _skip},
|
||
{"field": "answer.name", "passed": False, "expected": "最高评分地点名称", "actual": _skip},
|
||
{"field": "answer.distance", "passed": False, "expected": "驾车距离", "actual": _skip},
|
||
]
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
best = Map.best_rated_from_results(results, max_distance_meters=self.p.radius)
|
||
best_name = str(best["name"])
|
||
route = Map.geo_route_from_current(str(best["place_id"]), "DRIVING")
|
||
answer = str(input.answer or "")
|
||
return [
|
||
sc,
|
||
m.check_route(
|
||
mode="DRIVING",
|
||
destination_hint=best_name,
|
||
field="route_to_best_rated",
|
||
),
|
||
Map.check_answer_match(answer, best_name, field="answer.name"),
|
||
Map.check_geo_distance(answer, str(route["distance"]), field="answer.distance"),
|
||
]
|
||
|
||
|
||
class ModifyMultiSettings(CriteriaTask):
|
||
templates = ["把地图停车位置通知设为{parking_pref}，并将保存近期搜索设为{save_recent_searches}"]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
max_steps = 45
|
||
capabilities = ["settings"]
|
||
parameters = {
|
||
"parking_pref": {
|
||
"type": "enum",
|
||
"values": {"开启": "开启", "关闭": "关闭", "仅限应用": "仅限应用"},
|
||
"default": "仅限应用",
|
||
"description": "停车位置通知偏好",
|
||
},
|
||
"save_recent_searches": {
|
||
"type": "bool",
|
||
"values": {"开启": True, "关闭": False},
|
||
"default": False,
|
||
"description": "是否保存近期搜索",
|
||
},
|
||
}
|
||
criteria = {
|
||
"settings.notifications.traffic.parkingLocation": "{parking_pref}",
|
||
"settings.locationPrivacy.saveRecentSearches": "{save_recent_searches}",
|
||
}
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
await self._invert_criteria(env)
|
||
|
||
|
||
class DarkModeSettings(CriteriaTask):
|
||
templates = ["把地图主题设为{theme}"]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
max_steps = 15
|
||
capabilities = ["settings"]
|
||
parameters = {
|
||
"theme": {
|
||
"type": "enum",
|
||
"values": {
|
||
"浅色主题": "始终采用浅色主题",
|
||
"深色主题": "始终采用深色主题",
|
||
"跟随设备": "与设备主题背景一致",
|
||
},
|
||
"default": "始终采用深色主题",
|
||
"description": "地图主题",
|
||
},
|
||
}
|
||
criteria = {
|
||
"settings.appDisplay.theme": "{theme}",
|
||
}
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
await self._invert_criteria(env)
|
||
|
||
|
||
class FindNearestWithRating(AnswerTask):
|
||
templates = [
|
||
"最近的{category}叫什么、评分多少",
|
||
"帮我看一下最近的{category}名字和评分",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
capabilities = ["search", "extract"]
|
||
parameters = {"category": CATEGORY_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [
|
||
{"type": "text", "label": "地点名称", "hint": "如：全聚德"},
|
||
{"type": "text", "label": "评分", "hint": "如：4.2"},
|
||
]
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
Map.require_nearest_has_rating(self.p.category)
|
||
|
||
def get_answer(self, input: JudgeInput) -> dict[str, Any]:
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
nearest = Map.nearest_rated_from_results(results)
|
||
return {"name": str(nearest["name"]), "rating": Map.extract_rating(nearest)}
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
_skip = "前置搜索未完成"
|
||
return [
|
||
sc,
|
||
{"field": "answer.name", "passed": False, "expected": "最近地点名称", "actual": _skip},
|
||
{"field": "answer.rating", "passed": False, "expected": "最近地点评分", "actual": _skip},
|
||
]
|
||
expected = self.get_answer(input)
|
||
answer = str(input.answer or "")
|
||
return [
|
||
sc,
|
||
Map.check_answer_match(answer, expected["name"], field="answer.name"),
|
||
Map.check_rating_by_name(answer, expected["name"], expected["rating"]),
|
||
]
|
||
|
||
|
||
class CompareRouteDuration(AnswerTask):
|
||
templates = [
|
||
"查一下去{place}步行和开车哪个更快，各要多久",
|
||
"去{place}的话，步行和开车哪个花的时间更短，两种时间分别多少",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
capabilities = ["extract", "reasoning"]
|
||
parameters = {"place": PLACE_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [
|
||
{"type": "choice", "label": "更快的方式", "options": ["步行更快", "开车更快", "一样快"]},
|
||
{"type": "text", "label": "步行时长", "hint": "如：18分钟"},
|
||
{"type": "text", "label": "驾车时长", "hint": "如：12分钟"},
|
||
]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
answer = str(input.answer or "")
|
||
places = Map.resolve_places(self.p.place)
|
||
routes = [(Map.geo_route_from_current(str(p["place_id"]), "WALKING"), Map.geo_route_from_current(str(p["place_id"]), "DRIVING")) for p in places]
|
||
return check_alternatives(
|
||
[Map.check_route_faster_mode_answer(answer, walking_seconds=float(w["duration_seconds"]), driving_seconds=float(d["duration_seconds"])) for w, d in routes],
|
||
[Map.check_geo_duration(answer, str(w["duration"]), field="answer.walk_duration") for w, _ in routes],
|
||
[Map.check_geo_duration(answer, str(d["duration"]), field="answer.drive_duration") for _, d in routes],
|
||
)
|
||
class FindNearestAndRoute(BaseTask):
|
||
templates = ["帮我找最近的{category}，看看开车过去怎么走"]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
max_steps = 45
|
||
capabilities = ["search", "nav"]
|
||
parameters = {"category": CATEGORY_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
return [sc, {"field": "route_to_nearest", "passed": False, "expected": "驾车路线", "actual": "前置搜索未完成"}]
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
nearest_name = str(Map.nearest_from_results(results)["name"])
|
||
return [
|
||
sc,
|
||
m.check_route(
|
||
mode="DRIVING",
|
||
destination_hint=nearest_name,
|
||
field="route_to_nearest",
|
||
),
|
||
]
|
||
|
||
|
||
class EstimateDrivingCost(AnswerTask):
|
||
"""估算开车去目标地点的油费（查路线距离 × 费率）。
|
||
|
||
判定：使用 Map.check_driving_cost_answer 验证 Agent 回答中
|
||
包含正确的费用数值（距离来自离线路线数据，容忍 10% 相对误差）。
|
||
"""
|
||
|
||
templates = [
|
||
"帮我算一下开车去{place}的油费，按每公里{rate}元算",
|
||
"开车去{place}大概多少油钱，按每公里{rate}元估算一下",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
max_steps = 45
|
||
capabilities = ["nav", "extract", "reasoning"]
|
||
parameters = {
|
||
"place": PLACE_PARAM,
|
||
"rate": {
|
||
"type": "enum",
|
||
"values": {"0.5元": 0.5, "0.8元": 0.8, "1元": 1.0},
|
||
"default": 0.8,
|
||
"description": "每公里油费（元）",
|
||
},
|
||
}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [{"type": "number", "label": "油费（元）"}]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
return [Map.check_driving_cost_answer(
|
||
input.answer, self.p.place, float(self.p.rate),
|
||
)]
|
||
|
||
|
||
# =============================================================================
|
||
# L4 — Deep dive tasks
|
||
# =============================================================================
|
||
|
||
class NearestInRadiusRatingRank(AnswerTask):
|
||
templates = [
|
||
"最近的有评分的{category}在附近{radius}内同类评分里排第几",
|
||
"帮我查一下最近的有评分的{category}在附近{radius}范围内同类评分排名第几位",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "deep_dive"
|
||
difficulty = "L3"
|
||
max_steps = 60
|
||
capabilities = ["search", "extract", "reasoning"]
|
||
parameters = {
|
||
"category": CATEGORY_PARAM,
|
||
"radius": {
|
||
"type": "enum",
|
||
"values": {"2公里": 2000, "3公里": 3000},
|
||
"default": 2000,
|
||
"description": "搜索半径（米）",
|
||
},
|
||
}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [{"type": "number", "label": "评分排名"}]
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
Map.require_rated_in_radius(self.p.category, self.p.radius, min_results=2)
|
||
|
||
def get_answer(self, input: JudgeInput) -> int:
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
in_radius = Map.filter_results(
|
||
results,
|
||
max_distance_meters=self.p.radius,
|
||
)
|
||
nearest = Map.nearest_rated_from_results(in_radius)
|
||
return Map.rating_rank_from_results(
|
||
in_radius,
|
||
str(nearest["name"]),
|
||
min_results=2,
|
||
place_id=str(nearest["place_id"]),
|
||
)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
return [sc, {"field": "answer.rank", "passed": False, "expected": "评分排名", "actual": "前置搜索未完成"}]
|
||
return [sc, Map.check_answer_match(input.answer, self.get_answer(input), field="answer.rank")]
|
||
|
||
|
||
|
||
class BestRatedWithWalkRoute(BaseTask):
|
||
templates = [
|
||
"帮我找附近{radius}内的{category}里评分最高且最近的，看看走过去多远",
|
||
"附近{radius}内{category}里评分最高且最近的是哪家，步行过去大概多远",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "deep_dive"
|
||
difficulty = "L3"
|
||
max_steps = 60
|
||
capabilities = ["search", "nav", "extract"]
|
||
parameters = {"category": CATEGORY_PARAM, "radius": RADIUS_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [
|
||
{"type": "text", "label": "地点名称", "hint": "如：麦当劳"},
|
||
{"type": "text", "label": "步行距离", "hint": "如：800米"},
|
||
]
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
Map.require_rated_in_radius(self.p.category, self.p.radius)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
_skip = "前置搜索未完成"
|
||
return [
|
||
sc,
|
||
{"field": "walk_route_to_best_rated", "passed": False, "expected": "步行路线", "actual": _skip},
|
||
{"field": "answer.name", "passed": False, "expected": "最高评分地点名称", "actual": _skip},
|
||
{"field": "answer.walk_distance", "passed": False, "expected": "步行距离", "actual": _skip},
|
||
]
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
best = Map.best_rated_from_results(results, max_distance_meters=self.p.radius)
|
||
best_name = str(best["name"])
|
||
route = Map.geo_route_from_current(str(best["place_id"]), "WALKING")
|
||
answer = str(input.answer or "")
|
||
return [
|
||
sc,
|
||
m.check_route(
|
||
mode="WALKING",
|
||
destination_hint=best_name,
|
||
field="walk_route_to_best_rated",
|
||
),
|
||
Map.check_answer_match(answer, best_name, field="answer.name"),
|
||
Map.check_geo_distance(answer, str(route["distance"]), field="answer.walk_distance"),
|
||
]
|
||
class NearestDetailAndWalkRoute(BaseTask):
|
||
templates = [
|
||
"最近的有评分的{category}叫什么、评分多少，走过去要多久",
|
||
"帮我看下最近的有评分的{category}名字和评分，再看看步行多久能到",
|
||
]
|
||
apps = ["map"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "deep_dive"
|
||
difficulty = "L3"
|
||
max_steps = 60
|
||
capabilities = ["search", "nav", "extract"]
|
||
parameters = {"category": CATEGORY_PARAM}
|
||
expected_changes = MAP_SEARCH_CHANGES
|
||
answer_fields = [
|
||
{"type": "text", "label": "地点名称", "hint": "如：海底捞"},
|
||
{"type": "text", "label": "评分", "hint": "如：4.5"},
|
||
{"type": "text", "label": "步行时长", "hint": "如：9分钟", "matcher": "duration"},
|
||
]
|
||
|
||
async def _post_sample(self, env: Any) -> None:
|
||
Map.require_nearest_has_rating(self.p.category)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
m = Map(input.apps["map"], init=input.apps_init["map"])
|
||
sc = m.check_searched(category=self.p.category)
|
||
if not sc["passed"]:
|
||
_skip = "前置搜索未完成"
|
||
return [
|
||
sc,
|
||
{"field": "answer.name", "passed": False, "expected": "最近有评分地点名称", "actual": _skip},
|
||
{"field": "answer.rating", "passed": False, "expected": "最近有评分地点评分", "actual": _skip},
|
||
{"field": "answer.walk_duration", "passed": False, "expected": "步行时长", "actual": _skip},
|
||
]
|
||
results = Map.geo_search(self.p.category, limit=0)
|
||
nearest = Map.nearest_rated_from_results(results)
|
||
nearest_name = str(nearest["name"])
|
||
nearest_rating = Map.extract_rating(nearest)
|
||
route = Map.geo_route_from_current(str(nearest["place_id"]), "WALKING")
|
||
answer = str(input.answer or "")
|
||
return [
|
||
sc,
|
||
Map.check_answer_match(answer, nearest_name, field="answer.name"),
|
||
Map.check_rating_by_name(answer, nearest_name, nearest_rating),
|
||
Map.check_geo_duration(answer, str(route["duration"]), field="answer.walk_duration"),
|
||
]
|