"""The knowledge base as a *wiki* rather than as a pile of documents. Three things this module adds, and the specific failure each one exists to fix: **An index.** ``index()`` is the corpus as a list — one line per page, id plus title plus a one-sentence summary plus its alternate phrasings. Everything that has to reason about the corpus *as a whole* needs this and could not have it before: deciding whether new knowledge belongs on an existing page, spotting that two pages say the same thing, noticing that a page has no aliases. A hundred summaries fit in one prompt; a hundred bodies do not. **Write-back on ingest.** Until now every source of knowledge — a resolved ticket, a service-desk FAQ export, a Feishu wiki page — produced exactly one thing: a brand new draft. Nothing ever updated an existing page. That is the mechanism behind the recall ceiling this corpus is already known to have: production writes 「电脑」 and never 「笔记本」, so 「笔记本电脑坏了」 finds one barely-relevant page — and the fix for that is not a better ranker, it is somebody adding the word. Now ingest first asks *is this already written down*, and the usual answer produces an alias or an appended section rather than a competing page. **Links.** A typed edge between two pages, so a merge can record what it absorbed, a contradiction can be recorded without anyone pretending to know which side is right, and a reader who found the right page can reach the one next to it. **What this module deliberately does NOT do:** put a model hop inside the employee's turn. The Karpathy-style loop reads the index and decides which pages to open, replacing vector search. Here vector search already works and costs one embedding call, while an index-read hop would add a second model round-trip to every question an employee asks. So the index is used on the batch paths (ingest, lint, dedup) where it is free, and retrieval gets the one cheap improvement links make possible — expanding a hit with its related pages, which is a single extra query and no model call. Whether that helps is a question for the evaluation harness, not for an argument. """ from __future__ import annotations import json import logging import time import uuid from dataclasses import dataclass from typing import Any from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.knowledge_loop import service as loop from app.core.knowledge_loop.constants import ( CORPUS_SUPPORT_KB, KIND_ADD_ALIASES, KIND_APPEND_SECTION, KIND_CONTRADICTION, KIND_DEAD_LINK, KIND_LINK, KIND_MERGE, KIND_NEW_ARTICLE, KIND_RETIRE, MAX_CANDIDATES, SUMMARY_MAX_CHARS, ) from app.core.knowledge_loop.registry import ProposalApplyError, register_adapter from app.domains.support.models import ( KB_ARCHIVED, KB_LINK_KINDS, KB_PUBLISHED, KB_SOURCE_FEISHU_DOC, LINK_CONTRADICTS, LINK_RELATED, LINK_SUPERSEDES, SupportKbArticle, SupportKbArticleLink, ) logger = logging.getLogger(__name__) #: Symmetric kinds are stored once, in a canonical direction. Two rows for one #: symmetric fact is two rows that can disagree, and a reader who queries only #: one direction sees half the graph. _SYMMETRIC = (LINK_RELATED, LINK_CONTRADICTS) #: The index is read on every ingest and every lint pass and changes only when #: the corpus does. Same staleness probe as the vector cache: a COUNT plus a #: MAX(updated_at) over the same rows is a few bytes, and every write path that #: could alter the corpus moves one of them. _INDEX_CACHE: dict[str, tuple[tuple[int, str], float, list[dict[str, Any]]]] = {} _INDEX_TTL_SECONDS = 300.0 def reset_index_cache() -> None: _INDEX_CACHE.clear() # --------------------------------------------------------------------------- # Summaries # --------------------------------------------------------------------------- def fallback_summary(article: SupportKbArticle) -> str: """A summary for a page that has not been given one. The index has to work on the corpus as it is today, not only after somebody runs a summarisation pass over 246 articles. The head of the body is a poor summary and an honest one — and because it is derived rather than stored, a page whose real summary arrives later simply starts reading better. """ body = " ".join((article.body or "").split()) return body[:SUMMARY_MAX_CHARS] _SUMMARY_SYSTEM = ( "你在为一个企业 IT 知识库编目录。给定一篇文章的标题和正文," "写一句话说明**这篇文章是干什么用的** —— 让人在一百条的列表里能一眼分辨出" "它和相邻那几条的区别。不超过 60 个字,不要复述标题,不要写「本文介绍了」" "这类套话,不要加任何标点以外的格式。只输出这一句话。" ) async def summarise(db: AsyncSession, article: SupportKbArticle) -> str | None: """Write one line saying what this page is for. ``None`` when unavailable.""" from app.domains.support.copilot import _one_shot_llm # noqa: PLC0415 text = await _one_shot_llm( db, _SUMMARY_SYSTEM, f"标题:{article.title}\n\n正文:\n{(article.body or '')[:4000]}", timeout=30.0, ) if not text: return None line = " ".join(text.split())[:SUMMARY_MAX_CHARS] return line or None async def summarise_missing( db: AsyncSession, *, limit: int = 25, ) -> dict[str, Any]: """Fill in summaries for published pages that have none. Bounded, resumable.""" rows = ( ( await db.execute( select(SupportKbArticle) .where( SupportKbArticle.status == KB_PUBLISHED, SupportKbArticle.summary.is_(None), ) .order_by(SupportKbArticle.updated_at.desc()) .limit(max(1, min(limit, 100))) ) ) .scalars() .all() ) if not rows: return {"pending": 0, "summarised": 0} done = 0 for article in rows: line = await summarise(db, article) if line: article.summary = line done += 1 remaining = ( await db.execute( select(func.count()) .select_from(SupportKbArticle) .where( SupportKbArticle.status == KB_PUBLISHED, SupportKbArticle.summary.is_(None), ) ) ).scalar() or 0 if done: reset_index_cache() return {"pending": int(remaining) - done, "summarised": done} # --------------------------------------------------------------------------- # The index # --------------------------------------------------------------------------- async def index(db: AsyncSession, *, use_cache: bool = True) -> list[dict[str, Any]]: """The published corpus as a list of one-line entries.""" probe = ( await db.execute( select(func.count(), func.max(SupportKbArticle.updated_at)) .select_from(SupportKbArticle) .where(SupportKbArticle.status == KB_PUBLISHED) ) ).one() signature = (int(probe[0] or 0), str(probe[1])) cached = _INDEX_CACHE.get("published") if ( use_cache and cached is not None and cached[0] == signature and time.monotonic() - cached[1] < _INDEX_TTL_SECONDS ): return cached[2] rows = ( ( await db.execute( select(SupportKbArticle) .where(SupportKbArticle.status == KB_PUBLISHED) .order_by(SupportKbArticle.updated_at.desc()) ) ) .scalars() .all() ) entries = [ { "id": str(article.id), "title": article.title, "summary": article.summary or fallback_summary(article), "aliases": [ line.strip() for line in (article.search_aliases or "").splitlines() if line.strip() ], "category": article.category, "source": article.source, "language": article.language, "has_summary": bool(article.summary), "updated_at": str(article.updated_at or ""), } for article in rows ] _INDEX_CACHE["published"] = (signature, time.monotonic(), entries) logger.info("kb_wiki: indexed %d published articles", len(entries)) return entries async def index_health(db: AsyncSession) -> dict[str, Any]: """What the console shows about the index's own completeness.""" entries = await index(db) return { "articles": len(entries), "with_summary": sum(1 for e in entries if e["has_summary"]), "with_aliases": sum(1 for e in entries if e["aliases"]), "mirrored": sum(1 for e in entries if e["source"] == KB_SOURCE_FEISHU_DOC), } # --------------------------------------------------------------------------- # Links # --------------------------------------------------------------------------- def _canonical( a: uuid.UUID, b: uuid.UUID, kind: str, ) -> tuple[uuid.UUID, uuid.UUID]: if kind in _SYMMETRIC and str(b) < str(a): return b, a return a, b async def link_articles( db: AsyncSession, *, from_id: uuid.UUID, to_id: uuid.UUID, kind: str = LINK_RELATED, note: str | None = None, actor_person_id: uuid.UUID | None = None, ) -> SupportKbArticleLink | None: """Write one edge, idempotently. ``None`` when it already exists.""" if kind not in KB_LINK_KINDS: raise ValueError(f"unknown link kind: {kind!r}") if from_id == to_id: # A page related to itself is always a bug upstream, and it renders as # an infinite "see also" loop. return None left, right = _canonical(from_id, to_id, kind) existing = ( await db.execute( select(SupportKbArticleLink).where( SupportKbArticleLink.from_article_id == left, SupportKbArticleLink.to_article_id == right, SupportKbArticleLink.kind == kind, ) ) ).scalar_one_or_none() if existing is not None: return None row = SupportKbArticleLink( from_article_id=left, to_article_id=right, kind=kind, note=note, created_by_person_id=actor_person_id, ) db.add(row) await db.flush() return row async def links_for( db: AsyncSession, *, article_id: uuid.UUID, ) -> list[dict[str, Any]]: """Every edge touching this page, in both directions, with the other title.""" rows = ( ( await db.execute( select(SupportKbArticleLink).where( or_( SupportKbArticleLink.from_article_id == article_id, SupportKbArticleLink.to_article_id == article_id, ) ) ) ) .scalars() .all() ) if not rows: return [] others = { row.to_article_id if row.from_article_id == article_id else row.from_article_id for row in rows } titles = { row[0]: (row[1], row[2]) for row in ( await db.execute( select( SupportKbArticle.id, SupportKbArticle.title, SupportKbArticle.status, ).where(SupportKbArticle.id.in_(others)) ) ).all() } out: list[dict[str, Any]] = [] for row in rows: other = ( row.to_article_id if row.from_article_id == article_id else row.from_article_id ) title, status = titles.get(other, (None, None)) out.append( { "id": str(row.id), "kind": row.kind, "other_id": str(other), "other_title": title, # A link whose target has been archived is a dead link, and the # lint pass reads this to find them. "other_status": status, "outgoing": row.from_article_id == article_id, "note": row.note, } ) return out async def expand_with_links( db: AsyncSession, *, articles: list[SupportKbArticle], limit: int, ) -> list[SupportKbArticle]: """Add each hit's ``related`` neighbours, up to ``limit``. One extra query and no model call. The case it is for: an employee asks about a thing whose answer is split across two pages — a policy and the form that implements it — and retrieval can only ever return one of them because only one contains the words they used. Related-only on purpose. Following ``contradicts`` would hand the model a document the corpus already knows to be wrong, and ``supersedes`` points at something deliberately retired. """ if not articles or len(articles) >= limit: return articles have = {article.id for article in articles} rows = ( ( await db.execute( select(SupportKbArticleLink).where( SupportKbArticleLink.kind == LINK_RELATED, or_( SupportKbArticleLink.from_article_id.in_(have), SupportKbArticleLink.to_article_id.in_(have), ), ) ) ) .scalars() .all() ) wanted: list[uuid.UUID] = [] for row in rows: for candidate in (row.from_article_id, row.to_article_id): if candidate not in have and candidate not in wanted: wanted.append(candidate) if not wanted: return articles extra = ( ( await db.execute( select(SupportKbArticle).where( SupportKbArticle.id.in_(wanted[: limit - len(articles)]), # Only published neighbours: an archived one is not an # answer, and a mirrored page that vanished upstream is # exactly the thing archiving exists to take out of reach. SupportKbArticle.status == KB_PUBLISHED, ) ) ) .scalars() .all() ) return [*articles, *extra] # --------------------------------------------------------------------------- # Ingest — "is this already written down?" # --------------------------------------------------------------------------- @dataclass(frozen=True) class IngestOutcome: decision: str proposals: list[str] available: bool reason: str | None = None def as_dict(self) -> dict[str, Any]: return { "decision": self.decision, "proposals": self.proposals, "llm_available": self.available, "reason": self.reason, } _INGEST_SYSTEM = """你在维护一个企业 IT 知识库的目录。给定一条**新知识**,和目录里 最像的几条**已有条目**(每条含 id、标题、一句话说明、别名),判断这条新知识应该 怎么进入知识库。只能选一种: - "new":目录里没有讲这件事的条目。**这是最后的选择,不是默认选择。** - "duplicate":已有条目讲的就是同一件事,新知识没有带来新信息。 - "augments":已有条目讲的是同一件事,但新知识补充了它没有的内容(一个新场景、 一个新步骤、一个新的失败原因)。 - "alias_only":已有条目内容上完全覆盖了,新知识的价值只是**换了一种说法** —— 员工用的词和条目里写的词不一样。 - "contradicts":已有条目和新知识说法冲突,不能简单合并。 判断的重点是**讲的是不是同一件事**,不是措辞像不像。 「怎么连 VPN」和「VPN 连不上怎么办」是两件事(一个是配置,一个是排障); 「电脑坏了怎么报修」和「笔记本故障怎么处理」是同一件事。 选 "augments" 时给出要追加的小节(一个短标题 + 正文);选 "alias_only" 或 "duplicate" 时给出应当补进去的别名(员工真实会用的说法,每条一句,不要把整段 正文塞进别名)。reason 用中文写一句,点名依据。""" async def _candidates( db: AsyncSession, *, title: str, body: str, ) -> list[SupportKbArticle]: """The shortlist the model judges against. Reuses the corpus's own published search (semantic when embeddings are configured, ranked lexical otherwise) rather than inventing a second retrieval path — if the two disagreed, the one the employee actually gets would be the one that matters and this one would be measuring something else. """ from app.domains.support.kb_service import SupportKbService # noqa: PLC0415 probe = f"{title}\n{' '.join((body or '').split())[:300]}" found = await SupportKbService(db).search_published(probe, limit=MAX_CANDIDATES) return [article for article in found if article.status == KB_PUBLISHED] def _editable(article: SupportKbArticle) -> bool: """Can Helix write to this page at all? A mirrored Feishu doc is read-only here — Feishu is its only editing surface — and ``_guard_editable`` refuses one at APPLY time. Producing a proposal that mutates one is therefore a queue row whose only possible outcomes are "reject" and "accept, and get an error": the reviewer spends the attention, the bulk path spends it eleven-at-a-time, and (before the fingerprint fix) the failure burned the suggestion permanently. ``kb_lint`` already degrades a merge onto a mirror into a link for exactly this reason. This is the same rule on the ingest path. """ return article.source != KB_SOURCE_FEISHU_DOC def _entry(article: SupportKbArticle) -> dict[str, Any]: return { "id": str(article.id), "title": article.title, "summary": article.summary or fallback_summary(article), "aliases": [ line.strip() for line in (article.search_aliases or "").splitlines() if line.strip() ][:8], } async def propose_from_knowledge( db: AsyncSession, *, title: str, body: str, category: str, source: str, source_ref: str | None = None, language: str | None = None, aliases: str | None = None, evidence: dict[str, Any] | None = None, ) -> IngestOutcome: """Turn one piece of new knowledge into proposals against the corpus. Degrades to exactly today's behaviour — a single ``new_article`` proposal — when no model backend is configured or the model's answer is unusable. That is the point of the fallback: without a model this is no worse than what shipped before, and a deployment with no LLM key must not silently stop ingesting. """ title = (title or "").strip() body = (body or "").strip() if not title or not body: return IngestOutcome("skipped", [], True, "标题或正文为空") new_article_payload = { "title": title, "body": body, "category": category, "language": language, "aliases": aliases, } async def _propose_new(reason: str, available: bool) -> IngestOutcome: row = await loop.propose( db, corpus=CORPUS_SUPPORT_KB, kind=KIND_NEW_ARTICLE, title=title, source=source, source_ref=source_ref, payload=new_article_payload, rationale=reason, evidence=evidence, ) return IngestOutcome( "new", [str(row.id)] if row else [], available, reason, ) candidates = await _candidates(db, title=title, body=body) if not candidates: return await _propose_new("目录里没有相近的条目。", True) from app.domains.support.copilot import one_shot_json # noqa: PLC0415 payload = await one_shot_json( db, _INGEST_SYSTEM, ( "新知识:\n" f"标题:{title}\n正文:\n{body[:3000]}\n\n" "目录里最像的已有条目:\n" + json.dumps([_entry(a) for a in candidates], ensure_ascii=False) + '\n\n请输出 JSON:{"decision":"new|duplicate|augments|alias_only' '|contradicts","target_id":"命中的条目 id 或 null",' '"aliases":["应补的说法", ...],"section_heading":"追加小节的标题",' '"section_body":"追加小节的正文","reason":"中文一句话"}' ), timeout=45.0, max_tokens=1200, ) if not payload: return await _propose_new( "模型不可用,按既有行为落成一篇新草稿等待人工判断。", False, ) allowed = {str(a.id): a for a in candidates} decision = str(payload.get("decision") or "").strip().lower() target_id = str(payload.get("target_id") or "").strip() target = allowed.get(target_id) reason = str(payload.get("reason") or "").strip()[:2000] or None alias_list = [ " ".join(str(item).split()) for item in (payload.get("aliases") or []) if str(item).strip() ][:8] # A decision about a page the model was not shown is not a decision; it is # the model naming something. Fall back rather than write against an id we # never offered it. if decision in ("duplicate", "augments", "alias_only", "contradicts") and ( target is None ): logger.info( "kb_wiki: model chose %s but named an id outside the shortlist (%s)", decision, target_id[:40], ) return await _propose_new( f"模型判为 {decision},但给的条目 id 不在候选集里,按新条目处理。", True, ) if decision == "new": return await _propose_new(reason or "模型判断这是一件新的事。", True) assert target is not None # narrowed by the guard above common: dict[str, Any] = { "corpus": CORPUS_SUPPORT_KB, "source": source, "source_ref": source_ref, "target_ref": str(target.id), "rationale": reason, "evidence": {**(evidence or {}), "incoming_title": title}, } if decision in ("duplicate", "alias_only", "augments") and not _editable(target): # The model is right that these are the same thing — but the page it # named cannot be written to from here. A link keeps the pair visible # together, which is the whole value the finding still has. # # ⚠️ The payload carries the incoming text, and that is load-bearing # twice over. The ``KIND_LINK`` apply branch, when there is no # ``related_ref``, FILES THE INCOMING KNOWLEDGE AS A DRAFT and links the # pair — with an empty payload it calls ``service.create(body="")`` and # dies on 「标题和正文不能为空」, i.e. exactly the "reject, or accept and # get an error" row ``_editable`` exists to prevent. And # ``fingerprint_for`` hashes the payload: with ``{}`` and no # ``related_ref``, EVERY item that degraded onto the same mirror # collapsed to one fingerprint, so the second one onward was silently # dropped while the caller was told a proposal had been filed. row = await loop.propose( db, kind=KIND_LINK, title=f"关联「{target.title}」与新进来的内容", payload={ "incoming_title": title, "incoming_body": body[:4000], "category": category, "language": language, }, **{ **common, "rationale": ( (reason or "") + "(目标是飞书镜像页,Helix 改不了它 —— 接受会把新内容" "存成草稿并与它关联,而不是改写它。)" )[:2000], }, ) return IngestOutcome( "linked_mirror", [str(row.id)] if row else [], True, reason, ) if decision in ("duplicate", "alias_only"): if not alias_list: # Genuinely nothing to add: the knowledge is already written down in # the words people use. Recording that is more useful than filing an # empty proposal — it is the ingest path's only "no action needed" # outcome and the console counts it. return IngestOutcome( "duplicate", [], True, reason or "已有条目完全覆盖,且措辞一致。", ) row = await loop.propose( db, kind=KIND_ADD_ALIASES, title=f"给「{target.title}」补 {len(alias_list)} 条说法", payload={"aliases": alias_list}, **common, ) return IngestOutcome( "alias_only", [str(row.id)] if row else [], True, reason, ) if decision == "augments": heading = str(payload.get("section_heading") or "").strip()[:120] section = str(payload.get("section_body") or "").strip() if not section: return await _propose_new( "模型判为补充,但没有给出要追加的内容,按新条目处理。", True, ) row = await loop.propose( db, kind=KIND_APPEND_SECTION, title=f"给「{target.title}」追加一节:{heading or '补充'}", payload={ "heading": heading, "body": section[:4000], "aliases": alias_list, }, **common, ) return IngestOutcome("augments", [str(row.id)] if row else [], True, reason) # contradicts row = await loop.propose( db, kind=KIND_CONTRADICTION, title=f"与「{target.title}」的说法冲突", payload={ "incoming_title": title, "incoming_body": body[:4000], "category": category, "language": language, }, **common, ) return IngestOutcome("contradicts", [str(row.id)] if row else [], True, reason) # --------------------------------------------------------------------------- # Applying a decision # --------------------------------------------------------------------------- def _merge_aliases(existing: str | None, additions: list[str]) -> str: """Union, order-preserving, case-insensitively deduplicated.""" lines = [line.strip() for line in (existing or "").splitlines() if line.strip()] seen = {line.casefold() for line in lines} for alias in additions: cleaned = " ".join(str(alias).split()) if cleaned and cleaned.casefold() not in seen: lines.append(cleaned) seen.add(cleaned.casefold()) return "\n".join(lines) class SupportKbAdapter: """How an accepted proposal actually lands on the support knowledge base.""" corpus = CORPUS_SUPPORT_KB async def _article( self, db: AsyncSession, ref: str | None, *, what: str, ) -> SupportKbArticle: if not ref: raise ProposalApplyError(f"这条提案没有指明{what}。") try: key = uuid.UUID(str(ref)) except (TypeError, ValueError) as exc: raise ProposalApplyError(f"{what}的 id 不合法:{ref}") from exc article = await db.get(SupportKbArticle, key) if article is None: raise ProposalApplyError( f"{what}已经不存在了,这条提案没有可落地的对象。", superseded=True, ) if article.status == KB_ARCHIVED: raise ProposalApplyError( f"{what}已下架,这条提案不再适用。", superseded=True, ) return article async def describe( self, db: AsyncSession, *, proposal: Any, ) -> dict[str, Any]: """The page as it stands **today**, so the reviewer judges the real thing. A proposal written last Tuesday against a page somebody has since rewritten is a proposal about a document that no longer exists; showing the reviewer the version the proposer saw would be the most confident possible way to get a wrong decision. """ out: dict[str, Any] = {"corpus": self.corpus} for key, ref in (("target", proposal.target_ref), ("related", proposal.related_ref)): if not ref: continue try: article = await db.get(SupportKbArticle, uuid.UUID(str(ref))) except (TypeError, ValueError): continue if article is None: out[key] = {"missing": True, "id": ref} continue out[key] = { "id": str(article.id), "title": article.title, "summary": article.summary or fallback_summary(article), "body": (article.body or "")[:6000], "status": article.status, "source": article.source, "category": article.category, "aliases": [ line.strip() for line in (article.search_aliases or "").splitlines() if line.strip() ], "updated_at": str(article.updated_at or ""), # A mirrored page is machine-owned: Feishu is the only editing # surface, and an edit here would be overwritten by the next # sync. The console needs this to disable the accept button # rather than let a reviewer make a change that evaporates. "read_only": article.source == KB_SOURCE_FEISHU_DOC, } return out async def apply( self, db: AsyncSession, *, proposal: Any, actor_person_id: Any, trace_id: str, ) -> dict[str, Any]: from app.domains.support.kb_service import ( # noqa: PLC0415 SupportKbService, reset_vector_cache, ) payload = proposal.payload or {} kind = proposal.kind service = SupportKbService(db) if kind == KIND_NEW_ARTICLE: article = await service.create( title=str(payload.get("title") or proposal.title), body=str(payload.get("body") or ""), category=str(payload.get("category") or "other"), author=None, source=_article_source(proposal.source), publish=False, search_aliases=payload.get("aliases"), language=payload.get("language"), trace_id=trace_id, ) reset_index_cache() return {"applied_ref": str(article.id), "created": True} if kind == KIND_ADD_ALIASES: article = await self._article(db, proposal.target_ref, what="目标条目") _guard_editable(article) merged = _merge_aliases( article.search_aliases, list(payload.get("aliases") or []), ) # A no-op for the ARTICLE is not a no-op for the record. The early # return here skipped the audit row entirely, so a proposal marked # ``accepted`` left nothing naming the article — the same defect as # its security-library twin, which additionally skipped a gap # closure. Both halves of a pair move together. noop = merged == (article.search_aliases or "") if not noop: article.search_aliases = merged # Aliases are part of the embedding input, so a stale vector # would keep the corpus from finding the page by exactly the # words this change was made to catch. The router re-embeds in # the background. _invalidate_embedding(article) reset_vector_cache() await self._audit(db, article, "support.kb.aliases_added", actor_person_id, trace_id) reset_index_cache() return { "applied_ref": str(article.id), "aliases": merged.count("\n") + 1, "noop": noop, } if kind == KIND_APPEND_SECTION: article = await self._article(db, proposal.target_ref, what="目标条目") _guard_editable(article) heading = str(payload.get("heading") or "").strip() section = str(payload.get("body") or "").strip() if not section: raise ProposalApplyError("这条提案没有要追加的正文。") block = f"\n\n## {heading}\n{section}" if heading else f"\n\n{section}" if section in (article.body or ""): # Somebody already added it by hand, or the proposal was applied # and retried. Idempotence matters here because the reviewer's # click is retryable. return {"applied_ref": str(article.id), "noop": True} article.body = ((article.body or "") + block)[:20000] extra_aliases = list(payload.get("aliases") or []) if extra_aliases: article.search_aliases = _merge_aliases( article.search_aliases, extra_aliases, ) _invalidate_embedding(article) await self._audit( db, article, "support.kb.section_appended", actor_person_id, trace_id, ) reset_index_cache() reset_vector_cache() return {"applied_ref": str(article.id), "appended": len(section)} if kind == KIND_MERGE: survivor = await self._article(db, proposal.target_ref, what="保留的条目") absorbed = await self._article(db, proposal.related_ref, what="被合并的条目") _guard_editable(absorbed) # The SURVIVOR is written too — its aliases gain the absorbed # page's, and its embedding is cleared — so it needs the same guard. # Only the absorbed side was checked, which meant a merge INTO a # mirrored page would silently edit a row whose only editing surface # is Feishu. _guard_editable(survivor) if survivor.id == absorbed.id: raise ProposalApplyError("保留条目和被合并条目是同一条。") survivor.search_aliases = _merge_aliases( survivor.search_aliases, [ absorbed.title, *[ line.strip() for line in (absorbed.search_aliases or "").splitlines() if line.strip() ], ], ) _invalidate_embedding(survivor) absorbed.status = KB_ARCHIVED _invalidate_embedding(absorbed) await link_articles( db, from_id=survivor.id, to_id=absorbed.id, kind=LINK_SUPERSEDES, note=f"合并自「{absorbed.title}」", actor_person_id=actor_person_id, ) # Every other open proposal about the page we just retired is now # about a page nobody can act on. await loop.supersede_for_target( db, corpus=self.corpus, target_ref=str(absorbed.id), reason="这条条目已被合并到另一条,提案不再适用。", ) await self._audit(db, survivor, "support.kb.merged", actor_person_id, trace_id) # And one naming the page that was ARCHIVED. Without it the audit # spine records that something was merged and never which article # stopped being published — the half a reader actually needs when # asking why a document disappeared. await self._audit( db, absorbed, "support.kb.merged_away", actor_person_id, trace_id, ) reset_index_cache() reset_vector_cache() return {"applied_ref": str(survivor.id), "archived": str(absorbed.id)} if kind in (KIND_CONTRADICTION, KIND_LINK): left = await self._article(db, proposal.target_ref, what="条目") right_ref = proposal.related_ref if not right_ref: # A contradiction raised by ingest has no second article yet — # the incoming knowledge was never written down. Accepting it # files the incoming text as a draft and links the pair, which # is the only outcome that leaves the disagreement visible. article = await service.create( title=str(payload.get("incoming_title") or proposal.title), body=str(payload.get("incoming_body") or ""), category=str(payload.get("category") or left.category), author=None, source=_article_source(proposal.source), publish=False, language=payload.get("language"), trace_id=trace_id, ) right_ref = str(article.id) right = await self._article(db, right_ref, what="另一条条目") # ``link_articles`` returns None when the edge already exists or # when both ends are the same page. Reporting ``linked: True`` for # those said a write happened that did not — and this branch was # also the only mutating one with no audit row at all. created = await link_articles( db, from_id=left.id, to_id=right.id, kind=( LINK_CONTRADICTS if kind == KIND_CONTRADICTION else LINK_RELATED ), note=proposal.rationale, actor_person_id=actor_person_id, ) if created is not None: await self._audit( db, left, "support.kb.linked", actor_person_id, trace_id, ) reset_index_cache() return { "applied_ref": str(right.id), "linked": created is not None, "already_linked": created is None, } if kind == KIND_DEAD_LINK: # The rationale this proposal carries says 「接受即删除这条关联」, and # until now accepting one raised 「知识库不支持这种提案:dead_link」 — # the lint pass produced a kind the adapter had never been taught, # so every dead-link finding was un-closeable by construction. link_id = str((proposal.payload or {}).get("link_id") or "") try: key = uuid.UUID(link_id) except (TypeError, ValueError) as exc: raise ProposalApplyError( f"这条提案没有可删除的关联 id:{link_id or '(空)'}", ) from exc link = await db.get(SupportKbArticleLink, key) # ⚠️ The link has to be one of THIS proposal's own. ``link_id`` lives # in the payload, and ``payload_override`` lets a reviewer supply a # declared key — so without this the accept call could name any link # row in the corpus and delete it, while the decision audit named a # different, legitimately-dead one. The proposal's target and # related refs are what it was raised about. if link is not None: endpoints = {str(link.from_article_id), str(link.to_article_id)} claimed = { str(proposal.target_ref or ""), str(proposal.related_ref or ""), } if not endpoints <= claimed: raise ProposalApplyError( "这条关联不属于本提案指向的条目,拒绝删除。", ) if link is None: # Someone already removed it. Superseded, not failed: the # question stopped being askable, and nobody decided anything. raise ProposalApplyError( "这条关联已经不在了,提案不再适用。", superseded=True, ) from_id = link.from_article_id to_id = link.to_article_id link_kind = link.kind await db.delete(link) await db.flush() source = await db.get(SupportKbArticle, from_id) if source is not None: await self._audit( db, source, "support.kb.link_removed", actor_person_id, trace_id, ) reset_index_cache() return { "applied_ref": str(from_id), "removed_link": str(key), "to": str(to_id), "link_kind": link_kind, } if kind == KIND_RETIRE: article = await self._article(db, proposal.target_ref, what="目标条目") _guard_editable(article) article.status = KB_ARCHIVED _invalidate_embedding(article) await loop.supersede_for_target( db, corpus=self.corpus, target_ref=str(article.id), reason="这条条目已下架,提案不再适用。", ) await self._audit(db, article, "support.kb.retired", actor_person_id, trace_id) reset_index_cache() reset_vector_cache() return {"applied_ref": str(article.id), "archived": True} raise ProposalApplyError(f"知识库不支持这种提案:{kind}") async def _audit( self, db: AsyncSession, article: SupportKbArticle, action: str, actor_person_id: Any, trace_id: str, ) -> None: from app.audit.service import AuditService # noqa: PLC0415 await AuditService(db).log( trace_id=trace_id or "knowledge-loop", actor_type="user", actor_user_id=actor_person_id, action=action, status="success", resource_type="support_kb", resource_id=str(article.id), detail={"title": article.title, "via": "knowledge_loop"}, ) def _guard_editable(article: SupportKbArticle) -> None: """Mirrored pages are machine-owned; Feishu is the only editing surface. An edit written here would be overwritten by the next sync — so accepting such a proposal would look like it worked and quietly do nothing, which is worse than refusing. """ if article.source == KB_SOURCE_FEISHU_DOC: raise ProposalApplyError( "这是飞书文档的镜像条目,编辑面在飞书 —— 在 Helix 改会被下一轮同步覆盖。" "请到原文修改,或改为新建一条 Helix 自己的条目。", ) def _invalidate_embedding(article: SupportKbArticle) -> None: article.embedding = None article.embedding_model = None article.embedded_at = None def _article_source(proposal_source: str) -> str: """Which ``source`` a page created by the loop carries. Everything the loop creates is ``ai_ticket`` — the existing value for "a machine drafted this, a person published it" — rather than a new value per ingest path. A new source value would have to be taught to the KB admin filter, the employee-facing browse, the mirror guard and the export, and the distinction it would carry is already on the proposal row. """ from app.domains.support.models import KB_SOURCE_AI_TICKET # noqa: PLC0415 return KB_SOURCE_AI_TICKET register_adapter(SupportKbAdapter()) __all__ = [ "IngestOutcome", "SupportKbAdapter", "expand_with_links", "fallback_summary", "index", "index_health", "link_articles", "links_for", "propose_from_knowledge", "reset_index_cache", "summarise", "summarise_missing", ]