Spaces:
Runtime error
Runtime error
File size: 4,194 Bytes
4e837d8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import argparse
import json
import logging
import sys
from .config import load_app_config
from .demo_packs import load_demo_pack
from .logging_utils import setup_logging
from .storage import SQLiteStore
from .tracing import utc_now, write_trace_artifact
@dataclass
class EvalResult:
project: str
pack_id: str
passed: bool
findings: list[str]
result: dict[str, Any]
trace_path: str
def _expected_subset(expected: dict[str, Any], actual: dict[str, Any]) -> list[str]:
issues = []
for key, value in expected.items():
if key not in actual:
issues.append(f'missing key: {key}')
elif actual[key] != value:
issues.append(f'{key}: expected {value!r}, got {actual[key]!r}')
return issues
def run_eval_for_project(project_module: str, pack_path: str | Path, db_path: str | Path | None = None) -> EvalResult:
mod = __import__(project_module, fromlist=['create_project_spec'])
spec = mod.create_project_spec()
pack = load_demo_pack(pack_path)
config = load_app_config(project_key=spec.key, data_subdir=spec.data_subdir)
if db_path is not None:
config = config.__class__(
project_key=config.project_key,
app_mode=config.app_mode,
root_dir=config.root_dir,
data_dir=config.data_dir,
sqlite_path=Path(db_path),
artifact_dir=config.artifact_dir,
cache_dir=config.cache_dir,
model_registry_path=config.model_registry_path,
)
started_at = utc_now()
store = SQLiteStore(config.sqlite_path, config.artifact_dir)
try:
result = spec.run_pack(pack, store, config)
expected = pack.expected_signals
findings = _expected_subset(expected, result)
passed = not findings
finished_at = utc_now()
trace_path = write_trace_artifact(
config.artifact_dir,
{
'kind': 'eval',
'project': spec.key,
'pack_id': pack.pack_id,
'pack_path': str(pack_path),
'started_at': started_at,
'finished_at': finished_at,
'passed': passed,
'findings': findings,
'result': result,
},
)
finally:
store.close()
return EvalResult(project=spec.key, pack_id=pack.pack_id, passed=passed, findings=findings, result=result, trace_path=str(trace_path))
def main() -> int:
parser = argparse.ArgumentParser(description='Run golden-scenario evals for the ALL4 kit')
parser.add_argument('project_module', help='Python module path, e.g. apps.p4_food_waste.app')
parser.add_argument('pack_path', help='Path to a demo pack folder')
parser.add_argument('--db-path', help='Optional SQLite path for the run')
parser.add_argument(
'--json-only',
action='store_true',
help='Emit exactly one JSON object to stdout (no logging, no pretty-print).',
)
parser.add_argument(
'--quiet',
'--no-log',
dest='quiet',
action='store_true',
help='Disable JSONL logging (useful when piping stdout).',
)
parser.add_argument(
'--log-level',
choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
default='INFO',
help='Logging threshold for the JSONL status line.',
)
args = parser.parse_args()
logger = None
if not args.quiet and not args.json_only:
logger = setup_logging('app_kit.eval_runner', level=getattr(logging, args.log_level), stream=sys.stderr)
result = run_eval_for_project(args.project_module, args.pack_path, args.db_path)
if logger is not None:
logger.info('eval completed: %s', json.dumps(result.__dict__, ensure_ascii=False))
if args.json_only:
print(json.dumps(result.__dict__, ensure_ascii=False))
else:
print(json.dumps(result.__dict__, indent=2, ensure_ascii=False))
return 0 if result.passed else 1
if __name__ == '__main__':
raise SystemExit(main())
|