"""Real LLM NCM evaluation over known-answer cases (mission REAL LLM EVALUATION V1). Phases: gold first (65 cases, gold completely hidden from the prompt); silver and noisy run ONLY after the gold quality gate passes (selection accuracy >= 0.95 among valid answers AND high-confidence wrong rate <= 0.02). Batching verified progressively (5/10/20) with 1:1 id matching; results cached permanently per (input_fingerprint, candidate_set_hash, prompt_version, model, dataset_versions) so an identical re-execution makes ZERO new calls. AI_WRITE_ALLOWED=false; nothing here can produce a Firebird write. If no AI credential exists the run refuses to call anything and exits with BLOCKED_EXTERNAL_AI_CREDENTIAL (no key is ever invented). """ import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from sqlalchemy import select from central.app.ai.eval_metrics import attribution, confidence_buckets, metrics from central.app.ai.llm_provider import (PROMPT_VERSION_LLM, LlmAnswerError, LlmEvalItem, OpenAICompatibleProvider, candidate_set_hash, input_fingerprint, validate_answer) from central.app.ai.retrieval import EnrichedNcmCorpus, KbNeighborIndex, derive_abbreviation_lexicon, hybrid_candidates from central.app.config import settings from central.app.db import SessionLocal from central.app.models import AiEvalCache, Dataset, NcmEntry GATE_SELECTION_ACCURACY = 0.95 GATE_HIGH_CONF_WRONG = 0.02 BATCH_PROBE_SIZES = (20, 10, 5) def _load_samples(session, per_segment=40): from central.scripts.run_ncm_retrieval_eval_v2 import _samples return _samples(session, per_segment) def _build_items(samples, corpus, index, lexicon, top_k=20): items = [] meta = [] for sample in samples: ranked = hybrid_candidates(corpus, index, sample["description"], k=top_k, lexicon=lexicon, exclude_product_id=sample["product_id"]) item = LlmEvalItem( work_item_id=f"{sample['segment']}:{sample['product_id']}", description=sample["description"], normalized_description=sample["description"], brand=None, unit=None, segment=sample["segment"], candidates=ranked, ) items.append(item) meta.append({**sample, "gold_in_candidates": sample["true_ncm"] in {code for code, *_ in ranked}}) return items, meta def _cached_answers(session, items, model, ncm_sha, phase): hits = {} for item in items: row = session.scalar(select(AiEvalCache).where( AiEvalCache.input_fingerprint == input_fingerprint(item), AiEvalCache.candidate_set_hash == candidate_set_hash(item), AiEvalCache.prompt_version == PROMPT_VERSION_LLM, AiEvalCache.model == model, AiEvalCache.dataset_version_ncm == ncm_sha, AiEvalCache.phase == phase, )) if row: hits[item.work_item_id] = row.answer return hits def _diagnose_http(status_code: int) -> str: return {401: "credential invalid (AI_API_KEY)", 403: "permission/plan does not allow this endpoint or model", 404: "endpoint or model not found (AI_BASE_URL/AI_MODEL)", 429: "quota or rate limit"}.get(status_code, f"http {status_code}") def _select_model(available: list, requested: str) -> tuple: """Prefer the operator-requested model; else a current GLM suitable for classification. Never invents a name and never switches vendor — returns (model, source).""" if requested in available: return requested, "requested" glm = [model for model in available if model.lower().startswith("glm")] for candidate in glm: if "flash" in candidate.lower(): return candidate, "fallback_glm_flash" if glm: return sorted(glm)[-1], "fallback_glm_latest" return None, "no_glm_available" def _preflight(provider: OpenAICompatibleProvider, requested_model: str) -> dict: """Cheap connectivity check (GET /models — no completion tokens). Validates auth, base URL and model availability; diagnoses 401/403/404/429 autonomously.""" import httpx try: available = provider.list_models() except httpx.HTTPStatusError as error: return {"zai_connection": "FAIL", "diagnosis": _diagnose_http(error.response.status_code), "model_available": False} except httpx.HTTPError as error: return {"zai_connection": "FAIL", "diagnosis": f"transport: {type(error).__name__}", "model_available": False} model, source = _select_model(available, requested_model) return {"zai_connection": "PASS", "model_available": model is not None, "model": model, "model_source": source, "available_models": available if model is None else None, "base_url": provider.base_url} def _probe_batch_size(provider, items): """Try the largest batch that keeps valid JSON with 1:1 ids on a small probe slice.""" for size in BATCH_PROBE_SIZES: probe = items[:size] try: provider.select_batch(probe) return size except (LlmAnswerError, json.JSONDecodeError): continue return 1 def run_phase(provider_factory, phase, samples, corpus, index, lexicon, session, ncm_sha, valid_ncm): items, meta = _build_items(samples, corpus, index, lexicon) cached = _cached_answers(session, items, provider_factory.model if hasattr(provider_factory, "model") else settings.ai_model, ncm_sha, phase) todo = [item for item in items if item.work_item_id not in cached] answers = dict(cached) provider = None batch_size = 0 if todo: provider = provider_factory() batch_size = _probe_batch_size(provider, todo) for start in range(0, len(todo), batch_size): for answer in provider.select_batch(todo[start:start + batch_size]): answers[answer["work_item_id"]] = answer rows, errors = [], [] for item, sample in zip(items, meta): answer = answers.get(item.work_item_id) invalid = False if answer is None: invalid = True answer = {"selected_ncm": None, "confidence": 0.0, "abstain": True, "reason_short": "missing"} else: try: validate_answer(answer, item) if answer["selected_ncm"] is not None and answer["selected_ncm"] not in valid_ncm: invalid = True except LlmAnswerError: invalid = True selected = None if (invalid or answer["abstain"] or answer["selected_ncm"] is None) else answer["selected_ncm"] rows.append({"gold_in_candidates": sample["gold_in_candidates"], "selected": selected, "gold": sample["true_ncm"], "confidence": answer["confidence"], "invalid": invalid}) if selected is not None and selected != sample["true_ncm"]: errors.append({"work_item_id": item.work_item_id, "gold": sample["true_ncm"], "selected": selected, "confidence": answer["confidence"], "gold_in_candidates": sample["gold_in_candidates"], "reason_short": answer.get("reason_short", "")}) if invalid or answer.get("abstain") or selected is not None: if item.work_item_id not in cached: session.add(AiEvalCache(input_fingerprint=input_fingerprint(item), candidate_set_hash=candidate_set_hash(item), prompt_version=PROMPT_VERSION_LLM, model=provider.model if provider else settings.ai_model, dataset_version_ncm=ncm_sha, phase=phase, answer=answer, gold_ncm=sample["true_ncm"])) session.commit() return rows, errors, provider, batch_size, len(cached) def main(output_dir="/tmp", per_segment=40) -> dict: if not settings.ai_api_key: print(json.dumps({"status": "BLOCKED_EXTERNAL_AI_CREDENTIAL", "operator_action": "set AI_API_KEY (and optionally AI_BASE_URL/AI_MODEL) in the Hub environment (.env); never commit or log it"}, indent=2)) sys.exit(2) session = SessionLocal() dataset = session.scalar(select(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE")) valid_ncm = set(session.scalars(select(NcmEntry.code).where(NcmEntry.dataset_id == dataset.id)).all()) corpus = EnrichedNcmCorpus.build(session) index = KbNeighborIndex.build(session) lexicon = derive_abbreviation_lexicon(index.corpus(), min_support=3) samples = _load_samples(session, per_segment) # ---- preflight: cheap /models call, never spends completion tokens ---- pre_provider = OpenAICompatibleProvider(api_key=settings.ai_api_key, model=settings.ai_model, base_url=settings.ai_base_url) preflight = _preflight(pre_provider, settings.ai_model) if preflight["zai_connection"] != "PASS" or not preflight["model_available"]: print(json.dumps({"status": "ZAI_PREFLIGHT_FAIL", "preflight": preflight}, indent=2)) sys.exit(3) model_name = preflight["model"] factory = lambda: OpenAICompatibleProvider(api_key=settings.ai_api_key, model=model_name, base_url=settings.ai_base_url) gold = [sample for sample in samples if sample["quality"] == "GOLD"] provider = factory() rows, errors, provider_used, batch_size, cache_hits = run_phase(lambda: provider, "gold", gold, corpus, index, lexicon, session, dataset.sha256, valid_ncm) m = metrics(rows) prices_configured = settings.ai_price_in_per_m is not None and settings.ai_price_out_per_m is not None usage = provider.usage if provider is not None else {"input_tokens": 0, "output_tokens": 0, "cached_tokens": 0} cost_total = round(usage["input_tokens"] / 1e6 * settings.ai_price_in_per_m + usage["output_tokens"] / 1e6 * settings.ai_price_out_per_m, 6) if prices_configured else "UNKNOWN" report = { "artifact": "REAL_LLM_NCM_EVALUATION_V1", "phase": "gold", "provider": "z.ai", "base_url": settings.ai_base_url, "model": model_name, "prompt_version": PROMPT_VERSION_LLM, "preflight": preflight, "metrics": m, "attribution": attribution(rows), "confidence_buckets": confidence_buckets(rows), "batch_size_selected": batch_size, "gate": {"selection_accuracy_min": GATE_SELECTION_ACCURACY, "high_conf_wrong_max": GATE_HIGH_CONF_WRONG, "passed": (m["SELECTION_ACCURACY_WHEN_GOLD_AVAILABLE"] or 0) >= GATE_SELECTION_ACCURACY and (m["HIGH_CONFIDENCE_WRONG_RATE"] or 0) <= GATE_HIGH_CONF_WRONG}, "cache_hits_first_run": cache_hits, "ai_write_allowed": False, "usage": {**usage, "http_calls": provider.http_calls, "retries": provider.retries, "average_items_per_call": round(len(rows) / max(provider.http_calls, 1), 2), "tokens_per_item": round((usage["input_tokens"] + usage["output_tokens"]) / max(len(rows), 1), 1), "cost_total": cost_total, "cost_per_item": (round(cost_total / max(len(rows), 1), 6) if prices_configured else "UNKNOWN"), "prices_configured": prices_configured}, } Path(f"{output_dir}/REAL_LLM_NCM_EVALUATION_V1.json").write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") Path(f"{output_dir}/REAL_LLM_NCM_ERRORS_V1.json").write_text(json.dumps(errors, indent=2, ensure_ascii=False), encoding="utf-8") # cache proof: identical re-execution must make zero new calls provider2 = factory() rows2, _, _, _, cache_hits2 = run_phase(lambda: provider2, "gold", gold, corpus, index, lexicon, session, dataset.sha256, valid_ncm) second_calls = provider2.http_calls report["cache_proof"] = {"second_run_calls": second_calls, "second_run_cache_hits": cache_hits2, "cache_hit_reexecution": second_calls == 0, "identical_metrics": metrics(rows2)["CORRECT"] == m["CORRECT"]} Path(f"{output_dir}/REAL_LLM_NCM_EVALUATION_V1.json").write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") Path(f"{output_dir}/REAL_LLM_COST_V1.json").write_text(json.dumps({"artifact": "REAL_LLM_COST_V1", "usage": report["usage"], "previous_estimate_usd": 0.44, "cache_proof": report["cache_proof"]}, indent=2), encoding="utf-8") session.close() return report if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--output-dir", default="/tmp") parser.add_argument("--per-segment", type=int, default=40) args = parser.parse_args() print(json.dumps(main(args.output_dir, args.per_segment), indent=2))