feat: add evaluation label mapping

This commit is contained in:
wuyang6
2026-07-24 17:34:32 +08:00
parent e2bf012f60
commit 174d1d3d01
4 changed files with 653 additions and 31 deletions
+102 -1
View File
@@ -16,6 +16,7 @@ from src.evaluation_runtime import (
calculate_metrics,
extract_accessed_refs,
labels_equivalent,
normalize_label_mapping_output,
normalize_label_master_result,
normalize_evaluation_output,
parse_dataset_bytes,
@@ -288,6 +289,22 @@ class EvaluationRuntimeTests(unittest.TestCase):
'地图导航',
)
)
self.assertTrue(labels_equivalent('慢系统', ''))
self.assertTrue(
labels_equivalent(
'complex=Ture\n慢系统',
'',
True,
)
)
self.assertTrue(labels_equivalent('complex=Ture', '', True))
self.assertFalse(
labels_equivalent(
'complex=Ture\n慢系统',
'',
False,
)
)
snapshot_metadata = json.dumps(
{
'label_catalog': {
@@ -340,6 +357,24 @@ class EvaluationRuntimeTests(unittest.TestCase):
normalized_label['function_output'],
'Agent(tag="地图导航")',
)
mapping = normalize_label_mapping_output(
json.dumps(
{
'label_map': {
'慢系统': '',
'complex=Ture': 'complex=true',
},
'unmapped': [],
'reason': '统一快慢和复杂度格式',
},
ensure_ascii=False,
),
raw_labels=['慢系统', 'complex=Ture', '未知'],
skill_name='label-fast-slow-routing',
)
self.assertEqual(mapping['label_map']['慢系统'], '')
self.assertEqual(mapping['label_map']['complex=Ture'], 'complex=true')
self.assertEqual(mapping['unmapped'], ['未知'])
def test_fast_slow_prompt_uses_skill_contract_and_relative_paths(self) -> None:
skill = BundledSkill(
@@ -351,7 +386,15 @@ class EvaluationRuntimeTests(unittest.TestCase):
prompt, runtime_context = build_evaluation_prompt(
skill=skill,
snapshot_root=Path('/tmp/evaluation-snapshot'),
canonical_case={'query': '打开空调', 'gold_label': ''},
canonical_case={
'query': '打开空调',
'gold_label': '',
'source_row': {
'query': '打开空调',
'人工标签': '',
'secret': 'should-not-leak',
},
},
)
self.assertIn('填写“快”“慢”或“模糊”', prompt)
self.assertNotIn('填写 fast 或 slow', prompt)
@@ -360,6 +403,64 @@ class EvaluationRuntimeTests(unittest.TestCase):
'不要添加 `skills/label-fast-slow-routing/` 前缀',
prompt,
)
self.assertNotIn('should-not-leak', prompt)
self.assertNotIn('"人工标签"', prompt)
def test_label_mapping_suggestion_uses_one_agent_run(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
_write_fast_slow_skill(root)
runtime = EvaluationRuntime(
root=root / 'evaluations',
cwd_for_account=lambda _account: root,
model_config_for=lambda _account: ModelConfig(model='test-model'),
account_paths_for=lambda _account: {'python_env': root / '.venv'},
max_workers=1,
)
try:
dataset = runtime.create_dataset(
account_id='alice',
name='fast-slow',
filename='fast-slow.csv',
rows=[
{'query': '打开空调', 'label': '快系统'},
{'query': '帮我规划路线', 'label': '慢系统'},
],
)
fake_result = AgentRunResult(
final_output=json.dumps(
{
'label_map': {
'快系统': '',
'慢系统': '',
},
'unmapped': [],
'reason': '统一路由标签',
},
ensure_ascii=False,
),
turns=1,
tool_calls=1,
transcript=(),
usage=UsageStats(input_tokens=10, output_tokens=5),
)
with patch(
'src.evaluation_runtime.LocalCodingAgent.run',
return_value=fake_result,
) as run:
suggestion = runtime.suggest_label_mapping(
dataset['id'],
'alice',
mapping=dataset['mapping'],
skill_name='label-fast-slow-routing',
)
self.assertEqual(run.call_count, 1)
self.assertEqual(
suggestion['label_map'],
{'快系统': '', '慢系统': ''},
)
finally:
runtime.shutdown()
def test_single_analysis_uses_snapshot_without_dataset_records(self) -> None:
with tempfile.TemporaryDirectory() as directory: