"""Knowledge-base service for the help-desk (求助域 PR4). Published articles are the deflection surface the AI copilot searches and the help center can render. Articles are authored manually or AI-drafted from a resolved ticket, then adopted (published) by an admin. """ from __future__ import annotations import logging import math import time from datetime import UTC, datetime from typing import Any from uuid import UUID from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # Wall-clock ceiling for one bulk-draft request (#2230). nginx cuts the # hop at ``proxy_read_timeout 300s`` and the client aborts just under # that, so a batch that cannot finish inside the budget returns PARTIAL # counts + ``remaining`` rather than dying as a 504 over work that did # land. 240s leaves room for the response + audit write. _BULK_DRAFT_BUDGET_SECONDS = 240.0 from app.audit.service import AuditService from app.core.exceptions import AppException from app.domains.support.models import ( CATEGORIES, KB_ARCHIVED, KB_DRAFT, KB_PUBLISHED, KB_SOURCE_FEISHU_DOC, KB_SOURCE_MANUAL, SupportKbArticle, ) from app.domains.support.text_match import jaccard, like_escape, topical_shingles #: Collection aliases live at MODULE scope deliberately: inside #: ``SupportKbService`` the name ``list`` resolves to the service's own ``list`` #: method, so bare ``list[...]`` annotations written later in the class body #: become invalid types and can leak unknown return types into callers. Vector = list[float] ArticleList = list[SupportKbArticle] #: ``{"published": ((count, max_updated_at), fetched_at, {article_id: vector})}`` #: — the embedding corpus, held for the process and re-read when the signature #: moves. One entry, so this cannot grow: it is a corpus cache, not a query cache. _vector_cache: dict[str, tuple[tuple[int, str], float, dict[UUID, Vector]]] = {} #: Backstop on the signature probe. ``updated_at`` is second-precision, so two #: edits inside one second with an unchanged count produce one signature — the #: second edit would otherwise be invisible for the life of the process. A few #: minutes of a stale RANKING vector is tolerable; forever is not. _VECTOR_CACHE_TTL_SECONDS = 300.0 def reset_vector_cache() -> None: """Drop the cached corpus — for tests, and after a bulk re-embed.""" _vector_cache.clear() def _guard_mirrored(article: SupportKbArticle) -> None: """Mirrored wiki articles are machine-owned end to end — content is authored in Feishu and status is decided by the sync (published while the doc exists with content, archived when it disappears/empties). Any manual edit here would be silently clobbered by the next sync, so refuse it with directions instead.""" if article.source == KB_SOURCE_FEISHU_DOC: raise AppException( code="KB_MIRRORED_READONLY", message="该条目镜像自飞书文档,请直接在飞书中编辑;移出知识库请在飞书目录中移除该文档", status_code=422, ) from app.identity.person_models import IdentityPerson _TITLE_MAX = 255 _BODY_MAX = 16000 _EMBED_TEXT_MAX = 4000 # cap text sent to the embeddings API # Minimum cosine similarity for a semantic hit. text-embedding-v3 scores # unrelated CN/EN text around 0.1–0.3 and genuinely related text ≳0.4, so this # keeps clearly-irrelevant articles out of the result (and out of the LLM # context the copilot grounds on). Keyword fallback still tops up to the limit. _SEMANTIC_MIN_SCORE = 0.4 #: Shingle-recall knobs. ``_SHINGLE_MIN_SCORE`` matches the sibling #: ``find_similar_resolved`` threshold: low enough that a one-word overlap in a #: short question still counts, high enough to drop "shares the word 'the'". _SHINGLE_MIN_SCORE = 0.08 _SHINGLE_CANDIDATES = 400 _ANCHOR_TOKENS = 8 #: Relative relevance cliff: keep only what scores within this fraction of the #: BEST hit. The absolute floor alone cannot separate "shares one bigram by #: coincidence" from "is about this" — with a short query like 「怎么连VPN」a #: single shared 「怎么」 scores 1/12 ≈ 0.083 and clears the floor, so an #: unrelated article rode into the top four. A cliff relative to the winner #: adapts to how strong the match actually was, and by construction can never #: drop the top result. _SHINGLE_RELATIVE_CUTOFF = 0.35 _ALIASES_MAX = 4000 _LANGUAGES = ("zh", "en", "ja") def _clean_aliases(value: str | None) -> str | None: """Alternate phrasings → newline-separated text (blank → NULL). Newline-separated plain text, not JSON: the keyword path ILIKEs the column directly and the admin edits it in a textarea, exactly the shape the Feishu helpdesk template asks for ("添加多个相似问题时,请分行输入").""" if value is None: return None seen: set[str] = set() lines: list[str] = [] for raw in value.replace("\r\n", "\n").replace("\r", "\n").split("\n"): line = raw.strip() if line and line not in seen: seen.add(line) lines.append(line) return "\n".join(lines)[:_ALIASES_MAX] or None def _clean_language(value: str | None) -> str | None: lang = (value or "").strip().lower() return lang if lang in _LANGUAGES else None def _embedding_text(article: SupportKbArticle) -> str: """The searchable text for an article — title weighted by repetition, then the alternate phrasings, then body, trimmed to the embedding input cap. ``search_aliases`` sits directly after the title on purpose: they ARE titles — the other ways employees ask the same question — and they are the single most valuable retrieval signal the Feishu-helpdesk export carries (「打印机卡纸了怎么办」vs the article's「打印机发生故障」). Putting them ahead of the body also means the 4000-char cap can never drop them. """ aliases = (article.search_aliases or "").strip() parts = [article.title, article.title] if aliases: parts.append(aliases) parts.append(article.body) return "\n".join(parts)[:_EMBED_TEXT_MAX] def _pending_embedding_clause(active_model: str): """ "Needs a vector under ``active_model``" — shared by the bounded backfill and the pending counter so the number the admin sees can never disagree with the rows the backfill will actually pick up.""" return and_( SupportKbArticle.status == KB_PUBLISHED, or_( SupportKbArticle.embedding.is_(None), # An article with a stored vector but a NULL ``embedding_model`` # must also be re-embedded: in SQL ``embedding_model != # active_model`` evaluates to NULL (not true) when the column is # NULL, so without this explicit clause such a row would keep a # stale/unknown vector forever after the active model changes. SupportKbArticle.embedding_model.is_(None), SupportKbArticle.embedding_model != active_model, ), ) def _cosine(a: list[float], b: list[float]) -> float: """Cosine similarity of two equal-length float vectors. Returns 0.0 for a length mismatch or a zero vector (defensive — never raises).""" if not a or not b or len(a) != len(b): return 0.0 dot = sum(x * y for x, y in zip(a, b, strict=False)) na = math.sqrt(sum(x * x for x in a)) nb = math.sqrt(sum(y * y for y in b)) if na == 0.0 or nb == 0.0: return 0.0 return dot / (na * nb) class SupportKbService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) async def _audit( self, *, action: str, article: SupportKbArticle, actor_id, trace_id: str ) -> None: await self.audit.log( trace_id=trace_id or "", actor_type="user", action=action, status="success", actor_user_id=actor_id, resource_type="support_kb_article", resource_id=str(article.id), ) async def get(self, article_id: UUID) -> SupportKbArticle | None: return await self.db.get(SupportKbArticle, article_id) async def create( self, *, title: str, body: str, category: str, author: IdentityPerson | None, source: str = KB_SOURCE_MANUAL, source_ticket_id: UUID | None = None, publish: bool = False, search_aliases: str | None = None, language: str | None = None, source_category: str | None = None, trace_id: str = "", ) -> SupportKbArticle: title = (title or "").strip() body = (body or "").strip() if not title or not body: raise AppException( code="VALIDATION_ERROR", message="标题和正文不能为空", status_code=422 ) art = SupportKbArticle( title=title[:_TITLE_MAX], body=body[:_BODY_MAX], category=category if category in CATEGORIES else "other", status=KB_PUBLISHED if publish else KB_DRAFT, source=source, source_ticket_id=source_ticket_id, author_id=author.id if author else None, published_at=datetime.now(UTC) if publish else None, search_aliases=_clean_aliases(search_aliases), language=_clean_language(language), source_category=(source_category or "").strip()[:64] or None, ) self.db.add(art) await self.db.flush() await self._audit( action="support.kb.created", article=art, actor_id=author.id if author else None, trace_id=trace_id, ) return art async def update( self, *, article: SupportKbArticle, actor: IdentityPerson, trace_id: str = "", title: str | None = None, body: str | None = None, category: str | None = None, search_aliases: str | None = None, language: str | None = None, ) -> SupportKbArticle: _guard_mirrored(article) if title is not None and title.strip(): article.title = title.strip()[:_TITLE_MAX] content_changed = False if body is not None and body.strip(): article.body = body.strip()[:_BODY_MAX] content_changed = True if title is not None and title.strip(): content_changed = True if category is not None and category in CATEGORIES: article.category = category if search_aliases is not None: # Empty string is a real edit ("remove every alias"), so this # branch keys on ``is not None``, not on truthiness. cleaned = _clean_aliases(search_aliases) if cleaned != article.search_aliases: article.search_aliases = cleaned content_changed = True if language is not None: article.language = _clean_language(language) await self._audit( action="support.kb.updated", article=article, actor_id=actor.id, trace_id=trace_id ) # Invalidate the embedding when searchable text changed so stale vectors # don't drift from live content; the router re-embeds in the background. if content_changed and article.status == KB_PUBLISHED: article.embedding = None article.embedding_model = None article.embedded_at = None return article async def set_status( self, *, article: SupportKbArticle, status: str, actor: IdentityPerson, trace_id: str = "", ) -> SupportKbArticle: _guard_mirrored(article) if status not in (KB_DRAFT, KB_PUBLISHED, KB_ARCHIVED): raise AppException(code="VALIDATION_ERROR", message="无效状态", status_code=422) article.status = status if status == KB_PUBLISHED and article.published_at is None: article.published_at = datetime.now(UTC) await self._audit( action=f"support.kb.{status}", article=article, actor_id=actor.id, trace_id=trace_id ) return article async def delete( self, *, article: SupportKbArticle, actor: IdentityPerson, trace_id: str = "", ) -> None: """Remove an article outright. ``archived`` already takes an article out of retrieval, so this exists for the other half of the job: a wrong / duplicated import that should leave no row behind at all. Mirrored wiki rows are refused for the same reason edits are — the next sync would just recreate them. """ _guard_mirrored(article) # Audit BEFORE the delete: the row (and its id) must still exist when # the trail is written. await self._audit( action="support.kb.deleted", article=article, actor_id=actor.id, trace_id=trace_id ) await self.db.delete(article) async def list( self, *, status: str | None = None, category: str | None = None, source: str | None = None, language: str | None = None, q: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[SupportKbArticle], int]: """Filtered, paginated list — the ADMIN table. ``q`` is an exact-phrase filter here on purpose ("find the article I know exists"). The employee browser wants relevance instead; that is :meth:`browse_published`.""" stmt = select(SupportKbArticle) if status: stmt = stmt.where(SupportKbArticle.status == status) if category: stmt = stmt.where(SupportKbArticle.category == category) if source: stmt = stmt.where(SupportKbArticle.source == source) if language: stmt = stmt.where(SupportKbArticle.language == language) stmt = self._apply_q(stmt, q) page = max(1, page) page_size = max(1, min(page_size, 100)) total = ( await self.db.execute(select(func.count()).select_from(stmt.subquery())) ).scalar() or 0 rows = ( ( await self.db.execute( stmt.order_by(SupportKbArticle.updated_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) ) .scalars() .all() ) return list(rows), int(total) async def browse_published( self, *, q: str | None = None, category: str | None = None, page: int = 1, page_size: int = 50, # ``ArticleList``, not ``list[...]``: inside this class the bare name # ``list`` resolves to the service's own ``list`` method (see the # module docstring), and mypy rejects it as a type. ) -> tuple[ArticleList, int]: """The employee knowledge base. With a query: RANKED non-semantic recall (see :meth:`_keyword_search`) capped at ``page_size`` — an employee typing a problem wants the best few answers, not page 3 of everything sharing a word. Without one: the plain paginated browse, newest first. Deliberately not semantic: this endpoint is un-rate-limited and a debounced search box would spend one embedding call per keystroke. The semantic path stays on the rate-limited deflection / assistant entrances. """ qc = (q or "").strip() if not qc: return await self.list( status=KB_PUBLISHED, category=category, page=page, page_size=page_size, ) rows = await self._keyword_search(qc, page_size, category=category) return rows, len(rows) async def search_published( self, q: str | None, limit: int = 8, *, embed_timeout: float = 60.0, ) -> ArticleList: """Find published articles relevant to ``q``. Semantic (embedding cosine) when the copilot/embeddings backend is available AND embedded articles exist; always falls back to keyword (ILIKE), and back-fills semantic results with keyword hits to reach ``limit``. No-query → most recent. ``embed_timeout`` bounds the query-embedding HTTP call (short on the synchronous deflection path).""" limit = max(1, min(limit, 20)) qc = (q or "").strip() if not qc: return await self._recent_published(limit) semantic = await self._semantic_search(qc, limit, embed_timeout=embed_timeout) if semantic is None: return await self._keyword_search(qc, limit) if not semantic: # The semantic corpus found nothing close enough; a literal policy # title can still be the right answer, so now (and only now) try # the keyword path. return await self._keyword_search(qc, limit) # A semantic hit is already a relevance decision. The old "top up to # six" rule then appended keyword rows that happened to share one word, # making the administrator trace show documents unrelated to the # employee's question. Fewer, evidence-backed sources are strictly # better than a full-looking list polluted by noise. # # Link expansion is the one exception, and it is a different kind of # top-up: a page explicitly marked RELATED to a hit was related by a # person accepting a proposal, not by sharing a word. The case it is # for is an answer split across two pages — a policy and the form that # implements it — where retrieval can only ever return the one that # contains the employee's vocabulary. Follows ``related`` only: # ``contradicts`` would hand the model a document the corpus already # knows to be wrong, and ``supersedes`` points at something retired. return await self._expand_with_links(semantic, limit) async def _expand_with_links( self, articles: ArticleList, limit: int, ) -> ArticleList: from app.domains.support.kb_wiki import expand_with_links try: return await expand_with_links( self.db, articles=list(articles), limit=limit, ) except Exception: # noqa: BLE001 — retrieval must survive a link bug logger.exception("kb: link expansion failed; returning hits as-is") return articles async def _recent_published(self, limit: int) -> ArticleList: rows = ( ( await self.db.execute( select(SupportKbArticle) .where(SupportKbArticle.status == KB_PUBLISHED) .order_by(SupportKbArticle.published_at.desc()) .limit(limit) ) ) .scalars() .all() ) return list(rows) async def _keyword_search( self, q: str, limit: int, *, category: str | None = None, ) -> ArticleList: """Non-semantic recall, RANKED. Two tiers: 1. Shingle match over each article's **question forms** (its title and every alternate phrasing), scored by Jaccard. This is the tier that matters: an FAQ is a question, so comparing question-to-question is what "relevant" means here, and CJK bigrams are the only way a Chinese query matches at all — 「怎么连VPN」has no whitespace, so a word-split query is one token that ILIKEs nothing. 2. Only if that finds nothing: the old token OR-match, which also reads the body, so a literal phrase buried in an answer is still findable. Ranking is the point, not just recall. Un-ranked token OR-match returned **116 of 124 imported FAQs** for one ordinary English question ("how", "the", "in" match nearly everything) with an unrelated Chinese article on top — a result list that long is the same as no answer. """ ranked = await self._shingle_search(q, limit, category=category) if ranked: return ranked stmt = select(SupportKbArticle).where(SupportKbArticle.status == KB_PUBLISHED) if category: stmt = stmt.where(SupportKbArticle.category == category) stmt = self._apply_tokens(stmt, q) rows = ( ( await self.db.execute( stmt.order_by(SupportKbArticle.published_at.desc()).limit(limit) ) ) .scalars() .all() ) return list(rows) async def _shingle_search( self, q: str, limit: int, *, category: str | None = None, ) -> ArticleList: probe = topical_shingles(q) if not probe: return [] # Deterministic anchor order (length desc, then lexicographic): a pure # CJK probe is all length-2 bigrams, so a bare length sort would pick an # arbitrary set-iteration subset and recall would vary run to run. anchors = sorted(probe, key=lambda t: (-len(t), t))[:_ANCHOR_TOKENS] conds = [] for tok in anchors: like = f"%{like_escape(tok)}%" conds.append(SupportKbArticle.title.ilike(like, escape="\\")) conds.append(SupportKbArticle.search_aliases.ilike(like, escape="\\")) # Prefilter reads (id, title, aliases) only — the candidate set can be # most of the library on a common word, and the bodies of the ones that # lose are never going to be shown. rows = ( await self.db.execute( select( SupportKbArticle.id, SupportKbArticle.title, SupportKbArticle.search_aliases, ) .where( SupportKbArticle.status == KB_PUBLISHED, or_(*conds), *([SupportKbArticle.category == category] if category else []), ) .limit(_SHINGLE_CANDIDATES) ) ).all() scored: list[tuple[float, str, UUID]] = [] for art_id, title, aliases in rows: forms = [title, *(aliases or "").split("\n")] best = max( (jaccard(probe, topical_shingles(form)) for form in forms if form.strip()), default=0.0, ) if best >= _SHINGLE_MIN_SCORE: scored.append((best, title, art_id)) if not scored: return [] # Title breaks score ties so the order is stable across identical runs. scored.sort(key=lambda t: (-t[0], t[1])) floor = scored[0][0] * _SHINGLE_RELATIVE_CUTOFF winners = [ art_id for score, _title, art_id in scored[:limit] if score >= floor ] loaded = ( ( await self.db.execute( select(SupportKbArticle).where(SupportKbArticle.id.in_(winners)) ) ) .scalars() .all() ) by_id = {art.id: art for art in loaded} return [by_id[art_id] for art_id in winners if art_id in by_id] async def _semantic_search( self, q: str, limit: int, *, embed_timeout: float = 60.0, ) -> ArticleList | None: """Embedding cosine ranking. Returns None to signal "fall back to keyword" (backend unavailable, query couldn't embed, or no embedded articles exist yet). Ranking reads (id, embedding) only, and only the ``limit`` winners are then loaded in full. The first shape selected whole ORM rows: production holds 246 published articles carrying **5.6 MB of embedding JSON and 0.66 MB of bodies**, so every question an employee asked pulled 6.3 MB across the wire and hydrated 246 objects to show at most 6 — inside the turn they were waiting on. The vectors are still fetched (see ``_vector_cache`` for why they are not fetched twice); the bodies of the 240 articles nobody is going to read no longer are. """ from app.domains.support.copilot import embed_texts embedded = await embed_texts(self.db, [q], timeout=embed_timeout) if embedded is None: return None query_vec = embedded[0][0] vectors = await self._published_vectors() if not vectors: return None scored = [(_cosine(query_vec, vec), art_id) for art_id, vec in vectors.items()] scored.sort(key=lambda t: (-t[0], str(t[1]))) winners = [art_id for score, art_id in scored[:limit] if score >= _SEMANTIC_MIN_SCORE] if not winners: # An empty list, not None: the corpus was searched and held nothing # close enough. ``search_published`` still tops up from keyword. return [] rows = ( ( await self.db.execute( select(SupportKbArticle).where(SupportKbArticle.id.in_(winners)) ) ) .scalars() .all() ) by_id = {art.id: art for art in rows} # Ranking order, not database order. return [by_id[art_id] for art_id in winners if art_id in by_id] async def _published_vectors(self) -> dict[UUID, Vector]: """The published corpus's vectors, cached for the process. Re-read only when the corpus actually changed. The staleness probe is a COUNT + MAX(updated_at) over the same rows — a few bytes — and every write path that could alter the corpus moves one of them: publishing, archiving, re-embedding and the wiki mirror's own updates all touch ``updated_at`` (``onupdate``), and an added or removed article moves the count. So the cache cannot serve a corpus the database no longer has, while a burst of questions about the same 246 articles costs one fetch. """ probe = ( await self.db.execute( select( func.count(), func.max(SupportKbArticle.updated_at), ) .select_from(SupportKbArticle) .where( SupportKbArticle.status == KB_PUBLISHED, SupportKbArticle.embedding.is_not(None), ) ) ).one() signature = (int(probe[0] or 0), str(probe[1])) cached = _vector_cache.get("published") if ( cached is not None and cached[0] == signature and time.monotonic() - cached[1] < _VECTOR_CACHE_TTL_SECONDS ): return cached[2] rows = ( await self.db.execute( select(SupportKbArticle.id, SupportKbArticle.embedding).where( SupportKbArticle.status == KB_PUBLISHED, SupportKbArticle.embedding.is_not(None), ) ) ).all() vectors = {row[0]: row[1] for row in rows if isinstance(row[1], list) and row[1]} _vector_cache["published"] = (signature, time.monotonic(), vectors) logger.info( "kb: cached %d published article vectors (signature=%s)", len(vectors), signature, ) return vectors async def embed_article(self, article: SupportKbArticle) -> bool: """Compute + store the article's embedding (best-effort). Returns True on success. Never raises — embedding is an enhancement, not a gate; the keyword path always works.""" from app.domains.support.copilot import embed_texts try: result = await embed_texts(self.db, [_embedding_text(article)]) if result is None: return False vectors, model = result article.embedding = vectors[0] article.embedding_model = model article.embedded_at = datetime.now(UTC) return True except Exception: # noqa: BLE001 - enhancement only logger.exception("embed_article failed for %s", article.id) return False async def count_pending_embeddings(self) -> int: """How many published articles still need a vector under the active model. Surfaced next to a bounded backfill's ``embedded`` count so a per-call cap can never read as "everything is vectorized" (#2974).""" from app.domains.support.copilot import resolve_active_embed_model active_model = await resolve_active_embed_model(self.db) return int( ( await self.db.execute( select(func.count()) .select_from(SupportKbArticle) .where(_pending_embedding_clause(active_model)) ) ).scalar() or 0 ) async def embedding_stats(self) -> dict[str, int]: """Return the counts used by the admin sync dashboard. Keep these counts aligned with the backfill predicate: a vector from a previous model is not counted as ready for the current index. """ from app.domains.support.copilot import resolve_active_embed_model active_model = await resolve_active_embed_model(self.db) published = int( ( await self.db.execute( select(func.count()) .select_from(SupportKbArticle) .where( SupportKbArticle.status == KB_PUBLISHED, ) ) ).scalar() or 0 ) ready = int( ( await self.db.execute( select(func.count()) .select_from(SupportKbArticle) .where( SupportKbArticle.status == KB_PUBLISHED, SupportKbArticle.embedding.is_not(None), SupportKbArticle.embedding_model == active_model, ) ) ).scalar() or 0 ) return { "published": published, "vectorized_total": ready, "vectorized_capacity": published, "pending": max(0, published - ready), } async def backfill_embeddings(self, limit: int = 50) -> dict[str, int]: """Embed published articles that have no embedding yet (or whose embedding model differs from the active one). Bounded per call so it can be driven from an admin button or a future cron without a long transaction. ``pending`` in the result is the backlog left AFTER this batch — the cap is reported, never silent.""" from app.domains.support.copilot import embed_texts, resolve_active_embed_model active_model = await resolve_active_embed_model(self.db) rows = ( ( await self.db.execute( select(SupportKbArticle) .where(_pending_embedding_clause(active_model)) .order_by(SupportKbArticle.published_at.desc()) .limit(max(1, min(limit, 200))) ) ) .scalars() .all() ) if not rows: return {"candidates": 0, "embedded": 0, "pending": 0} embedded = 0 for art in rows: result = await embed_texts(self.db, [_embedding_text(art)]) if result is None: break # backend unavailable / failed — stop the batch early vectors, model = result art.embedding = vectors[0] art.embedding_model = model art.embedded_at = datetime.now(UTC) embedded += 1 # Commit each article before the next embedding call # (#2193-5). Up to 100 serial LLM round-trips in ONE # transaction makes its lifetime the SUM of those calls; # committing per row also means a mid-batch backend outage # keeps the embeddings already computed instead of throwing # away the whole batch's API spend. await self.db.commit() return { "candidates": len(rows), "embedded": embedded, "pending": await self.count_pending_embeddings(), } async def drain_embeddings( self, *, batch: int = 100, max_total: int = 1000, ) -> dict[str, int]: """Run :meth:`backfill_embeddings` repeatedly until the backlog is empty (or the embeddings backend stops answering, or ``max_total`` vectors were produced this run). One bounded batch is not enough for a real library: a 300-doc Feishu mirror reported ``embedded=100`` on every sync — the per-call cap, which reads like a total that refuses to grow — and the leftovers only drained one click at a time (#2974). The loop keeps the per-row commits (short transactions, no wasted API spend on a mid-run outage) while making one click finish the job. """ embedded = 0 rounds = 0 pending = await self.count_pending_embeddings() while embedded < max_total and pending > 0: result = await self.backfill_embeddings( limit=min(batch, max_total - embedded), ) rounds += 1 embedded += result["embedded"] pending = result["pending"] # Backend unavailable / failing mid-batch: backfill_embeddings # breaks out early, so a zero-progress round means retrying would # spin forever against the same wall. if result["embedded"] == 0: break return {"embedded": embedded, "pending": pending, "rounds": rounds} async def bulk_draft_from_history(self, limit: int = 20) -> dict[str, Any]: """Seed the KB from resolved/closed tickets that don't yet have an article drafted from them. Each becomes a ``source=ai_ticket`` DRAFT (LLM-drafted when available, deterministic from the resolution otherwise) for admin review/publish. Bounded per call. Idempotent: a ticket already referenced by a KB article is skipped.""" from app.core.knowledge_loop.constants import ( CORPUS_SUPPORT_KB, SOURCE_INGEST_TICKET, ) from app.core.knowledge_loop.models import KnowledgeProposal from app.domains.support import kb_wiki from app.domains.support.copilot import draft_kb_from_ticket from app.domains.support.models import ( STATUS_CLOSED, STATUS_RESOLVED, SupportTicket, ) already = select(SupportKbArticle.source_ticket_id).where( SupportKbArticle.source_ticket_id.is_not(None) ) # A ticket that already produced a PROPOSAL must be skipped too — # otherwise every re-run re-asks the model about the same tickets and # (because the proposal deduplicates on content, not on ticket) quietly # burns the budget on work that changes nothing. proposed_tickets = { str(row[0] or "").removeprefix("ticket:") for row in ( await self.db.execute( select(KnowledgeProposal.source_ref).where( KnowledgeProposal.corpus == CORPUS_SUPPORT_KB, KnowledgeProposal.source == SOURCE_INGEST_TICKET, KnowledgeProposal.source_ref.is_not(None), ) ) ).all() } rows = ( ( await self.db.execute( select(SupportTicket) .where( SupportTicket.status.in_((STATUS_RESOLVED, STATUS_CLOSED)), SupportTicket.resolution.is_not(None), SupportTicket.id.not_in(already), ) .order_by(SupportTicket.resolved_at.desc()) .limit(max(1, min(limit, 100))) ) ) .scalars() .all() ) if not rows: return {"candidates": 0, "drafted": 0, "remaining": 0} drafted = 0 decisions: dict[str, int] = {} deadline = time.monotonic() + _BULK_DRAFT_BUDGET_SECONDS for idx, ticket in enumerate(rows): if str(ticket.id) in proposed_tickets: continue # Wall-clock budget (#2230). Twenty serial 60s drafts is 20 # MINUTES worst case, and nginx cuts the response at # ``proxy_read_timeout 300s`` — so the un-bounded version # could only ever end as a 504 over work that was still # running. Stop early and report what is left; the operation # is idempotent (a ticket already sourced is skipped), so # pressing the button again continues where this stopped. if time.monotonic() > deadline: logger.info( "kb.bulk_draft: wall-clock budget reached after %d/%d " "tickets — returning partial; re-run to continue", drafted, len(rows), ) return { "candidates": len(rows), "drafted": drafted, "remaining": len(rows) - idx, "decisions": decisions, } draft = await draft_kb_from_ticket(self.db, ticket) # Through the knowledge loop rather than straight to a new draft. # A resolved ticket is usually about something already written # down, and the old shape could only ever produce a SECOND page # about it — which is how a corpus grows without its recall # improving. The loop asks the index first, so the common outcome # is an alias or an appended section on the page that already # exists. When no model backend is configured it falls back to # exactly the previous behaviour: one new-article proposal. outcome = await kb_wiki.propose_from_knowledge( self.db, title=draft["title"], body=draft["body"], category=draft["category"], source=SOURCE_INGEST_TICKET, source_ref=f"ticket:{ticket.id}", evidence={ "ticket_id": str(ticket.id), "ticket_title": (ticket.title or "")[:200], }, ) decisions[outcome.decision] = decisions.get(outcome.decision, 0) + 1 drafted += 1 # Commit per draft (#2230, the shape #2193-5 already fixed in # the sibling ``backfill_embeddings``). Two reasons: the # "already drafted" de-dup query reads COMMITTED rows, so an # uncommitted batch means a natural retry click while run #1 # is still in flight drafts EVERY ticket twice # (``source_ticket_id`` has no unique constraint); and a # single transaction spanning ≤20 LLM round-trips is the # long-transaction shape this repo has been burned by. await self.db.commit() return { "candidates": len(rows), "drafted": drafted, "remaining": 0, "decisions": decisions, } @staticmethod def _apply_q(stmt, q: str | None): """Single-substring match — for the admin LIST filter (exact phrase).""" qc = (q or "").strip()[:100] if not qc: return stmt esc = qc.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") like = f"%{esc}%" return stmt.where( or_( SupportKbArticle.title.ilike(like, escape="\\"), SupportKbArticle.body.ilike(like, escape="\\"), SupportKbArticle.search_aliases.ilike(like, escape="\\"), ) ) @staticmethod def _apply_tokens(stmt, q: str | None): """Tokenised OR-match — for keyword SEARCH / deflection fallback, where the query is a whole problem description ("VPN 连不上 内网"). Each token OR-matches title/body, so a long query still hits a relevant article; the full-phrase substring match almost never would.""" qc = (q or "").strip() tokens = [t for t in qc.split() if len(t) >= 2][:8] if not tokens: return SupportKbService._apply_q(stmt, qc) conds = [] for tok in tokens: esc = tok.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") like = f"%{esc}%" conds.append(SupportKbArticle.title.ilike(like, escape="\\")) conds.append(SupportKbArticle.body.ilike(like, escape="\\")) conds.append(SupportKbArticle.search_aliases.ilike(like, escape="\\")) return stmt.where(or_(*conds)) async def run_kb_embed(article_id) -> None: """Background entrypoint — fire-and-forget after an article is published / its text edited (so the publish response isn't blocked on embedding latency). Opens a fresh session, re-embeds only if still published + not already embedded, commits. Exceptions caught/logged by spawn().""" from app.core.database import async_session_factory async with async_session_factory() as db: article = await db.get(SupportKbArticle, article_id) if article is None or article.status != KB_PUBLISHED: return if article.embedding is not None: return # already embedded (idempotent) if await SupportKbService(db).embed_article(article): await db.commit() async def run_kb_drain() -> None: """Background entrypoint — drain the whole embedding backlog. The per-article :func:`run_kb_embed` is right after ONE publish; a bulk import creates a hundred at a time, and a hundred fire-and-forget tasks all calling the same serial embeddings backend is the wrong shape. ``drain_embeddings`` already commits per row and stops early when the backend is unavailable, so this is safe to fire and forget. """ from app.core.database import async_session_factory async with async_session_factory() as db: result = await SupportKbService(db).drain_embeddings() logger.info("kb.drain: embedded=%(embedded)s pending=%(pending)s", result) __all__ = ["SupportKbService", "run_kb_drain", "run_kb_embed"]