ZNJJ-api-server/evals/prompt_compare.py

"""Compare unchanged live evaluation cases against the local LangGraph application.

Run with: conda run -n py3.11 python -m evals.prompt_compare --output-dir ...
Uses the configured real LLM; never substitutes a fake provider.
"""
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import redirect_stdout
from hashlib import sha256
import json
from pathlib import Path
import subprocess
import time

from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.core.config import Config
from src.api.endpoints import router
from src.core.fastgpt_client import lifespan
from evals.run import DEFAULT_CASES_DIR, load_cases, run_case, save_transcript

# Exercise the production text router and backend lifespan, without loading
# unrelated voice models or downloading their assets during text evaluations.
app = FastAPI(lifespan=lifespan)
app.include_router(router)


def main():
    parser = argparse.ArgumentParser(__doc__)
    parser.add_argument('--output-dir', type=Path, required=True)
    parser.add_argument('--workers', type=int, default=4)
    parser.add_argument('--case', action='append', default=[])
    parser.add_argument('--prompt-dir', type=Path, help='Optional prompt snapshot; other prompts use the project version')
    args = parser.parse_args()
    if args.prompt_dir:
        import src.agent.prompts as registry
        original_loader = registry.load_prompt
        def snapshot_loader(name):
            path = args.prompt_dir / (name+'.md')
            return path.read_text(encoding='utf-8') if path.exists() else original_loader(name)
        registry.load_prompt = snapshot_loader
    if Config.AGENT_BACKEND != 'langgraph' or not Config.LLM_PROVIDER or Config.LLM_PROVIDER == 'fake':
        raise SystemExit('Requires langgraph backend and real openai provider')
    args.output_dir.mkdir(parents=True, exist_ok=True)
    cases = load_cases(DEFAULT_CASES_DIR, args.case)
    from src.agent.prompts import load_prompt
    metadata = {
        **{k: v for k, v in (Config.llm_profile(Config.LLM_PROVIDER) or {}).items() if k in ('model', 'temperature')},
        'backend': Config.AGENT_BACKEND, 'workers': args.workers,
        'commit': subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip(),
        'case_hashes': {str(p.relative_to(DEFAULT_CASES_DIR)): sha256(p.read_bytes()).hexdigest()
                        for p in sorted(DEFAULT_CASES_DIR.rglob('*.yaml'))},
        'prompt_hashes': {p.name: sha256(load_prompt(p.stem).encode('utf-8')).hexdigest()
                          for p in Path('src/agent/prompts').glob('*.md')},    }
    (args.output_dir / 'metadata.json').write_text(json.dumps(metadata, indent=2), encoding='utf-8')
    started = time.monotonic()
    results = []
    # TestClient runs the real HTTP endpoint and application lifespan in process.
    # The original runner, inputs, expectations and cleanup stay unchanged.
    with (args.output_dir / 'dialogues.log').open('w', encoding='utf-8') as log:
        with redirect_stdout(log), TestClient(app) as client:
            with ThreadPoolExecutor(max_workers=args.workers) as pool:
                pending = {pool.submit(run_case, client, case): case for case in cases}
                for future in as_completed(pending):
                    case = pending[future]
                    try:
                        result = future.result()
                        save_transcript(args.output_dir, result)
                        results.append({'name': result.name, 'passed': result.passed,
                                        'failures': [t.error for t in result.turns if not t.ok]})
                    except Exception as exc:
                        results.append({'name': case['name'], 'passed': False,
                                        'failures': [type(exc).__name__ + ': ' + str(exc)]})
                    summary = {'completed': len(results), 'total': len(cases),
                               'passed': sum(r['passed'] for r in results),
                               'duration_seconds': round(time.monotonic() - started, 2),
                               'results': sorted(results, key=lambda r: r['name'])}
                    (args.output_dir / 'summary.json').write_text(
                        json.dumps(summary, ensure_ascii=False, indent=2), encoding='utf-8')
                    log.flush()
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()