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_skill_versions_are_immutable_and_account_scoped(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: versions = runtime.list_skill_versions('alice', 'label-master') self.assertEqual(len(versions['versions']), 1) current_id = versions['current_snapshot_id'] self.assertTrue(versions['versions'][0]['is_current']) manifest = runtime.get_skill_version_manifest( 'alice', 'label-master', current_id, ) skill_file = next( item for item in manifest['files'] if item['path'] == 'SKILL.md' ) self.assertTrue(skill_file['editable']) current_file = runtime.read_skill_version_file( 'alice', 'label-master', current_id, 'SKILL.md', ) updated_content = current_file['content'].replace( 'Read the manifest', 'Read the manifest carefully', ) saved = runtime.save_skill_version( 'alice', 'label-master', base_snapshot_id=current_id, version_name='边界调整 v1', note='测试版本', files={'SKILL.md': updated_content}, ) self.assertEqual(saved['version_name'], '边界调整 v1') self.assertEqual(saved['source_type'], 'local') self.assertNotEqual(saved['id'], current_id) original = runtime.read_skill_version_file( 'alice', 'label-master', current_id, 'SKILL.md', ) forked = runtime.read_skill_version_file( 'alice', 'label-master', saved['id'], 'SKILL.md', ) self.assertNotIn('carefully', original['content']) self.assertIn('carefully', forked['content']) dataset = runtime.create_dataset( account_id='alice', name='versioned', filename='versioned.csv', rows=[{'query': '导航去公司'}], ) experiment = runtime.create_experiment( account_id='alice', dataset_id=dataset['id'], name='local version', snapshot_id=saved['id'], ) self.assertEqual(experiment['snapshot_id'], saved['id']) self.assertEqual( experiment['skill_version'], saved['content_hash'][:12], ) with self.assertRaisesRegex( ValueError, 'Skill 版本不存在', ): runtime.read_skill_version_file( 'bob', 'label-master', saved['id'], 'SKILL.md', ) 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'), []) history = runtime.list_single_analyses('alice') self.assertEqual(len(history), 1) self.assertEqual(history[0]['id'], result['id']) self.assertEqual(history[0]['query'], '导航去公司') self.assertEqual(history[0]['prediction'], '地图导航') self.assertNotIn('raw_output', history[0]) detail = runtime.get_single_analysis(result['id'], 'alice') self.assertEqual(detail['prediction'], '地图导航') self.assertEqual(detail['canonical']['query'], '导航去公司') 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()