"""助手评测台:题库 / 运行 / 每次尝试的结果。 飞书里的 Helix 助手已经跑在生产上,但它答得好不好,一直只有两个来源: 员工按的那个 👎,和有人偶尔在管理页手打一句话试一下。前者只覆盖"糟到有人 愿意点一下"的那一小撮,后者不可重复、不可比较、也从来没有人记下来过。 换句话说:**这个产品的质量此前没有被测量过**,只被投诉过。 这次建四张表,把"测量"变成一件可重复的事: suites ──< cases 规格:我们期望助手能处理什么 │ └──< runs ──< results 测量:每问一次,它做了什么 四张表而不是两张,理由分别是: **cases 和 results 必须分开,而且 results 要冗余存一份题面。** 用例会被人改(措辞、期望、分类都会改),而一次历史运行测的是"当时那句话"。 不冗余的话,今天改一个字,三个月前那次运行的记录就悄悄变成在测别的东西了。 **results 存"每一次尝试",不存"每条用例的汇总"。** 同一句话问三遍可能得到两种结果 —— 这个事实是这套东西最有价值的产出, 而汇总行会把它平均掉。`(run_id, case_id, attempt)` 上有唯一约束:一次运行 被中断后重跑,是靠"减掉已经存在的 (case, attempt)"来续跑的,这个约束就是 续跑不会重复计数的保证。 **回答正文整段存下来。** 本仓其它与助手有关的台账都只存"形状"不存原话 (见 `gap_verification.TurnShape`),因为那些行是真实员工的提问,提问是他们 自己的数据。这里不是:评测用例是 IT 自己写下来的题,以模拟身份、走干跑 路径提问 —— 既发不出卡片也建不了申请。要判一个回答好不好就得读它,所以存。 唯一的例外是从生产语料挖出来的问句,它们一律以 `draft` 落库,由人审过才 启用,而那次审阅就是把别人的话摘出去的地方(挖的时候已经先做了一遍脱敏)。 **pass_rate 的分母不含 error。** 模型后端抖一下不是"助手答得差";把它算进 失败,一次故障就会在趋势线上长得和一次质量回归一模一样,而这两件事要找的 人完全不同。 顺带播两个内置题库(`regression` / `broad`)—— 真正的用例由 `app.assistant.evaluation.service.seed_builtin_suites` 在应用侧播,不在这里 硬编码 SQL:那些题面带多轮对话和结构化期望,写成 INSERT 既读不懂也维护不了。 这里只保证表和 cron 存在。 再播一个每周的回归 cron(`assistant.eval_regression`)。 Revision ID: 592 Revises: 591 ⚠️ 迁移号要对着 `origin/main` 抢,而且每次 push 之前重新抢一次 —— 本仓前几次 踩它是 462 / 472 / 582 / 587,两个分支同时拿了同一个号,合进 main 之后 链上出现两个 head,后端每个测试分片和镜像构建一起红。 """ from __future__ import annotations import json import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for revision = "592" down_revision = "591" branch_labels = None depends_on = None SUITES = "assistant_eval_suites" CASES = "assistant_eval_cases" RUNS = "assistant_eval_runs" RESULTS = "assistant_eval_results" JOB_KEY = "assistant.eval_regression" _JOB_DESC = ( "每周把回归题库跑一遍,拿模拟员工身份走助手的生产路径(干跑,发不出卡片、" "建不了申请),再由模型评委逐条判分,和上一次终态运行对比。" "通过率掉超过 10 个百分点就告警 IT,并点名是哪几条用例从通过变成了不通过。" "同一条用例默认问 3 遍 —— 一遍问出来的通过率是样本量为 1 的估计," "而「这条用例本身就时好时坏」是这套东西最有价值的产出,一遍看不见。" ) def _has_table(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :n" ), {"n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() person_fk = id_type_for("identity_persons") if not _has_table(bind, SUITES): op.create_table( SUITES, sa.Column("id", sa.String(36), primary_key=True), sa.Column("slug", sa.String(64), nullable=False), sa.Column("name", sa.String(128), nullable=False), sa.Column("description", sa.Text()), sa.Column( "status", sa.String(16), nullable=False, server_default="active", ), sa.Column( "is_builtin", sa.Boolean(), nullable=False, server_default="0", ), # 助手的工具是按调用者作用域的,所以"以谁的身份问"是测量的一部分, # 不是实现细节:同一句话,经理问和新人问,正确答案不一样。把它钉在 # 题库上,相隔几个月的两次运行才可比 —— 不钉的话,兜底员工一换, # 基线就悄悄挪了。 sa.Column( "default_persona_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column( "created_by_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint("slug", name="uq_assistant_eval_suite_slug"), ) op.create_index( "idx_assistant_eval_suite_status", SUITES, ["status"], ) if not _has_table(bind, CASES): op.create_table( CASES, sa.Column("id", sa.String(36), primary_key=True), sa.Column( "suite_id", id_type_for(SUITES), sa.ForeignKey(f"{SUITES}.id", ondelete="CASCADE"), nullable=False, ), sa.Column("ordinal", sa.Integer(), nullable=False, server_default="0"), sa.Column("question", sa.Text(), nullable=False), # 这一轮之前说过的话。生产上的聊天是多轮的,「这台」的所指在上一条 # 消息里 —— 只会问单句的评测测不到产品真正在被用的那个形状。 sa.Column("setup_turns", sa.JSON()), sa.Column("lang", sa.String(8), nullable=False, server_default="zh"), sa.Column("category", sa.String(32), nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default="draft", ), sa.Column( "origin", sa.String(24), nullable=False, server_default="curated", ), sa.Column("origin_ref", sa.String(128)), sa.Column( "persona_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), # 规格里可以被机器逐字核对的那一半:必须/不得出现的说法、必须调用的 # 工具、该不该给动作卡片、该不该转工单,以及给评委看的参考答案。 sa.Column("expectations", sa.JSON()), sa.Column( "created_by_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column( "reviewed_by_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column("reviewed_at", sa.DateTime(timezone=True)), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), ) op.create_index( "idx_assistant_eval_case_suite", CASES, ["suite_id", "status"], ) op.create_index( "idx_assistant_eval_case_category", CASES, ["category"], ) op.create_index( "idx_assistant_eval_case_origin", CASES, ["origin", "status"], ) if not _has_table(bind, RUNS): op.create_table( RUNS, sa.Column("id", sa.String(36), primary_key=True), sa.Column( "suite_id", id_type_for(SUITES), sa.ForeignKey(f"{SUITES}.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "status", sa.String(16), nullable=False, server_default="running", ), sa.Column( "trigger", sa.String(16), nullable=False, server_default="manual", ), sa.Column( "triggered_by_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column("trace_id", sa.String(64)), sa.Column("repeats", sa.Integer(), nullable=False, server_default="1"), sa.Column( "concurrency", sa.Integer(), nullable=False, server_default="1", ), # 换了供应商/模型是趋势线上出现台阶最可能的解释。记在运行上, # 就不用把一次换模型当成质量回归去查。 sa.Column("provider", sa.String(32)), sa.Column("model", sa.String(64)), sa.Column("judge_provider", sa.String(32)), sa.Column("judge_model", sa.String(64)), sa.Column("cases_total", sa.Integer(), nullable=False, server_default="0"), sa.Column("cases_run", sa.Integer(), nullable=False, server_default="0"), sa.Column("cases_flaky", sa.Integer(), nullable=False, server_default="0"), sa.Column( "attempts_total", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "attempts_passed", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "attempts_failed", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "attempts_errored", sa.Integer(), nullable=False, server_default="0", ), sa.Column("pass_rate", sa.Float()), sa.Column("dimension_means", sa.JSON()), sa.Column("category_breakdown", sa.JSON()), sa.Column("failure_breakdown", sa.JSON()), sa.Column("notes", sa.Text()), sa.Column( "started_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("finished_at", sa.DateTime(timezone=True)), ) op.create_index( "idx_assistant_eval_run_suite", RUNS, ["suite_id", "started_at"], ) op.create_index("idx_assistant_eval_run_status", RUNS, ["status"]) if not _has_table(bind, RESULTS): op.create_table( RESULTS, sa.Column("id", sa.String(36), primary_key=True), sa.Column( "run_id", id_type_for(RUNS), sa.ForeignKey(f"{RUNS}.id", ondelete="CASCADE"), nullable=False, ), # 刻意**不**级联删除:退役一条用例不能把"助手过去是怎么答它的" # 一起抹掉,那些行就是趋势线本身。用例只退役、不删除(服务层在 # 有结果行时会自动改成退役)。 sa.Column( "case_id", id_type_for(CASES), sa.ForeignKey(f"{CASES}.id"), nullable=False, ), sa.Column("attempt", sa.Integer(), nullable=False, server_default="0"), sa.Column("question", sa.Text(), nullable=False), sa.Column("category", sa.String(32), nullable=False), sa.Column("lang", sa.String(8), nullable=False, server_default="zh"), sa.Column( "persona_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column("reply", sa.Text()), sa.Column("tool_calls", sa.JSON()), sa.Column("sources", sa.JSON()), sa.Column("proposed_actions", sa.JSON()), sa.Column("policy_conflicts", sa.JSON()), sa.Column("provider", sa.String(32)), sa.Column("latency_ms", sa.Integer()), sa.Column("verdict", sa.String(16), nullable=False), sa.Column("failure_type", sa.String(32)), sa.Column("scores", sa.JSON()), sa.Column("judge_reasoning", sa.Text()), sa.Column("judge_model", sa.String(64)), sa.Column("findings", sa.JSON()), sa.Column("error", sa.Text()), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint( "run_id", "case_id", "attempt", name="uq_assistant_eval_result_attempt", ), ) op.create_index( "idx_assistant_eval_result_run", RESULTS, ["run_id", "verdict"], ) op.create_index( "idx_assistant_eval_result_case", RESULTS, ["case_id", "created_at"], ) op.create_index( "idx_assistant_eval_result_failure", RESULTS, ["run_id", "failure_type"], ) if not _has_table(bind, "scheduled_jobs"): return exists = bind.execute( sa.text("SELECT 1 FROM scheduled_jobs WHERE name = :name LIMIT 1"), {"name": JOB_KEY}, ).scalar() if exists: return bind.execute( sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (:id, :name, :display_name, :description, :name, 'cron', NULL, '30 3 * * 1', 'Asia/Shanghai', 1, :config, 0, 0, UTC_TIMESTAMP(), UTC_TIMESTAMP()) """ ), { # 必须是带横线的 uuid4。存成 32 位无横线的话,GUID TypeDecorator # 读回来永远匹配不上,handler 在 run 包装器外面就抛了 —— # scheduled_job_runs 一行都不会写,run_count / last_run_at 永远是 # 0 / NULL。本仓已经踩过三次(059 / 162 / 248)。 "id": str(uuid.uuid4()), "name": JOB_KEY, "display_name": "助手 · 每周回归评测", "description": _JOB_DESC, # 周一 03:30(上海)。放在上班之前:这一跑会占用和员工同一个模型 # 后端,不该和早高峰的提问抢。 "config": json.dumps( {"suite": "regression", "repeats": 3, "concurrency": 3} ), }, ) def downgrade() -> None: bind = op.get_bind() if _has_table(bind, "scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE name = :name"), {"name": JOB_KEY}, ) for table in (RESULTS, RUNS, CASES, SUITES): if _has_table(bind, table): op.drop_table(table)