"""Reads and writes behind the evaluation console. Holds three things the router should not: the built-in seed (idempotent, so re-seeding after a schema change never duplicates), the question-bank mining that turns production into test cases, and the run-to-run comparison the scheduled job alerts on. """ from __future__ import annotations import logging import re import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.assistant.evaluation.constants import ( CASE_ACTIVE, CASE_DRAFT, CASE_RETIRED, CASE_STATUSES, CATEGORIES, DEFAULT_CONCURRENCY, DEFAULT_REPEATS, LANGS, MAX_CASES_PER_RUN, MAX_CONCURRENCY, MAX_REPEATS, ORIGIN_CURATED, ORIGIN_EVAL_FAILURE, ORIGIN_GAP, ORIGIN_GENERATED, ORIGIN_MINED, REGRESSION_MIN_ATTEMPTS, RUN_RUNNING, RUN_TERMINAL_STATUSES, SUITE_ACTIVE, TRIGGER_MANUAL, VERDICT_FAIL, ) from app.assistant.evaluation.models import ( AssistantEvalCase, AssistantEvalResult, AssistantEvalRun, AssistantEvalSuite, ) from app.assistant.evaluation.runner import EvalRunError from app.assistant.evaluation.seed_data import SUITE_CASES, SUITE_DEFINITIONS logger = logging.getLogger(__name__) def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) # --------------------------------------------------------------------------- # Seed # --------------------------------------------------------------------------- async def seed_builtin_suites(db: AsyncSession) -> dict[str, Any]: """Create the built-in suites and their cases if they are not there. Idempotent in both directions and deliberately conservative: an existing suite is never rewritten, and an existing case is matched **by its question text within its suite** and left alone. Someone has edited these — that is what the console is for — and a seed that overwrites edits is a seed nobody dares run twice. New cases added to :mod:`.seed_data` in a later release DO appear, as ``draft``. They are questions the platform's authors thought of, not questions this deployment agreed to gate on, so a person promotes them. """ created_suites = 0 created_cases = 0 for definition in SUITE_DEFINITIONS: suite = ( await db.execute( select(AssistantEvalSuite).where( AssistantEvalSuite.slug == definition["slug"] ) ) ).scalar_one_or_none() if suite is None: suite = AssistantEvalSuite( slug=definition["slug"], name=definition["name"], description=definition["description"], status=SUITE_ACTIVE, is_builtin=True, ) db.add(suite) await db.flush() created_suites += 1 first_seed = True else: first_seed = False existing = { row[0] for row in ( await db.execute( select(AssistantEvalCase.question).where( AssistantEvalCase.suite_id == suite.id ) ) ).all() } ordinal = ( await db.execute( select(func.coalesce(func.max(AssistantEvalCase.ordinal), 0)).where( AssistantEvalCase.suite_id == suite.id ) ) ).scalar() or 0 for spec in SUITE_CASES.get(definition["slug"], ()): if spec["question"] in existing: continue ordinal += 1 db.add( AssistantEvalCase( suite_id=suite.id, ordinal=ordinal, question=spec["question"], setup_turns=spec.get("setup_turns"), lang=spec.get("lang", "zh"), category=spec["category"], # On the very first seed the bank ships ready to run: # forcing a human to promote 90 cases they did not write is # how a harness ends up never being used at all. Cases that # appear LATER, in a suite that already exists, arrive as # drafts — by then this deployment has its own baseline and # a silently widening gate would move it. status=CASE_ACTIVE if first_seed else CASE_DRAFT, origin=ORIGIN_CURATED, expectations=spec.get("expectations") or None, ) ) created_cases += 1 await db.flush() return {"suites_created": created_suites, "cases_created": created_cases} # --------------------------------------------------------------------------- # Suites and cases # --------------------------------------------------------------------------- async def list_suites(db: AsyncSession) -> list[dict[str, Any]]: suites = ( ( await db.execute( select(AssistantEvalSuite).order_by(AssistantEvalSuite.name.asc()) ) ) .scalars() .all() ) if not suites: return [] counts = { (row[0], row[1]): row[2] for row in ( await db.execute( select( AssistantEvalCase.suite_id, AssistantEvalCase.status, func.count(), ).group_by(AssistantEvalCase.suite_id, AssistantEvalCase.status) ) ).all() } latest = { row[0]: row[1] for row in ( await db.execute( select( AssistantEvalRun.suite_id, func.max(AssistantEvalRun.started_at) ).group_by(AssistantEvalRun.suite_id) ) ).all() } out: list[dict[str, Any]] = [] for suite in suites: out.append( { "id": str(suite.id), "slug": suite.slug, "name": suite.name, "description": suite.description, "status": suite.status, "is_builtin": suite.is_builtin, "default_persona_person_id": ( str(suite.default_persona_person_id) if suite.default_persona_person_id else None ), "cases_active": counts.get((suite.id, CASE_ACTIVE), 0), "cases_draft": counts.get((suite.id, CASE_DRAFT), 0), "cases_retired": counts.get((suite.id, CASE_RETIRED), 0), "last_run_at": _iso(latest.get(suite.id)), } ) return out def _iso(value: datetime | None) -> str | None: return value.isoformat() if value else None def serialize_case(case: AssistantEvalCase) -> dict[str, Any]: return { "id": str(case.id), "suite_id": str(case.suite_id), "ordinal": case.ordinal, "question": case.question, "setup_turns": case.setup_turns or [], "lang": case.lang, "category": case.category, "status": case.status, "origin": case.origin, "origin_ref": case.origin_ref, "persona_person_id": ( str(case.persona_person_id) if case.persona_person_id else None ), "expectations": case.expectations or {}, "reviewed_at": _iso(case.reviewed_at), "created_at": _iso(case.created_at), "updated_at": _iso(case.updated_at), } async def list_cases( db: AsyncSession, *, suite_id: uuid.UUID | None = None, status: str | None = None, category: str | None = None, origin: str | None = None, q: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[dict[str, Any]], int]: stmt = select(AssistantEvalCase) count_stmt = select(func.count()).select_from(AssistantEvalCase) conditions = [] if suite_id: conditions.append(AssistantEvalCase.suite_id == suite_id) if status in CASE_STATUSES: conditions.append(AssistantEvalCase.status == status) if category in CATEGORIES: conditions.append(AssistantEvalCase.category == category) if origin: conditions.append(AssistantEvalCase.origin == origin) if q and q.strip(): needle = f"%{q.strip()}%" conditions.append(AssistantEvalCase.question.like(needle)) for condition in conditions: stmt = stmt.where(condition) count_stmt = count_stmt.where(condition) total = (await db.execute(count_stmt)).scalar() or 0 rows = ( ( await db.execute( stmt.order_by( AssistantEvalCase.ordinal.asc(), AssistantEvalCase.id.asc() ) .offset(max(0, (page - 1) * page_size)) .limit(page_size) ) ) .scalars() .all() ) return [serialize_case(case) for case in rows], total def _clean_expectations(raw: dict[str, Any] | None) -> dict[str, Any] | None: """Keep only the keys the checker and the judge understand. An unknown key in this blob is not harmless: it looks like a specification to whoever reads the case and constrains nothing at all, which is the worst of both worlds. """ if not isinstance(raw, dict): return None allowed_lists = ("must_mention", "must_not_mention", "expect_tools") allowed_bools = ("expect_action", "expect_escalate") allowed_text = ("reference_answer", "notes") out: dict[str, Any] = {} for key in allowed_lists: value = raw.get(key) if isinstance(value, list) and value: cleaned = [ item for item in value if (isinstance(item, str) and item.strip()) or (isinstance(item, list) and any(str(x).strip() for x in item)) ] if cleaned: out[key] = cleaned for key in allowed_bools: if isinstance(raw.get(key), bool): out[key] = raw[key] for key in allowed_text: value = raw.get(key) if isinstance(value, str) and value.strip(): out[key] = value.strip()[:4000] return out or None def _clean_turns(raw: Any) -> list[dict[str, str]] | None: if not isinstance(raw, list): return None out: list[dict[str, str]] = [] for item in raw[:20]: if not isinstance(item, dict): continue role = str(item.get("role") or "").strip().lower() content = str(item.get("content") or "").strip() if role in ("user", "assistant") and content: out.append({"role": role, "content": content[:4000]}) return out or None async def create_case( db: AsyncSession, *, suite_id: uuid.UUID, question: str, category: str, lang: str = "zh", setup_turns: Any = None, expectations: dict[str, Any] | None = None, persona_person_id: uuid.UUID | None = None, status: str = CASE_ACTIVE, origin: str = ORIGIN_CURATED, origin_ref: str | None = None, actor_person_id: uuid.UUID | None = None, ) -> AssistantEvalCase: suite = await db.get(AssistantEvalSuite, suite_id) if suite is None: raise EvalRunError("ASSISTANT_EVAL_SUITE_NOT_FOUND", "题库不存在。") ordinal = ( await db.execute( select(func.coalesce(func.max(AssistantEvalCase.ordinal), 0)).where( AssistantEvalCase.suite_id == suite_id ) ) ).scalar() or 0 case = AssistantEvalCase( suite_id=suite_id, ordinal=ordinal + 1, question=question.strip()[:4000], setup_turns=_clean_turns(setup_turns), lang=lang if lang in LANGS else "zh", category=category if category in CATEGORIES else CATEGORIES[0], status=status if status in CASE_STATUSES else CASE_DRAFT, origin=origin, origin_ref=origin_ref, persona_person_id=persona_person_id, expectations=_clean_expectations(expectations), created_by_person_id=actor_person_id, ) if case.status == CASE_ACTIVE: case.reviewed_by_person_id = actor_person_id case.reviewed_at = _naive_now() db.add(case) await db.flush() return case async def update_case( db: AsyncSession, *, case: AssistantEvalCase, actor_person_id: uuid.UUID | None, **fields: Any, ) -> AssistantEvalCase: if "question" in fields and fields["question"]: case.question = str(fields["question"]).strip()[:4000] if "category" in fields and fields["category"] in CATEGORIES: case.category = fields["category"] if "lang" in fields and fields["lang"] in LANGS: case.lang = fields["lang"] if "setup_turns" in fields: case.setup_turns = _clean_turns(fields["setup_turns"]) if "expectations" in fields: case.expectations = _clean_expectations(fields["expectations"]) if "persona_person_id" in fields: case.persona_person_id = fields["persona_person_id"] if "status" in fields and fields["status"] in CASE_STATUSES: previous = case.status case.status = fields["status"] # Promotion out of draft is the review. Recording who and when is what # makes a mined question — which carried somebody's own words — safe to # keep: there is a name against the decision to keep it. if previous == CASE_DRAFT and case.status != CASE_DRAFT: case.reviewed_by_person_id = actor_person_id case.reviewed_at = _naive_now() await db.flush() return case async def delete_case(db: AsyncSession, case: AssistantEvalCase) -> dict[str, Any]: """Remove a case, or retire it when history would be lost. A case with results is never deleted: the rows recording how the assistant used to answer it are the trend line, and deleting the question would take them with it. The caller is told which of the two happened rather than being left to assume. """ used = ( await db.execute( select(func.count()) .select_from(AssistantEvalResult) .where(AssistantEvalResult.case_id == case.id) ) ).scalar() or 0 if used: case.status = CASE_RETIRED await db.flush() return {"deleted": False, "retired": True, "results": int(used)} await db.execute( delete(AssistantEvalCase).where(AssistantEvalCase.id == case.id) ) return {"deleted": True, "retired": False, "results": 0} # --------------------------------------------------------------------------- # Runs # --------------------------------------------------------------------------- async def start_run( db: AsyncSession, *, suite_id: uuid.UUID, repeats: int = DEFAULT_REPEATS, concurrency: int = DEFAULT_CONCURRENCY, trigger: str = TRIGGER_MANUAL, actor_person_id: uuid.UUID | None = None, trace_id: str | None = None, ) -> AssistantEvalRun: """Create the run row. Execution is the caller's next move. Refuses when the same suite already has a run in flight: two concurrent runs of one suite produce two half-measurements and double the provider bill, and the second one is almost always a double-click. """ suite = await db.get(AssistantEvalSuite, suite_id) if suite is None: raise EvalRunError("ASSISTANT_EVAL_SUITE_NOT_FOUND", "题库不存在。") in_flight = ( await db.execute( select(AssistantEvalRun.id).where( AssistantEvalRun.suite_id == suite_id, AssistantEvalRun.status == RUN_RUNNING, ) ) ).first() if in_flight: raise EvalRunError( "ASSISTANT_EVAL_RUN_IN_FLIGHT", "这个题库已经有一次评测在跑了。等它结束,或者先取消它。", ) active = ( await db.execute( select(func.count()) .select_from(AssistantEvalCase) .where( AssistantEvalCase.suite_id == suite_id, AssistantEvalCase.status == CASE_ACTIVE, ) ) ).scalar() or 0 if not active: raise EvalRunError( "ASSISTANT_EVAL_NO_ACTIVE_CASES", "这个题库里没有已启用的用例。先把草稿用例审过再跑。", ) run = AssistantEvalRun( suite_id=suite_id, status=RUN_RUNNING, trigger=trigger, triggered_by_person_id=actor_person_id, trace_id=trace_id, repeats=max(1, min(int(repeats), MAX_REPEATS)), concurrency=max(1, min(int(concurrency), MAX_CONCURRENCY)), cases_total=min(int(active), MAX_CASES_PER_RUN), ) db.add(run) await db.flush() return run def serialize_run(run: AssistantEvalRun, *, suite_name: str | None = None) -> dict[str, Any]: return { "id": str(run.id), "suite_id": str(run.suite_id), "suite_name": suite_name, "status": run.status, "trigger": run.trigger, "repeats": run.repeats, "concurrency": run.concurrency, "provider": run.provider, "model": run.model, "judge_model": run.judge_model, "cases_total": run.cases_total, "cases_run": run.cases_run, "cases_flaky": run.cases_flaky, "attempts_total": run.attempts_total, "attempts_passed": run.attempts_passed, "attempts_failed": run.attempts_failed, "attempts_errored": run.attempts_errored, "pass_rate": run.pass_rate, "dimension_means": run.dimension_means or {}, "category_breakdown": run.category_breakdown or {}, "failure_breakdown": run.failure_breakdown or {}, # Why the unscored attempts were unscored. Separate from the # failure breakdown on purpose — see the column comment. "error_breakdown": run.error_breakdown or {}, "notes": run.notes, "started_at": _iso(run.started_at), "finished_at": _iso(run.finished_at), } async def list_runs( db: AsyncSession, *, suite_id: uuid.UUID | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: stmt = select(AssistantEvalRun, AssistantEvalSuite.name).join( AssistantEvalSuite, AssistantEvalSuite.id == AssistantEvalRun.suite_id ) count_stmt = select(func.count()).select_from(AssistantEvalRun) if suite_id: stmt = stmt.where(AssistantEvalRun.suite_id == suite_id) count_stmt = count_stmt.where(AssistantEvalRun.suite_id == suite_id) total = (await db.execute(count_stmt)).scalar() or 0 rows = ( await db.execute( stmt.order_by(AssistantEvalRun.started_at.desc()) .offset(max(0, (page - 1) * page_size)) .limit(page_size) ) ).all() return [serialize_run(row[0], suite_name=row[1]) for row in rows], total def serialize_result(result: AssistantEvalResult, *, full: bool = False) -> dict[str, Any]: data: dict[str, Any] = { "id": str(result.id), "case_id": str(result.case_id), "attempt": result.attempt, "question": result.question, "category": result.category, "lang": result.lang, "verdict": result.verdict, "failure_type": result.failure_type, "scores": result.scores or {}, "provider": result.provider, "latency_ms": result.latency_ms, "tool_calls": result.tool_calls or [], "findings": result.findings or [], "created_at": _iso(result.created_at), } if full: data.update( { "reply": result.reply, "sources": result.sources or [], "proposed_actions": result.proposed_actions or [], "policy_conflicts": result.policy_conflicts or [], "judge_reasoning": result.judge_reasoning, "judge_model": result.judge_model, "error": result.error, "persona_person_id": ( str(result.persona_person_id) if result.persona_person_id else None ), } ) else: # The list view shows enough of the reply to recognise the answer # without shipping a megabyte of transcripts to render one table. data["reply_excerpt"] = (result.reply or "")[:240] return data async def list_results( db: AsyncSession, *, run_id: uuid.UUID, verdict: str | None = None, category: str | None = None, failure_type: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[dict[str, Any]], int]: stmt = select(AssistantEvalResult).where(AssistantEvalResult.run_id == run_id) count_stmt = ( select(func.count()) .select_from(AssistantEvalResult) .where(AssistantEvalResult.run_id == run_id) ) conditions = [] if verdict: conditions.append(AssistantEvalResult.verdict == verdict) if category: conditions.append(AssistantEvalResult.category == category) if failure_type: conditions.append(AssistantEvalResult.failure_type == failure_type) for condition in conditions: stmt = stmt.where(condition) count_stmt = count_stmt.where(condition) total = (await db.execute(count_stmt)).scalar() or 0 rows = ( ( await db.execute( stmt.order_by( # Failures first: a results table whose first screen is # thirty passes is a table nobody scrolls. # # ASCENDING, and that is not a typo. The column holds the # literals ``pass`` / ``fail`` / ``error``, so DESCENDING # string order is ``pass > fail > error`` — exactly the # thirty passes this comment says it is preventing, which is # what it shipped as. Ascending gives error, fail, pass. AssistantEvalResult.verdict.asc(), AssistantEvalResult.created_at.asc(), ) .offset(max(0, (page - 1) * page_size)) .limit(page_size) ) ) .scalars() .all() ) return [serialize_result(row) for row in rows], total async def trend( db: AsyncSession, *, suite_id: uuid.UUID, limit: int = 20 ) -> list[dict[str, Any]]: """The last N terminal runs of one suite, oldest first, for the chart.""" rows = ( ( await db.execute( select(AssistantEvalRun) .where( AssistantEvalRun.suite_id == suite_id, AssistantEvalRun.status.in_(RUN_TERMINAL_STATUSES), AssistantEvalRun.pass_rate.is_not(None), ) .order_by(AssistantEvalRun.started_at.desc()) .limit(max(1, min(limit, 100))) ) ) .scalars() .all() ) points = [ { "run_id": str(run.id), "date": _iso(run.started_at), "pass_rate": run.pass_rate, "attempts": run.attempts_total, "flaky": run.cases_flaky, "trigger": run.trigger, "model": run.model, **{f"dim_{k}": v for k, v in (run.dimension_means or {}).items()}, } for run in rows ] points.reverse() return points async def previous_terminal_run( db: AsyncSession, *, run: AssistantEvalRun ) -> AssistantEvalRun | None: return ( await db.execute( select(AssistantEvalRun) .where( AssistantEvalRun.suite_id == run.suite_id, AssistantEvalRun.id != run.id, AssistantEvalRun.status.in_(RUN_TERMINAL_STATUSES), AssistantEvalRun.pass_rate.is_not(None), AssistantEvalRun.started_at < run.started_at, ) .order_by(AssistantEvalRun.started_at.desc()) .limit(1) ) ).scalar_one_or_none() async def compare_runs( db: AsyncSession, *, run: AssistantEvalRun, baseline: AssistantEvalRun ) -> dict[str, Any]: """What moved between two runs, per case. Per-case rather than headline-only because the headline is not actionable: "down four points" starts an investigation, "these three cases went from passing to failing, and here is what the assistant said" ends one. """ async def _by_case(target: AssistantEvalRun) -> dict[uuid.UUID, dict[str, Any]]: rows = ( ( await db.execute( select(AssistantEvalResult).where( AssistantEvalResult.run_id == target.id ) ) ) .scalars() .all() ) grouped: dict[uuid.UUID, dict[str, Any]] = {} for row in rows: bucket = grouped.setdefault( row.case_id, {"question": row.question, "category": row.category, "verdicts": []}, ) bucket["verdicts"].append(row.verdict) for bucket in grouped.values(): scored = [v for v in bucket["verdicts"] if v != "error"] bucket["passed"] = sum(1 for v in scored if v == "pass") bucket["scored"] = len(scored) bucket["rate"] = ( bucket["passed"] / bucket["scored"] if bucket["scored"] else None ) return grouped current = await _by_case(run) before = await _by_case(baseline) regressed: list[dict[str, Any]] = [] improved: list[dict[str, Any]] = [] for case_id, now in current.items(): then = before.get(case_id) if then is None or now["rate"] is None or then["rate"] is None: continue if now["rate"] < then["rate"]: regressed.append( { "case_id": str(case_id), "question": now["question"], "category": now["category"], "before": round(then["rate"] * 100, 1), "after": round(now["rate"] * 100, 1), } ) elif now["rate"] > then["rate"]: improved.append( { "case_id": str(case_id), "question": now["question"], "category": now["category"], "before": round(then["rate"] * 100, 1), "after": round(now["rate"] * 100, 1), } ) regressed.sort(key=lambda item: item["after"] - item["before"]) improved.sort(key=lambda item: item["before"] - item["after"]) # 🔴 The headline drop is computed over the cases BOTH runs actually # SCORED — not from the two headline numbers. # # Two ways the headline comparison lied. **Selection bias**: a run whose # budget runs out covers a PREFIX of the suite (``_pending_attempts`` is # case-major on purpose), and a partial run is both alertable and eligible # as a baseline — so a run that reached only the eight hardest retrieval # cases "dropped 45 points" against a complete baseline with nothing having # changed. **Error inflation**: ``pass_rate`` correctly excludes errored # attempts from its denominator, but the comparability floor counted # ``attempts_total``, which includes them — so a provider outage that # errored 28 of 30 attempts read as "comparable" on the strength of two # samples, during exactly the outage the error exclusion exists to keep off # the trend line. shared = [ (current[case_id], before[case_id]) for case_id in current.keys() & before.keys() if current[case_id]["rate"] is not None and before[case_id]["rate"] is not None ] now_scored = sum(item[0]["scored"] for item in shared) then_scored = sum(item[1]["scored"] for item in shared) now_rate = ( round(sum(item[0]["passed"] for item in shared) / now_scored * 100, 2) if now_scored else None ) then_rate = ( round(sum(item[1]["passed"] for item in shared) / then_scored * 100, 2) if then_scored else None ) drop = ( round(then_rate - now_rate, 2) if now_rate is not None and then_rate is not None else None ) return { "run_id": str(run.id), "baseline_run_id": str(baseline.id), # The headline numbers stay, for display. Each is what that run scored # over whatever it covered — and neither is what ``drop_pp`` measures. "pass_rate": run.pass_rate, "baseline_pass_rate": baseline.pass_rate, # What the drop IS drawn from, so a reader can see the base it rests on. "shared_cases": len(shared), "shared_pass_rate": now_rate, "shared_baseline_pass_rate": then_rate, "shared_scored": now_scored, "shared_baseline_scored": then_scored, "drop_pp": drop, # A comparison drawn from a handful of SCORED attempts on a handful of # SHARED cases can "drop 20 points" with nothing having changed. # Callers that alert must check this. "comparable": ( now_scored >= REGRESSION_MIN_ATTEMPTS and then_scored >= REGRESSION_MIN_ATTEMPTS ), "regressed": regressed[:20], "improved": improved[:20], "model_changed": (run.model or "") != (baseline.model or ""), } # --------------------------------------------------------------------------- # Mining the bank from production # --------------------------------------------------------------------------- #: Redaction applied to every mined question before it is stored. These are #: real employees' words: a mined case may name a colleague, quote an address, #: or carry a token somebody pasted into chat. The reviewer sees the redacted #: form, and the redaction happens at mining time rather than at display time — #: a row that never held the value cannot leak it later. _REDACTIONS: tuple[tuple[re.Pattern[str], str], ...] = ( (re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"), "[邮箱]"), (re.compile(r"(? str: out = text or "" for pattern, replacement in _REDACTIONS: out = pattern.sub(replacement, out) return out def _normalise_question(text: str) -> str: """Collapse a question to a dedup key. Not a display value.""" return re.sub(r"[\s\W_]+", "", (text or "").casefold()) async def mine_candidates( db: AsyncSession, *, suite_id: uuid.UUID, days: int = 30, limit: int = 50, actor_person_id: uuid.UUID | None = None, ) -> dict[str, Any]: """Turn recent production into draft cases. Three sources, all of which are questions somebody actually asked: * employee turns from the Feishu transcript — what the assistant is really being used for, as opposed to what its authors imagined; * answers an employee marked unhelpful — the assistant's own failures; * deflection queries that hit an empty knowledge base — questions the product cannot answer at all. Everything lands as ``draft``. A person promotes it, and that review is where the asker's data comes out — the redaction here is a floor, not a substitute for reading it. """ from app.assistant.models import ( # noqa: PLC0415 ROLE_USER, AssistantMessage, AssistantMessageFeedback, ) from app.domains.support.models import SupportDeflectionEvent # noqa: PLC0415 since = _naive_now() - timedelta(days=max(1, min(days, 365))) existing = { _normalise_question(row[0]) for row in ( await db.execute( select(AssistantEvalCase.question).where( AssistantEvalCase.suite_id == suite_id ) ) ).all() } candidates: list[tuple[str, str, str]] = [] # (question, origin, ref) unhelpful_ids = { row[0] for row in ( await db.execute( select(AssistantMessageFeedback.message_id).where( AssistantMessageFeedback.verdict == "unhelpful", AssistantMessageFeedback.created_at >= since, ) ) ).all() } if unhelpful_ids: # The message that was rated is the ASSISTANT's answer; the question is # the user turn immediately before it in the same conversation. answers = ( ( await db.execute( select(AssistantMessage).where( AssistantMessage.id.in_(unhelpful_ids) ) ) ) .scalars() .all() ) for answer in answers: asked = ( await db.execute( select(AssistantMessage) .where( AssistantMessage.conversation_id == answer.conversation_id, AssistantMessage.role == ROLE_USER, AssistantMessage.seq < answer.seq, ) .order_by(AssistantMessage.seq.desc()) .limit(1) ) ).scalar_one_or_none() if asked and asked.content.strip(): candidates.append((asked.content, ORIGIN_GAP, f"message:{asked.id}")) turns = ( ( await db.execute( select(AssistantMessage) .where( AssistantMessage.role == ROLE_USER, AssistantMessage.created_at >= since, ) .order_by(AssistantMessage.created_at.desc()) .limit(600) ) ) .scalars() .all() ) for turn in turns: if turn.content and turn.content.strip(): candidates.append((turn.content, ORIGIN_MINED, f"message:{turn.id}")) deflections = ( ( await db.execute( select(SupportDeflectionEvent) .where( SupportDeflectionEvent.created_at >= since, or_( SupportDeflectionEvent.kb_article_ids.is_(None), func.json_length(SupportDeflectionEvent.kb_article_ids) == 0, ), ) .order_by(SupportDeflectionEvent.created_at.desc()) .limit(200) ) ) .scalars() .all() ) for event in deflections: if event.query and event.query.strip(): candidates.append((event.query, ORIGIN_GAP, f"deflection:{event.id}")) created = 0 skipped_duplicate = 0 seen: set[str] = set(existing) ordinal = ( await db.execute( select(func.coalesce(func.max(AssistantEvalCase.ordinal), 0)).where( AssistantEvalCase.suite_id == suite_id ) ) ).scalar() or 0 for raw_question, origin, ref in candidates: if created >= max(1, min(limit, 200)): break question = redact(raw_question.strip())[:1000] if len(question) < 4: continue key = _normalise_question(question) if key in seen: skipped_duplicate += 1 continue seen.add(key) ordinal += 1 db.add( AssistantEvalCase( suite_id=suite_id, ordinal=ordinal, question=question, lang=_guess_lang(question), category=CATEGORIES[0], status=CASE_DRAFT, origin=origin, origin_ref=ref, created_by_person_id=actor_person_id, expectations={ "notes": ( "从生产语料挖出来的草稿。请先确认:①问题里没有个人信息;" "②分类对不对;③写一句参考答案再启用。" ) }, ) ) created += 1 await db.flush() return { "created": created, "skipped_duplicate": skipped_duplicate, "candidates_seen": len(candidates), "window_days": days, } _KANA = re.compile("[\\u3040-\\u309f\\u30a0-\\u30ff]") _CJK = re.compile("[\\u4e00-\\u9fff]") def _guess_lang(text: str) -> str: if _KANA.search(text): return "ja" if _CJK.search(text): return "zh" return "en" async def promote_failures( db: AsyncSession, *, run_id: uuid.UUID, suite_id: uuid.UUID, actor_person_id: uuid.UUID | None = None, ) -> dict[str, Any]: """Copy this run's failing questions into a suite as drafts. The loop that makes the harness improve the product rather than only describe it: a failure becomes a permanent case, so the fix has something to be measured against and cannot silently regress later. """ rows = ( ( await db.execute( select(AssistantEvalResult).where( AssistantEvalResult.run_id == run_id, AssistantEvalResult.verdict == VERDICT_FAIL, ) ) ) .scalars() .all() ) existing = { _normalise_question(row[0]) for row in ( await db.execute( select(AssistantEvalCase.question).where( AssistantEvalCase.suite_id == suite_id ) ) ).all() } ordinal = ( await db.execute( select(func.coalesce(func.max(AssistantEvalCase.ordinal), 0)).where( AssistantEvalCase.suite_id == suite_id ) ) ).scalar() or 0 created = 0 for row in rows: key = _normalise_question(row.question) if key in existing: continue existing.add(key) ordinal += 1 db.add( AssistantEvalCase( suite_id=suite_id, ordinal=ordinal, question=row.question, lang=row.lang, category=row.category, status=CASE_DRAFT, origin=ORIGIN_EVAL_FAILURE, origin_ref=f"result:{row.id}", created_by_person_id=actor_person_id, expectations={ "notes": ( f"由评测失败提升而来(判定 {row.failure_type or '未分类'})。" f"当时评委的说法:{(row.judge_reasoning or '')[:600]}" ) }, ) ) created += 1 await db.flush() return {"created": created, "failures_seen": len(rows)} async def generate_variants( db: AsyncSession, *, case: AssistantEvalCase, count: int = 4, actor_person_id: uuid.UUID | None = None, ) -> dict[str, Any]: """Ask the model for other ways an employee might say the same thing. Paraphrase only. The variants inherit the original's category, language intent and expectations, because they are supposed to be the SAME question — if a variant needs a different expected answer it is a different case and a person should write it. Everything lands as ``draft``: a generated question that nobody read is a gate nobody agreed to. """ from app.domains.support.copilot import one_shot_json # noqa: PLC0415 system = ( "你在为一个企业 IT 助手的评测集造同义问法。给定一句员工的提问," "给出同一件事的其他真实说法。要求:①意思完全不变;②像人在飞书里打字," "可以口语、可以简写、可以有错别字;③覆盖不同长度;" "④如果原句是中文,至少给一条英文说法,反之亦然。" "不要改变提问的对象和范围,不要引入原句没有的信息。" ) prompt = ( f"原句:{case.question}\n" f"语言:{case.lang}\n\n" f'请输出 JSON:{{"variants":["…", "…"]}},最多 {max(1, min(count, 8))} 条。' ) payload = await one_shot_json(db, system, prompt, timeout=45.0, max_tokens=800) if not payload: return {"created": 0, "available": False} raw = payload.get("variants") variants = [ str(item).strip() for item in (raw if isinstance(raw, list) else []) if str(item).strip() ][: max(1, min(count, 8))] if not variants: return {"created": 0, "available": True} existing = { _normalise_question(row[0]) for row in ( await db.execute( select(AssistantEvalCase.question).where( AssistantEvalCase.suite_id == case.suite_id ) ) ).all() } ordinal = ( await db.execute( select(func.coalesce(func.max(AssistantEvalCase.ordinal), 0)).where( AssistantEvalCase.suite_id == case.suite_id ) ) ).scalar() or 0 created = 0 for variant in variants: key = _normalise_question(variant) if key in existing: continue existing.add(key) ordinal += 1 inherited = dict(case.expectations or {}) inherited["notes"] = ( f"由「{case.question[:60]}」生成的同义问法。" + str(inherited.get("notes") or "") )[:4000] db.add( AssistantEvalCase( suite_id=case.suite_id, ordinal=ordinal, question=variant[:4000], setup_turns=case.setup_turns, lang=_guess_lang(variant), category=case.category, status=CASE_DRAFT, origin=ORIGIN_GENERATED, origin_ref=f"case:{case.id}", persona_person_id=case.persona_person_id, expectations=inherited, created_by_person_id=actor_person_id, ) ) created += 1 await db.flush() return {"created": created, "available": True, "proposed": len(variants)} __all__ = [ "compare_runs", "create_case", "delete_case", "generate_variants", "list_cases", "list_results", "list_runs", "list_suites", "mine_candidates", "previous_terminal_run", "promote_failures", "redact", "seed_builtin_suites", "serialize_case", "serialize_result", "serialize_run", "start_run", "trend", "update_case", ]