purewhiter--mobilegym
wehub-resource-sync
424 行
19 KiB
Python
424 行
19 KiB
Python
"""
|
||
eBay app task definitions.
|
||
"""
|
||
# -- Task Index (auto-generated, do not edit) --
|
||
# 8 tasks | L1×1 L2×2 L3×4 L4×1
|
||
#
|
||
# [L1] SwitchTheme 把 eBay 的主题切换成{theme}。
|
||
# [L2] SortSearchResults 在 eBay 搜索「{query}」，按{sort}排序。
|
||
# [L2] SearchFirstResult 在 eBay 搜索「{query}」，告诉我第一个商品{metric}。
|
||
# [L3] CountSonyHeadphonesEurope 帮我看看 eBay 上{location}发货的{condition}{brand}{query}，有多少个。
|
||
# [L3] CountNikeSneakersInRange eBay 上{location}发货的{brand}{query}，要{condition}的，{price_min} 到 {price_max} 块之间的有多少个？
|
||
# [L4] FindCheapestProduct 我想买个{location}发货的{brand}{query}，要{condition}的，最便宜的是哪一个，算上运费多少钱？
|
||
# [L3] CompareTwoProductPrices 帮我在 eBay 上分别搜亚洲发货的{item1}和{item2}，要全新的，看看各自{extreme}的算上运费多少钱，哪个{comparison}？
|
||
# [L3] CompareTwoGroupCounts 帮我比较两组筛选结果：{location1}发货的{condition1} {brand1} {query1}里，{price_min1} 到 {price_max1} 块的；以及{location2}发货的{condition2} {brand2} {query2}里，{price_min2} 到 {price_max2} 块的。哪个选择更多，各有多少个？
|
||
# -- 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, build_answer_checks
|
||
from bench_env.task.ebay.app import (
|
||
EBAY_CATEGORY_VALUES,
|
||
EBAY_SEARCH_QUERY_PARAM,
|
||
EBAY_SORT_PARAM,
|
||
EBAY_THEME_PARAM,
|
||
Ebay,
|
||
build_compare_counts_checks,
|
||
build_compare_two_totals_checks,
|
||
expect_count,
|
||
expect_top,
|
||
)
|
||
from bench_env.task.judge import JudgeInput
|
||
|
||
|
||
# =============================================================================
|
||
# L1 — Atomic navigation & simple settings
|
||
# =============================================================================
|
||
class SwitchTheme(CriteriaTask):
|
||
templates = [
|
||
"把 eBay 的主题切换成{theme}。",
|
||
"帮我把 eBay 设成{theme}主题。",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "atomic"
|
||
difficulty = "L1"
|
||
capabilities = ["settings"]
|
||
parameters = {"theme": EBAY_THEME_PARAM}
|
||
criteria = {"settings.themeId": "{theme}"}
|
||
optimal_paths = [["tab.me", "me.settings.open"]]
|
||
|
||
async def _post_sample(self, env):
|
||
await self._invert_criteria(env)
|
||
|
||
|
||
class SortSearchResults(CriteriaTask):
|
||
templates = [
|
||
"在 eBay 搜索「{query}」，按{sort}排序。",
|
||
"帮我搜一下 eBay 上的「{query}」，结果按{sort}排列。",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
capabilities = ["search"]
|
||
parameters = {"query": EBAY_SEARCH_QUERY_PARAM, "sort": EBAY_SORT_PARAM}
|
||
criteria = {
|
||
"search.current.query": "{query}",
|
||
"search.current.sortOption": "{sort}",
|
||
}
|
||
optimal_paths = [["tab.search"]]
|
||
expected_changes = ["search.current", "search.history", "search.lastCompare", "recentSearches"]
|
||
|
||
|
||
class SearchFirstResult(CriteriaTask):
|
||
templates = [
|
||
"在 eBay 搜索「{query}」，告诉我第一个商品{metric}。",
|
||
"帮我看一下 eBay 搜「{query}」时排在最前面的商品{metric}。",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
capabilities = ["search", "extract"]
|
||
parameters = {
|
||
"query": EBAY_SEARCH_QUERY_PARAM,
|
||
"metric": {
|
||
"type": "enum",
|
||
"values": {"叫什么": "title", "算上运费一共多少钱": "total_cost"},
|
||
"default": "title",
|
||
"description": "查询指标",
|
||
},
|
||
}
|
||
criteria = {"search.current.query": "{query}"}
|
||
optimal_paths = [["tab.search"]]
|
||
expected_changes = ["search.current", "search.history", "search.lastCompare", "recentSearches"]
|
||
@property
|
||
def answer_fields(self): # type: ignore[override]
|
||
metric = getattr(self.p, "metric", None)
|
||
label = next(
|
||
(k for k, v in self.parameters["metric"]["values"].items() if v == metric),
|
||
"商品信息",
|
||
)
|
||
t = "number" if metric == "total_cost" else "text"
|
||
field: dict = {"type": t, "label": label}
|
||
if metric == "title":
|
||
field["hint"] = "请填写商品完整标题"
|
||
elif metric == "total_cost":
|
||
field["hint"] = "填写¥金额数字"
|
||
return [field]
|
||
|
||
def get_answer(self, input: JudgeInput) -> Any:
|
||
product = expect_top(query=self.p.query, sort_id="bestMatch", n=1)[0]
|
||
if self.p.metric == "title":
|
||
return product.title
|
||
return round(product.total_cost, 2)
|
||
|
||
|
||
# =============================================================================
|
||
# L3 — Multi-filter search tasks
|
||
# =============================================================================
|
||
|
||
|
||
class CountSonyHeadphonesEurope(BaseTask):
|
||
templates = [
|
||
"帮我看看 eBay 上{location}发货的{condition}{brand}{query}，有多少个。",
|
||
"eBay 里{location}发货的{condition}{brand}{query}有几个？",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
capabilities = ["search", "extract"]
|
||
parameters = {
|
||
"query": {"type": "string", "default": "耳机", "description": "搜索词"},
|
||
"brand": {"type": "string", "default": "Sony", "description": "品牌"},
|
||
"location": {"type": "string", "default": "欧洲", "description": "发货地"},
|
||
"condition": {"type": "string", "default": "全新", "description": "成色"},
|
||
"_case": {
|
||
"sampler": Ebay.sample_brand_location_case,
|
||
"fields": {"query": "query", "brand": "brand", "location": "location", "condition": "condition"},
|
||
},
|
||
|
||
}
|
||
|
||
expected_changes = ["search.current", "search.history", "search.lastCompare", "recentSearches"]
|
||
optimal_paths = [["tab.search"]]
|
||
answer_fields = [{"type": "number", "label": "商品数量"}]
|
||
|
||
def get_answer(self, input: JudgeInput) -> Any:
|
||
return expect_count(
|
||
query=self.p.query, brand=self.p.brand,
|
||
condition=self.p.condition, location=self.p.location,
|
||
)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
ebay = Ebay(input.apps["ebay"])
|
||
checks: list[dict[str, Any]] = []
|
||
checks.append(ebay.check_has_snapshot(
|
||
query=self.p.query, brand=self.p.brand,
|
||
condition=self.p.condition, location=self.p.location,
|
||
))
|
||
checks.extend(build_answer_checks(self.get_answer(input), input.answer))
|
||
return checks
|
||
|
||
|
||
|
||
class CountNikeSneakersInRange(BaseTask):
|
||
templates = [
|
||
"eBay 上{location}发货的{brand}{query}，要{condition}的，{price_min} 到 {price_max} 块之间的有多少个？",
|
||
"帮我看看{location}发货的{condition}的{brand}{query}里，{price_min} 到 {price_max} 这个价位有多少个。",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
capabilities = ["search", "extract"]
|
||
parameters = {
|
||
"query": {"type": "string", "default": "运动鞋", "description": "搜索词"},
|
||
"brand": {"type": "string", "default": "Nike", "description": "品牌"},
|
||
"location": {"type": "string", "default": "欧洲", "description": "发货地"},
|
||
"condition": {"type": "string", "default": "全新", "description": "成色"},
|
||
"price_min": {"type": "string", "default": "510", "description": "总价下限"},
|
||
"price_max": {"type": "string", "default": "540", "description": "总价上限"},
|
||
"_case": {
|
||
"sampler": Ebay.sample_range_case,
|
||
"fields": {
|
||
"query": "query", "brand": "brand", "location": "location",
|
||
"condition": "condition", "price_min": "price_min", "price_max": "price_max",
|
||
},
|
||
},
|
||
}
|
||
|
||
expected_changes = ["search.current", "search.history", "search.lastCompare", "recentSearches"]
|
||
optimal_paths = [["tab.search"]]
|
||
answer_fields = [{"type": "number", "label": "商品数量"}]
|
||
|
||
def get_answer(self, input: JudgeInput) -> Any:
|
||
return expect_count(
|
||
query=self.p.query, brand=self.p.brand,
|
||
condition=self.p.condition, location=self.p.location,
|
||
min_total=float(self.p.price_min), max_total=float(self.p.price_max),
|
||
)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
ebay = Ebay(input.apps["ebay"])
|
||
checks: list[dict[str, Any]] = []
|
||
checks.append(ebay.check_has_snapshot(
|
||
query=self.p.query, brand=self.p.brand,
|
||
condition=self.p.condition, location=self.p.location,
|
||
price_min=self.p.price_min, price_max=self.p.price_max,
|
||
))
|
||
checks.extend(build_answer_checks(self.get_answer(input), input.answer))
|
||
return checks
|
||
|
||
|
||
class FindCheapestProduct(AnswerTask):
|
||
templates = [
|
||
"我想买个{location}发货的{brand}{query}，要{condition}的，最便宜的是哪一个，算上运费多少钱？",
|
||
"帮我找一下 eBay 上{location}发货、{condition}的{brand}{query}里最便宜的那个，告诉我算上运费多少钱。",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "sequential"
|
||
difficulty = "L4"
|
||
max_steps = 45
|
||
capabilities = ["search", "extract"]
|
||
parameters = {
|
||
"query": {"type": "string", "default": "吸尘器", "description": "搜索词"},
|
||
"brand": {"type": "string", "default": "Dyson", "description": "品牌"},
|
||
"location": {"type": "string", "default": "亚洲", "description": "发货地"},
|
||
"condition": {"type": "string", "default": "全新", "description": "成色"},
|
||
"_case": {
|
||
"sampler": Ebay.sample_brand_location_case,
|
||
"fields": {"query": "query", "brand": "brand", "location": "location", "condition": "condition"},
|
||
},
|
||
}
|
||
expected_changes = ["search.current", "search.history", "search.lastCompare", "recentSearches"]
|
||
optimal_paths = [["tab.search"]]
|
||
answer_fields = [
|
||
{"type": "text", "label": "最便宜商品的标题", "hint": "如：Dyson V15 Detect"},
|
||
{"type": "number", "label": "总价(¥)"},
|
||
]
|
||
|
||
def get_answer(self, input: JudgeInput) -> Any:
|
||
ebay = Ebay(input.apps["ebay"])
|
||
return ebay.cheapest_product_answer(
|
||
query=self.p.query, brand=self.p.brand,
|
||
condition=self.p.condition, location=self.p.location,
|
||
)
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
ebay = Ebay(input.apps["ebay"])
|
||
checks = [ebay.check_has_snapshot(
|
||
query=self.p.query, brand=self.p.brand,
|
||
condition=self.p.condition, location=self.p.location,
|
||
)]
|
||
checks.extend(build_answer_checks(self.get_answer(input), input.answer))
|
||
return checks
|
||
|
||
|
||
# =============================================================================
|
||
# L4 — Deep-dive comparisons
|
||
# =============================================================================
|
||
|
||
|
||
class CompareTwoProductPrices(BaseTask):
|
||
templates = [
|
||
"帮我在 eBay 上分别搜亚洲发货的{item1}和{item2}，要全新的，看看各自{extreme}的算上运费多少钱，哪个{comparison}？",
|
||
"帮我比较一下 eBay 上亚洲发货的全新的{item1}和{item2}，各自{extreme}的算上运费各是多少？哪个{comparison}？",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "deep_dive"
|
||
difficulty = "L3"
|
||
max_steps = 60
|
||
capabilities = ["search", "extract", "reasoning"]
|
||
parameters = {
|
||
"item1": {"type": "string", "default": "电脑", "description": "第一个商品"},
|
||
"item2": {"type": "string", "default": "电视", "description": "第二个商品"},
|
||
"sort_id": {"type": "string", "default": "priceLow", "description": "排序方式"},
|
||
"extreme": {"type": "string", "default": "最便宜", "description": "极值描述"},
|
||
"comparison": {"type": "string", "default": "更便宜", "description": "比较词"},
|
||
"_pair": {
|
||
"sampler": Ebay.sample_compare_pair,
|
||
"fields": {
|
||
"item1": "item1", "item2": "item2",
|
||
"sort_id": "sort_id",
|
||
"extreme": "extreme", "comparison": "comparison",
|
||
},
|
||
},
|
||
}
|
||
optimal_paths = [["tab.search"]]
|
||
expected_changes = ["search.current", "search.history", "search.lastCompare", "recentSearches"]
|
||
answer_fields = [
|
||
{"type": "choice", "label": "价格{comparison}的", "options": ["{item1}{comparison}", "{item2}{comparison}", "相同"]},
|
||
{"type": "number", "label": "{item1}总价(¥)"},
|
||
{"type": "number", "label": "{item2}总价(¥)"},
|
||
]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
ebay = Ebay(input.apps["ebay"])
|
||
t1, t2 = Ebay.compare_top_totals(
|
||
self.p.item1, self.p.item2,
|
||
condition="全新", location="亚洲", sort_id=self.p.sort_id,
|
||
)
|
||
if self.p.sort_id == "priceLow":
|
||
winner = self.p.item1 if t1 < t2 else (self.p.item2 if t2 < t1 else "相同")
|
||
else:
|
||
winner = self.p.item1 if t1 > t2 else (self.p.item2 if t2 > t1 else "相同")
|
||
checks = [
|
||
ebay.check_has_snapshot(
|
||
query=self.p.item1, condition="全新",
|
||
location="亚洲", field=f"history.{self.p.item1}_search",
|
||
),
|
||
ebay.check_has_snapshot(
|
||
query=self.p.item2, condition="全新",
|
||
location="亚洲", field=f"history.{self.p.item2}_search",
|
||
),
|
||
]
|
||
checks.extend(build_compare_two_totals_checks(
|
||
label_expected=winner, label_key="winner",
|
||
first_total=t1, first_key=f"{self.p.item1}Total",
|
||
second_total=t2, second_key=f"{self.p.item2}Total",
|
||
winner_marker_words=[self.p.comparison], answer=input.answer,
|
||
))
|
||
return checks
|
||
|
||
|
||
class CompareTwoGroupCounts(BaseTask):
|
||
templates = [
|
||
"帮我比较两组筛选结果：{location1}发货的{condition1} {brand1} {query1}里，{price_min1} 到 {price_max1} 块的；以及{location2}发货的{condition2} {brand2} {query2}里，{price_min2} 到 {price_max2} 块的。哪个选择更多，各有多少个？",
|
||
"我想对比两个范围：{location1}发货的{condition1} {brand1} {query1}（{price_min1} 到 {price_max1}）和{location2}发货的{condition2} {brand2} {query2}（{price_min2} 到 {price_max2}）。哪个结果更多，把两个数量都告诉我。",
|
||
]
|
||
apps = ["ebay"]
|
||
scope = "S1"
|
||
objective = "hybrid"
|
||
composition = "deep_dive"
|
||
difficulty = "L3"
|
||
max_steps = 60
|
||
capabilities = ["search", "extract", "reasoning"]
|
||
parameters = {
|
||
"query1": {"type": "string", "default": "耳机"},
|
||
"brand1": {"type": "string", "default": "Sony"},
|
||
"location1": {"type": "string", "default": "欧洲"},
|
||
"condition1": {"type": "string", "default": "全新"},
|
||
"price_min1": {"type": "string", "default": "620"},
|
||
"price_max1": {"type": "string", "default": "690"},
|
||
"query2": {"type": "string", "default": "运动鞋"},
|
||
"brand2": {"type": "string", "default": "Nike"},
|
||
"location2": {"type": "string", "default": "欧洲"},
|
||
"condition2": {"type": "string", "default": "全新"},
|
||
"price_min2": {"type": "string", "default": "510"},
|
||
"price_max2": {"type": "string", "default": "540"},
|
||
"_groups": {
|
||
"sampler": Ebay.sample_compare_counts_groups,
|
||
"fields": {
|
||
"query1": "query1", "brand1": "brand1", "location1": "location1",
|
||
"condition1": "condition1", "price_min1": "price_min1", "price_max1": "price_max1",
|
||
"query2": "query2", "brand2": "brand2", "location2": "location2",
|
||
"condition2": "condition2", "price_min2": "price_min2", "price_max2": "price_max2",
|
||
},
|
||
},
|
||
}
|
||
optimal_paths = [["tab.search"]]
|
||
expected_changes = ["search.current", "search.history", "search.lastCompare", "recentSearches"]
|
||
answer_fields = [
|
||
{"type": "choice", "label": "选择更多的", "options": ["{query1}更多", "{query2}更多", "数量相同"]},
|
||
{"type": "number", "label": "{query1}数量"},
|
||
{"type": "number", "label": "{query2}数量"},
|
||
]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
ebay = Ebay(input.apps["ebay"])
|
||
snap1 = ebay.find_latest_snapshot(
|
||
query=self.p.query1, brand=self.p.brand1, condition=self.p.condition1,
|
||
location=self.p.location1, price_min=self.p.price_min1, price_max=self.p.price_max1,
|
||
)
|
||
snap2 = ebay.find_latest_snapshot(
|
||
query=self.p.query2, brand=self.p.brand2, condition=self.p.condition2,
|
||
location=self.p.location2, price_min=self.p.price_min2, price_max=self.p.price_max2,
|
||
)
|
||
c1 = expect_count(
|
||
query=self.p.query1, brand=self.p.brand1, condition=self.p.condition1,
|
||
location=self.p.location1, min_total=int(self.p.price_min1), max_total=int(self.p.price_max1),
|
||
)
|
||
c2 = expect_count(
|
||
query=self.p.query2, brand=self.p.brand2, condition=self.p.condition2,
|
||
location=self.p.location2, min_total=int(self.p.price_min2), max_total=int(self.p.price_max2),
|
||
)
|
||
if snap1 and isinstance(snap1.get("resultsCount"), (int, float)):
|
||
c1 = int(snap1["resultsCount"])
|
||
if snap2 and isinstance(snap2.get("resultsCount"), (int, float)):
|
||
c2 = int(snap2["resultsCount"])
|
||
more = self.p.query1 if c1 > c2 else (self.p.query2 if c2 > c1 else "相同")
|
||
checks = [
|
||
ebay.check_has_snapshot(
|
||
query=self.p.query1, brand=self.p.brand1, condition=self.p.condition1,
|
||
location=self.p.location1, price_min=self.p.price_min1, price_max=self.p.price_max1,
|
||
field=f"history.{self.p.query1}_search",
|
||
),
|
||
ebay.check_has_snapshot(
|
||
query=self.p.query2, brand=self.p.brand2, condition=self.p.condition2,
|
||
location=self.p.location2, price_min=self.p.price_min2, price_max=self.p.price_max2,
|
||
field=f"history.{self.p.query2}_search",
|
||
),
|
||
]
|
||
checks.extend(build_compare_counts_checks(
|
||
more_expected=more,
|
||
label1=self.p.query1, count1=c1,
|
||
label2=self.p.query2, count2=c2,
|
||
answer=input.answer,
|
||
))
|
||
return checks
|