"""知识循环:提案队列 + 知识库的目录列与互链。 **要修的是什么。** 知识库此前只会长大,不会变好。每一个新知识的来源 —— 工单 结案、服务台 FAQ 导入、飞书 wiki 镜像 —— 产出的都是同一件东西:一篇**全新的 草稿**。没有任何东西会去更新已有的条目、合并两条讲同一件事的、补上员工真正会 用的那个词,或者发现两条条目互相矛盾。 这正是这套语料已知的召回天花板背后的机制:整个语料写「电脑」从不写「笔记本」, 所以「笔记本电脑坏了」只召回一条勉强相关的。那**不是排序算法的问题**,修法是 有人把那个词加进去 —— 而在此之前,系统里没有任何地方能承载「该加这个词」这件事。 **三张变更:** ① `knowledge_proposals`(core 拥有,不属于任何一个域)—— 机器提出的每一个改动 都落在这里,由人决定。放 core 是因为安全答案库之后也要用同一个队列,而域之间 不互相 import;两份表意味着两个评审页、同一个决定有两套说法、一个 bug 要修两遍。 队列的去重由**数据库**保证,不是先读后写:`open_fingerprint` 在提案打开时 等于指纹、决定之后置 NULL,而 MySQL 的唯一索引允许任意多个 NULL。于是历史可 以无限累积,而「同一个建议同时开着两条」不可能发生。先读后写会输 —— 每周的 lint 和一次工单结案可以在同一秒里得出同一个结论。 ② `support_kb_articles.summary` —— 一句话说明这条条目是干什么用的。它的存在是 为了让语料能被当成一个**列表**来推理:判断新知识该进已有条目还是新开一条、发现 两条讲的是同一件事、找出没有别名的条目,这些都需要一百条同时在眼前 —— 而一百 篇正文放不进任何地方。可空,读的地方一律回落到正文开头,因为目录必须对**今天 的语料**可用,而不是只对跑过摘要之后的语料可用。 ③ `support_kb_article_links` —— 条目之间的有类型的边。此前语料是一堆互不相连的 文档:找到对的那条的人没有办法到达它旁边那条,一次合并没有地方记录它吸收了什么, 两条互相矛盾的条目只有同时读过两篇的人才可能发现。对称的关系(related / contradicts)**只存一行**,方向规范化 —— 两行表示一个事实,就是两行可以互相 矛盾,而只查一个方向的读者会看到半张图。 Revision ID: 593 Revises: 592 ⚠️ 迁移号要对着 `origin/main` 抢,而且每次 push 之前重新抢一次 —— 本仓前几次 踩它是 462 / 472 / 582 / 587。 """ 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 = "593" down_revision = "592" branch_labels = None depends_on = None PROPOSALS = "knowledge_proposals" JOB_KEY = "knowledge.loop_lint" LINKS = "support_kb_article_links" ARTICLES = "support_kb_articles" 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 _has_column(bind, table: str, column: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.columns " "WHERE table_schema = DATABASE() " "AND table_name = :t AND column_name = :c" ), {"t": table, "c": column}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() person_fk = id_type_for("identity_persons") if not _has_table(bind, PROPOSALS): op.create_table( PROPOSALS, sa.Column("id", sa.String(36), primary_key=True), # 语料名写在行上,不从 target_ref 反推:一条提案要能不 join 就读懂, # 评审界面也要能在不知道 target_ref 指向什么表的情况下筛选。 sa.Column("corpus", sa.String(32), nullable=False), sa.Column("kind", sa.String(24), nullable=False), # 刻意是**字符串**而不是多态外键:真外键只能指向一张表,而这一行 # 存在的全部意义就是与语料无关。代价是目标可能在提案之下消失 —— # 所以有 superseded 这个终态,而且落地前一定重读目标。 sa.Column("target_ref", sa.String(64)), sa.Column("related_ref", sa.String(64)), sa.Column("title", sa.String(255), nullable=False), sa.Column("payload", sa.JSON()), sa.Column("rationale", sa.Text()), sa.Column("evidence", sa.JSON()), sa.Column("source", sa.String(24), nullable=False), sa.Column("source_ref", sa.String(128)), sa.Column("fingerprint", sa.String(64), nullable=False), # 打开时等于 fingerprint,决定之后置 NULL —— 见下面的唯一约束。 sa.Column("open_fingerprint", sa.String(64)), sa.Column( "status", sa.String(16), nullable=False, server_default="open", ), sa.Column( "decided_by_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column("decided_at", sa.DateTime(timezone=True)), sa.Column("decision_note", sa.Text()), sa.Column("applied_ref", sa.String(64)), sa.Column("error", sa.Text()), 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(), ), # 队列里同一个建议最多开着一条。MySQL 的唯一索引允许任意多个 NULL, # 所以历史(已接受 / 已拒绝 / 已失效)可以无限累积。 sa.UniqueConstraint( "corpus", "open_fingerprint", name="uq_knowledge_proposal_open", ), ) op.create_index( "idx_knowledge_proposal_queue", PROPOSALS, ["corpus", "status", "created_at"], ) op.create_index( "idx_knowledge_proposal_target", PROPOSALS, ["corpus", "target_ref"], ) op.create_index( "idx_knowledge_proposal_source", PROPOSALS, ["source", "status"], ) if not _has_column(bind, ARTICLES, "summary"): op.add_column(ARTICLES, sa.Column("summary", sa.String(255), nullable=True)) if not _has_table(bind, LINKS): op.create_table( LINKS, sa.Column("id", sa.String(36), primary_key=True), sa.Column( "from_article_id", id_type_for(ARTICLES), sa.ForeignKey(f"{ARTICLES}.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "to_article_id", id_type_for(ARTICLES), sa.ForeignKey(f"{ARTICLES}.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "kind", sa.String(24), nullable=False, server_default="related", ), sa.Column("note", sa.Text()), 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.UniqueConstraint( "from_article_id", "to_article_id", "kind", name="uq_support_kb_link", ), ) op.create_index( "idx_support_kb_link_from", LINKS, ["from_article_id", "kind"], ) op.create_index( "idx_support_kb_link_to", LINKS, ["to_article_id", "kind"], ) 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, '10 4 * * 1', 'Asia/Shanghai', 1, :config, 0, 0, UTC_TIMESTAMP(), UTC_TIMESTAMP()) """ ), { # 带横线的 uuid4。存成 32 位无横线的话 GUID TypeDecorator 读回来 # 永远匹配不上,handler 在 run 包装器外面就抛 —— scheduled_job_runs # 一行都不会写。本仓已经踩过三次(059 / 162 / 248)。 "id": str(uuid.uuid4()), "name": JOB_KEY, "display_name": "知识库 · 每周自检与回填", "description": ( "把已发布语料两两比一遍(重复 / 冲突 / 该互相关联)、把零命中的" "员工问句回接到本该回答它的条目上得出该补的说法、清理指向已下架" "条目的关联,并把员工点过「有帮助」的助手回答提回语料。" "全部产出为待人工确认的提案,不自动改动任何已发布内容。" "与既有的知识库缺口周报互补而不重复:那一条回答「该写哪篇新文档」," "这一条回答「已有的这些哪里不对」。" ), # 周一 04:10(上海)—— 排在助手回归评测(03:30)之后,这样评测那一轮 # 判出来的语料类失败,当天就能被人工提回队列。 "config": json.dumps( {"window_days": 14, "summarise_limit": 25, "backfill_limit": 10} ), }, ) 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}, ) if _has_table(bind, LINKS): op.drop_table(LINKS) if _has_column(bind, ARTICLES, "summary"): op.drop_column(ARTICLES, "summary") if _has_table(bind, PROPOSALS): op.drop_table(PROPOSALS)