feat: add skill evaluation workbench
This commit is contained in:
wuyang6
@@ -0,0 +1,502 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from src.agent_types import AgentRunResult, ModelConfig, UsageStats
|
||||
from src.bundled_skills import BundledSkill
|
||||
from src.evaluation_runtime import (
|
||||
EvaluationRuntime,
|
||||
build_evaluation_prompt,
|
||||
calculate_metrics,
|
||||
extract_accessed_refs,
|
||||
labels_equivalent,
|
||||
normalize_label_master_result,
|
||||
normalize_evaluation_output,
|
||||
parse_dataset_bytes,
|
||||
prediction_is_allowed,
|
||||
)
|
||||
|
||||
|
||||
def _write_test_skill(root: Path) -> None:
|
||||
skill_dir = root / 'skills' / 'label-master'
|
||||
manifest_dir = skill_dir / 'knowledge' / '索引'
|
||||
manifest_dir.mkdir(parents=True)
|
||||
(skill_dir / 'SKILL.md').write_text(
|
||||
(
|
||||
'---\n'
|
||||
'name: label-master\n'
|
||||
'description: Test label skill.\n'
|
||||
'allowed_tools: read_file, grep_search\n'
|
||||
'---\n'
|
||||
'Read the manifest and classify the query.\n'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
(manifest_dir / 'label_manifest.json').write_text(
|
||||
json.dumps(
|
||||
{
|
||||
'counts': {'labels': 2, 'functions': 1},
|
||||
'labels': [
|
||||
{
|
||||
'name': '地图导航',
|
||||
'domain': '地图',
|
||||
'path': 'knowledge/标签/地图导航.md',
|
||||
},
|
||||
{
|
||||
'name': 'QA',
|
||||
'domain': '问答',
|
||||
'path': 'knowledge/标签/QA.md',
|
||||
},
|
||||
],
|
||||
'agent_tags': ['地图导航'],
|
||||
'functions': [{'name': 'QA'}],
|
||||
'intents': [],
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
|
||||
|
||||
def _write_fast_slow_skill(root: Path) -> None:
|
||||
skill_dir = root / 'skills' / 'label-fast-slow-routing'
|
||||
skill_dir.mkdir(parents=True)
|
||||
(skill_dir / 'SKILL.md').write_text(
|
||||
(
|
||||
'---\n'
|
||||
'name: label-fast-slow-routing\n'
|
||||
'description: Test fast slow routing skill.\n'
|
||||
'---\n'
|
||||
'Classify the query as 快、慢 or 模糊.\n'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
|
||||
|
||||
class EvaluationRuntimeTests(unittest.TestCase):
|
||||
def test_parse_csv_and_jsonl(self) -> None:
|
||||
csv_rows, csv_format = parse_dataset_bytes(
|
||||
'sample.csv',
|
||||
'query,label\n打开地图,地图导航\n'.encode(),
|
||||
)
|
||||
self.assertEqual(csv_format, 'csv')
|
||||
self.assertEqual(csv_rows[0]['query'], '打开地图')
|
||||
|
||||
jsonl_rows, jsonl_format = parse_dataset_bytes(
|
||||
'sample.jsonl',
|
||||
b'{"query":"hello","label":"QA"}\n',
|
||||
)
|
||||
self.assertEqual(jsonl_format, 'jsonl')
|
||||
self.assertEqual(jsonl_rows[0]['label'], 'QA')
|
||||
|
||||
def test_dataset_mapping_snapshot_and_experiment(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
_write_test_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:
|
||||
metadata = runtime.metadata('alice')
|
||||
self.assertEqual(metadata['default_skill'], 'label-master')
|
||||
self.assertEqual(
|
||||
[item['name'] for item in metadata['label_catalog']['labels']],
|
||||
['地图导航', 'QA'],
|
||||
)
|
||||
|
||||
content = base64.b64encode(
|
||||
'rid,query,人工标签\n1,导航去公司,Agent(tag="地图导航")\n'.encode()
|
||||
).decode()
|
||||
dataset = runtime.create_dataset(
|
||||
account_id='alice',
|
||||
name='test',
|
||||
filename='test.csv',
|
||||
content_base64=content,
|
||||
)
|
||||
self.assertEqual(dataset['mapping']['fields']['query'], 'query')
|
||||
self.assertEqual(
|
||||
dataset['mapping']['fields']['gold_label'],
|
||||
'人工标签',
|
||||
)
|
||||
self.assertEqual(dataset['mapping']['fields']['request_id'], 'rid')
|
||||
|
||||
experiment = runtime.create_experiment(
|
||||
account_id='alice',
|
||||
dataset_id=dataset['id'],
|
||||
name='first',
|
||||
)
|
||||
self.assertEqual(experiment['total_cases'], 1)
|
||||
self.assertEqual(experiment['skill_name'], 'label-master')
|
||||
self.assertEqual(len(experiment['skill_version']), 12)
|
||||
self.assertEqual(experiment['cases'][0]['request_id'], '1')
|
||||
finally:
|
||||
runtime.shutdown()
|
||||
|
||||
def test_fast_slow_skill_is_the_evaluation_default(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
_write_test_skill(root)
|
||||
_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:
|
||||
metadata = runtime.metadata('alice')
|
||||
self.assertEqual(
|
||||
metadata['default_skill'],
|
||||
'label-fast-slow-routing',
|
||||
)
|
||||
defaults = [
|
||||
item['name']
|
||||
for item in metadata['skills']
|
||||
if item['default']
|
||||
]
|
||||
self.assertEqual(defaults, ['label-fast-slow-routing'])
|
||||
finally:
|
||||
runtime.shutdown()
|
||||
|
||||
def test_result_normalization_and_label_equivalence(self) -> None:
|
||||
result = normalize_evaluation_output(
|
||||
json.dumps(
|
||||
{
|
||||
'prediction': '地图导航',
|
||||
'function_output': 'Agent(tag="地图导航")',
|
||||
'complex': False,
|
||||
'reason': '导航执行',
|
||||
'candidates': ['地图导航', '地图问答'],
|
||||
'evidence_refs': ['knowledge/标签/地图导航.md'],
|
||||
'confidence': 1.4,
|
||||
'uncertainty': '',
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
)
|
||||
self.assertEqual(result['prediction'], '地图导航')
|
||||
self.assertEqual(result['confidence'], 1.0)
|
||||
self.assertTrue(labels_equivalent('Agent(tag="地图导航")', '地图导航'))
|
||||
self.assertTrue(
|
||||
labels_equivalent(
|
||||
'complex=false\nAgent(tag="地图导航")',
|
||||
'地图导航',
|
||||
)
|
||||
)
|
||||
snapshot_metadata = json.dumps(
|
||||
{
|
||||
'label_catalog': {
|
||||
'labels': ['地图导航'],
|
||||
'agent_tags': ['地图导航'],
|
||||
'functions': ['QA'],
|
||||
'intents': [],
|
||||
}
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
self.assertTrue(
|
||||
prediction_is_allowed(
|
||||
'label-master',
|
||||
'Agent(tag="地图导航")',
|
||||
snapshot_metadata,
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
prediction_is_allowed(
|
||||
'label-master',
|
||||
'已完成第一步知识读取',
|
||||
snapshot_metadata,
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
prediction_is_allowed(
|
||||
'label-fast-slow-routing',
|
||||
'模糊',
|
||||
'{}',
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
prediction_is_allowed(
|
||||
'label-fast-slow-routing',
|
||||
'fast',
|
||||
'{}',
|
||||
)
|
||||
)
|
||||
normalized_label = normalize_label_master_result(
|
||||
{
|
||||
**result,
|
||||
'prediction': 'Agent(tag="地图导航")',
|
||||
'function_output': '一段解释文字',
|
||||
},
|
||||
snapshot_metadata,
|
||||
)
|
||||
self.assertEqual(normalized_label['prediction'], '地图导航')
|
||||
self.assertEqual(
|
||||
normalized_label['function_output'],
|
||||
'Agent(tag="地图导航")',
|
||||
)
|
||||
|
||||
def test_fast_slow_prompt_uses_skill_contract_and_relative_paths(self) -> None:
|
||||
skill = BundledSkill(
|
||||
name='label-fast-slow-routing',
|
||||
description='Test fast slow routing skill.',
|
||||
source='directory',
|
||||
get_prompt=lambda _agent, _args: 'Read references/policy.md.',
|
||||
)
|
||||
prompt, runtime_context = build_evaluation_prompt(
|
||||
skill=skill,
|
||||
snapshot_root=Path('/tmp/evaluation-snapshot'),
|
||||
canonical_case={'query': '打开空调', 'gold_label': '快'},
|
||||
)
|
||||
self.assertIn('填写“快”“慢”或“模糊”', prompt)
|
||||
self.assertNotIn('填写 fast 或 slow', prompt)
|
||||
self.assertIn('`references/...`', runtime_context)
|
||||
self.assertNotIn(
|
||||
'不要添加 `skills/label-fast-slow-routing/` 前缀',
|
||||
prompt,
|
||||
)
|
||||
|
||||
def test_single_analysis_uses_snapshot_without_dataset_records(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
_write_test_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,
|
||||
)
|
||||
fake_result = AgentRunResult(
|
||||
final_output=json.dumps(
|
||||
{
|
||||
'prediction': '地图导航',
|
||||
'function_output': 'Agent(tag="地图导航")',
|
||||
'complex': False,
|
||||
'reason': '导航执行',
|
||||
'candidates': ['地图导航'],
|
||||
'evidence_refs': [
|
||||
'knowledge/索引/label_manifest.json'
|
||||
],
|
||||
'confidence': 0.98,
|
||||
'uncertainty': '',
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
turns=1,
|
||||
tool_calls=1,
|
||||
transcript=(
|
||||
{
|
||||
'role': 'assistant',
|
||||
'tool_calls': [
|
||||
{
|
||||
'function': {
|
||||
'name': 'read_file',
|
||||
'arguments': json.dumps(
|
||||
{
|
||||
'path': (
|
||||
'knowledge/索引/'
|
||||
'label_manifest.json'
|
||||
)
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
}
|
||||
}
|
||||
],
|
||||
},
|
||||
),
|
||||
usage=UsageStats(input_tokens=10, output_tokens=5),
|
||||
)
|
||||
try:
|
||||
with patch(
|
||||
'src.evaluation_runtime.LocalCodingAgent.run',
|
||||
return_value=fake_result,
|
||||
):
|
||||
result = runtime.analyze_case(
|
||||
account_id='alice',
|
||||
query='导航去公司',
|
||||
)
|
||||
self.assertEqual(result['prediction'], '地图导航')
|
||||
self.assertEqual(result['skill_name'], 'label-master')
|
||||
self.assertEqual(result['model'], 'test-model')
|
||||
self.assertEqual(runtime.list_datasets('alice'), [])
|
||||
self.assertEqual(runtime.list_experiments('alice'), [])
|
||||
finally:
|
||||
runtime.shutdown()
|
||||
|
||||
def test_background_runner_completes_cases(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
_write_test_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='test',
|
||||
filename='test.csv',
|
||||
rows=[
|
||||
{'query': '导航去公司', 'label': '地图导航'},
|
||||
{'query': '导航回家', 'label': '地图导航'},
|
||||
],
|
||||
)
|
||||
experiment = runtime.create_experiment(
|
||||
account_id='alice',
|
||||
dataset_id=dataset['id'],
|
||||
name='runner',
|
||||
concurrency=1,
|
||||
)
|
||||
fake_result = AgentRunResult(
|
||||
final_output=json.dumps(
|
||||
{
|
||||
'prediction': '地图导航',
|
||||
'function_output': 'Agent(tag="地图导航")',
|
||||
'complex': False,
|
||||
'reason': '导航执行',
|
||||
'candidates': ['地图导航'],
|
||||
'evidence_refs': ['knowledge/标签/地图导航.md'],
|
||||
'confidence': 0.98,
|
||||
'uncertainty': '',
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
turns=1,
|
||||
tool_calls=1,
|
||||
transcript=(
|
||||
{
|
||||
'role': 'assistant',
|
||||
'tool_calls': [
|
||||
{
|
||||
'function': {
|
||||
'name': 'read_file',
|
||||
'arguments': json.dumps(
|
||||
{
|
||||
'path': (
|
||||
'knowledge/索引/'
|
||||
'label_manifest.json'
|
||||
)
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
}
|
||||
}
|
||||
],
|
||||
},
|
||||
),
|
||||
usage=UsageStats(input_tokens=10, output_tokens=5),
|
||||
)
|
||||
with patch(
|
||||
'src.evaluation_runtime.LocalCodingAgent.run',
|
||||
return_value=fake_result,
|
||||
):
|
||||
runtime.start(experiment['id'], 'alice')
|
||||
deadline = time.time() + 3
|
||||
current = runtime.get_experiment(experiment['id'], 'alice')
|
||||
while (
|
||||
current['status'] not in {'completed', 'completed_with_errors'}
|
||||
and time.time() < deadline
|
||||
):
|
||||
time.sleep(0.02)
|
||||
current = runtime.get_experiment(
|
||||
experiment['id'],
|
||||
'alice',
|
||||
)
|
||||
self.assertEqual(current['status'], 'completed')
|
||||
self.assertEqual(current['completed_cases'], 2)
|
||||
self.assertEqual(current['metrics']['accuracy'], 1.0)
|
||||
finally:
|
||||
runtime.shutdown()
|
||||
|
||||
def test_accessed_refs_only_include_snapshot_paths(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
snapshot_root = Path(directory)
|
||||
evidence = snapshot_root / 'skills' / 'label-master' / 'knowledge.md'
|
||||
evidence.parent.mkdir(parents=True)
|
||||
evidence.write_text('evidence', encoding='utf-8')
|
||||
transcript = (
|
||||
{
|
||||
'role': 'assistant',
|
||||
'tool_calls': [
|
||||
{
|
||||
'function': {
|
||||
'name': 'read_file',
|
||||
'arguments': json.dumps(
|
||||
{
|
||||
'path': (
|
||||
'skills/label-master/knowledge.md'
|
||||
)
|
||||
}
|
||||
),
|
||||
}
|
||||
},
|
||||
{
|
||||
'function': {
|
||||
'name': 'read_file',
|
||||
'arguments': json.dumps({'path': '/etc/hosts'}),
|
||||
}
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
self.assertEqual(
|
||||
extract_accessed_refs(
|
||||
transcript,
|
||||
snapshot_root=snapshot_root,
|
||||
),
|
||||
[
|
||||
{
|
||||
'tool': 'read_file',
|
||||
'path': 'skills/label-master/knowledge.md',
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
def test_metrics_count_disagreements(self) -> None:
|
||||
metrics = calculate_metrics(
|
||||
[
|
||||
{
|
||||
'status': 'completed',
|
||||
'gold_label': 'A',
|
||||
'prediction': 'A',
|
||||
'correct': True,
|
||||
'confidence': 0.9,
|
||||
'evidence_refs': ['a.md'],
|
||||
'domain': 'one',
|
||||
},
|
||||
{
|
||||
'status': 'completed',
|
||||
'gold_label': 'A',
|
||||
'prediction': 'B',
|
||||
'correct': False,
|
||||
'confidence': 0.5,
|
||||
'evidence_refs': [],
|
||||
'domain': 'one',
|
||||
},
|
||||
]
|
||||
)
|
||||
self.assertEqual(metrics['labeled'], 2)
|
||||
self.assertEqual(metrics['disagreements'], 1)
|
||||
self.assertEqual(metrics['accuracy'], 0.5)
|
||||
self.assertEqual(metrics['low_confidence'], 1)
|
||||
self.assertEqual(metrics['no_evidence'], 1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user