"""安全 FAQ 的检索增强(RAG)—— 语义召回 + 有据可查的成文回答. 员工的原话是「这个搜索很蠢」。它不是错觉,也不是权重没调好:答案库里的 71 条全部是按**客户安全问卷**的口吻写的 ——「贵公司是否部署数据防泄漏 (DLP)方案?」——而员工问的是「我能用个人电脑写公司代码吗」。两句话没有 任何公共子串,LIKE 找不到;二字切词(``query_terms``)能救回一部分,但它 救的是**用词碰巧重合**的那部分,救不了「同一件事换一种说法」。这是词法 检索的能力边界,不是参数问题。 这个模块补的就是那条能力边界,分两层,顺序不能反: **① 召回(retrieval)—— 词法 + 语义混合.** 每条启用且对员工可见的条目嵌一个向量(双语问句加权 + 标签 + 答案摘要, 一条一个向量);提问也嵌一个;余弦排序。语义召回与既有的词法打分用 **RRF(Reciprocal Rank Fusion)** 融合,而不是「词法没命中才走语义」—— 后者是这个页面原来的形状,意味着一个词法**勉强**命中(问句里碰巧有 「公司」)就永远看不到语义那一路更好的答案。RRF 只看名次不看分数,因此 不需要把两套量纲不可比的分数硬拉到一起。 **② 成文(generation)—— 但只准照着给定条目说话.** 召回回来的是条目,员工要的是答案。所以模型在这里第一次被允许写字 —— 严格限制在:只能用我们递给它的条目原文,必须给出所引条目的 id,不准引入 任何条目里没有的事实,答不了就明说答不了。服务端还有一道闸:模型给的 引用 id 必须落在这次召回的集合里,一条都不落就整段丢弃(降级成「只列条目」)。 **条目原文永远原样展示在成文答案下面**,所以读者随时能对账 —— 这是把 「模型说的」和「安全团队写的」分开的唯一诚实做法。 降级链是完整的,每一环断了都还能用: 没配 LLM Key → 没有向量 → 纯词法检索(今天的行为); 有向量、成文调用失败 → 只列召回到的条目; 模型说「答不了」 → 记缺口台账,照旧引导到工单。 向量存 JSON 浮点数组、Python 算余弦(与 ``support_kb_articles`` 同款,见 迁移 561):库只有几百条精编条目,全表打分远比维护向量索引便宜,而且本地 MySQL 测试一行不用改。 Naive-UTC datetime discipline (codebase-wide MySQL/TiDB convention). """ from __future__ import annotations import logging import math import time import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.auditor.models import AuditorAnswerLibraryEntry from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) # The AI/LLM connector keys, spelled the same way ``inbound_ai`` spells them: # one place to configure the platform's model access (管理 → 连接器 → AI / LLM). # Kept as literals rather than imported to avoid a cycle, matching the # existing precedent in this domain. _KEY_API = "ai_spending.llm.api_key" _KEY_MODEL = "ai_spending.llm.model" _KEY_REGION = "ai_spending.llm.region" _KEY_EMBED_MODEL = "ai_spending.llm.embed_model" #: Cap on the text sent to the embeddings API for one entry. Answers here run #: long (a DLP answer is a paragraph per track); the question carries most of #: the retrieval signal, so the answer is a tail, not the body. _EMBED_TEXT_MAX = 3000 #: Minimum cosine for a semantic hit to be fused at all. text-embedding-v3 #: scores unrelated CN/EN text around 0.1–0.3 and genuinely related text ≳0.4 — #: the same threshold the 求助域 knowledge base runs on, for the same model. #: Without a floor, RRF would happily rank the least-unrelated row first: every #: query returns *something*, and rank alone cannot tell "closest" from "close". _SEMANTIC_MIN_SCORE = 0.4 #: Confidence tiers. Retrieval already knows how sure it is — the cosine it #: ranked by, and whether the two independent channels agreed — and throwing #: that away is what makes a result list read as an oracle. Two things are done #: with it: the reader is told, and the model is NOT asked to summarise #: material the retrieval itself considers weak. #: #: The thresholds are the same scale as ``_SEMANTIC_MIN_SCORE`` (which is the #: floor for being retrievable at all): text-embedding-v3 puts unrelated CN/EN #: text at 0.1–0.3, related at ≳0.4, and closely-on-topic at ≳0.65. _CONFIDENCE_HIGH_COSINE = 0.65 _CONFIDENCE_MEDIUM_COSINE = 0.5 #: A lexical hit this near the top is a strong signal on its own: the reader's #: words are literally in the question. Deeper than this, a lexical-only hit is #: a term-overlap coincidence — which is the failure this whole feature exists #: to fix, so it is not allowed to read as confident. _CONFIDENCE_LEXICAL_TOP_RANK = 3 TIER_HIGH = "high" TIER_MEDIUM = "medium" TIER_LOW = "low" def confidence_tier( *, channels: list[str], cosine: float | None, lexical_rank: int | None, ) -> str: """How sure retrieval is about ONE entry, from the evidence it already has. Agreement between two independent channels is the strongest signal available here and outranks either channel's own score: the lexical matcher and the embedding model share no mechanism, so both of them surfacing the same row is not a coincidence either can manufacture. """ if len(channels) > 1: return TIER_HIGH if cosine is not None: if cosine >= _CONFIDENCE_HIGH_COSINE: return TIER_HIGH if cosine >= _CONFIDENCE_MEDIUM_COSINE: return TIER_MEDIUM return TIER_LOW if lexical_rank is not None and lexical_rank <= _CONFIDENCE_LEXICAL_TOP_RANK: return TIER_MEDIUM return TIER_LOW #: RRF constant. 60 is the value from the original Cormack et al. formulation #: and the one every mainstream hybrid-search implementation uses; it flattens #: the head enough that a rank-1 hit in ONE channel cannot outrank a document #: both channels agree on. _RRF_K = 60 #: How many candidates each channel contributes to the fusion. _CHANNEL_DEPTH = 20 #: Corpus vector cache: ``{"visible": (signature, fetched_at, {id: vector})}``. #: One entry — a corpus cache, not a query cache, so it cannot grow. _vector_cache: dict[str, tuple[tuple[int, str], float, dict[uuid.UUID, list[float]]]] = {} #: Backstop on the signature probe. ``updated_at`` is second-precision, so two #: edits inside one second with an unchanged count produce one signature. _VECTOR_CACHE_TTL_SECONDS = 300.0 #: Bound the embeddings HTTP call on the ask path — an employee is waiting on #: it. A slow provider must degrade to "lexical only" fast, not hold the request. _QUERY_EMBED_TIMEOUT = 15.0 #: Bound the composition call. Longer than the embedding hop (it generates #: prose) but far below the provider's own 90s default: this now runs on EVERY #: hit, with a person watching a spinner, so a degraded provider must fall #: back to "entries without a lead paragraph" while the browser is still #: listening. Public — ``security_faq_service`` passes it when it builds the #: client, so the bound lives next to the reason for it. COMPOSE_TIMEOUT = 40.0 #: Wall-clock ceiling for one backfill request driven from the admin button. #: nginx cuts the hop at ``proxy_read_timeout 300s`` and the client aborts #: just under that, so a run that cannot finish returns PARTIAL progress plus #: the remaining backlog rather than dying as a 504 over vectors that did #: land. Each embedded row is committed before the next call, so "partial" #: means partial-and-durable; pressing the button again continues. _DRAIN_BUDGET_SECONDS = 240.0 def reset_vector_cache() -> None: """Drop the cached corpus. Called by tests and after a bulk re-embed.""" _vector_cache.clear() def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) # ── Configuration ───────────────────────────────────────────────────── async def _setting(db: AsyncSession, key: str) -> str | None: row = await db.get(PlatformSetting, key) v = row.value if row else None return v if isinstance(v, str) else None async def resolve_embed_model(db: AsyncSession) -> str: """The embedding model id currently configured (or the built-in default). Reads config only — no network call — so the pending-backfill counter can ask "which rows are stale under the ACTIVE model" without a provider round trip. """ from app.domains.bailian.llm_client import DEFAULT_EMBED_MODEL return (await _setting(db, _KEY_EMBED_MODEL) or "").strip() or DEFAULT_EMBED_MODEL async def embed_texts( db: AsyncSession, texts: list[str], *, timeout: float = 60.0, ) -> tuple[list[list[float]], str] | None: """``(vectors, model_id)`` or ``None`` when embeddings are unavailable. ``None`` — not an exception — is the whole contract: no key configured, a provider outage, a short vector count. Every caller degrades to lexical search, which is the behaviour this page shipped with. """ if not texts: return None api_key = (await _setting(db, _KEY_API) or "").strip() if not api_key: return None model = await resolve_embed_model(db) try: from app.domains.bailian.llm_client import BailianLLMClient client = BailianLLMClient( api_key=api_key, model=await _setting(db, _KEY_MODEL), region=await _setting(db, _KEY_REGION), timeout=timeout, purpose="security.faq_rag") vectors = await client.embed(texts, model=model) except Exception as exc: # noqa: BLE001 — enhancement, never a gate logger.info("security FAQ: embedding call failed (%s)", exc) return None if not vectors or len(vectors) != len(texts): # A short reply would silently misalign vector→text. Refuse it. logger.warning( "security FAQ: embeddings returned %d vectors for %d inputs", len(vectors or []), len(texts), ) return None return vectors, model # ── Indexing ────────────────────────────────────────────────────────── def embedding_text(e: AuditorAnswerLibraryEntry) -> str: """The text that represents one entry in vector space. BOTH languages go into ONE vector rather than one vector per language. text-embedding-v3 is multilingual and places a sentence near its own translation, so a bilingual body is not noise — and an employee reading the English UI who types a Chinese term (which is exactly how people search for 「飞连」 or 「等保」) must still find the entry. Two vectors per row would double the storage, the API spend and the re-embed surface to buy a distinction the reader does not have. Questions are repeated because they carry the retrieval signal: an entry is a question people ask, and the answer is a paragraph that mentions many neighbouring topics in passing. Tags and the category key join them — they are the security team's own keywords for the row. """ tags = " ".join(e.tags or []) # Employee phrasings sit WITH the questions, and are repeated the same way, # because they are the same signal in a different register: an entry is a # question people ask, and these are the words the people asking actually # use. Putting them next to the answer instead would bury them under a # paragraph that mentions a dozen neighbouring topics in passing. phrasings = " ".join( line.strip() for line in (e.employee_phrasings or "").splitlines() if line.strip() ) asked = f"{e.question_zh}\n{e.question_en}" if phrasings: asked = f"{asked}\n{phrasings}" return ( f"{asked}\n{asked}\n" f"{e.category} {tags}\n{e.answer_zh}\n{e.answer_en}" )[:_EMBED_TEXT_MAX] def pending_clause(active_model: str): """"Needs a vector under ``active_model``" — shared by the counter and the backfill so the number an admin reads can never disagree with the rows the backfill will actually pick up. The explicit ``embedding_model IS NULL`` arm is load-bearing: in SQL ``embedding_model != :active`` evaluates to NULL (not true) on a NULL column, so a row carrying a vector of unknown provenance would otherwise keep it forever after the active model changed. """ return and_( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), or_( AuditorAnswerLibraryEntry.embedding.is_(None), AuditorAnswerLibraryEntry.embedding_model.is_(None), AuditorAnswerLibraryEntry.embedding_model != active_model, ), ) async def count_pending(db: AsyncSession) -> int: """How many employee-visible entries still need a vector.""" active = await resolve_embed_model(db) return int((await db.execute( select(func.count()).select_from(AuditorAnswerLibraryEntry).where( pending_clause(active), ) )).scalar_one()) async def embed_entry(db: AsyncSession, e: AuditorAnswerLibraryEntry) -> bool: """Compute + store one entry's vector. Never raises — an entry that fails to embed is simply not semantically searchable yet.""" result = await embed_texts(db, [embedding_text(e)]) if result is None: return False vectors, model = result e.embedding = vectors[0] e.embedding_model = model e.embedded_at = _naive_now() return True async def backfill(db: AsyncSession, *, limit: int = 50) -> dict[str, int]: """Embed up to ``limit`` entries that need it. ``pending`` in the result is the backlog left AFTER this batch — the cap is reported, never silent. Commits per row: a batch of serial HTTP round-trips inside one transaction makes that transaction's lifetime the SUM of the calls, and a mid-batch provider outage would throw away every vector already paid for. """ active = await resolve_embed_model(db) rows = list((await db.execute( select(AuditorAnswerLibraryEntry) .where(pending_clause(active)) .order_by(AuditorAnswerLibraryEntry.updated_at.desc()) .limit(max(1, min(limit, 200))) )).scalars().all()) if not rows: return {"candidates": 0, "embedded": 0, "pending": 0} embedded = 0 for e in rows: if not await embed_entry(db, e): # Backend unavailable / failing — stop rather than burn the batch # against the same wall. break embedded += 1 await db.commit() if embedded: reset_vector_cache() return { "candidates": len(rows), "embedded": embedded, "pending": await count_pending(db), } async def drain( db: AsyncSession, *, batch: int = 50, max_total: int = 1000, budget_seconds: float = _DRAIN_BUDGET_SECONDS, ) -> dict[str, int]: """Run :func:`backfill` until the backlog is empty, the provider stops answering, ``max_total`` vectors were produced, or the wall-clock budget runs out. One bounded batch is not enough for a real library: a seed migration lands dozens of entries at once and a per-call cap reads like a total that refuses to grow. A budget is not optional either — this is reachable from an admin button, and a few hundred serial provider round trips would sail past nginx's 300s hop and return a 504 over vectors that were in fact written. ``pending`` in the result is the honest remainder either way. """ started = time.monotonic() embedded = 0 rounds = 0 timed_out = False pending = await count_pending(db) while embedded < max_total and pending > 0: if time.monotonic() - started > budget_seconds: timed_out = True break result = await backfill(db, limit=min(batch, max_total - embedded)) rounds += 1 embedded += result["embedded"] pending = result["pending"] if result["embedded"] == 0: break return { "embedded": embedded, "pending": pending, "rounds": rounds, "timed_out": timed_out, } async def index_stats(db: AsyncSession) -> dict[str, Any]: """What the semantic index currently covers, for the admin console. ``pending`` is computed against the ACTIVE model, so switching models makes the whole library show as pending — which is the truth: those vectors can no longer be compared with a query embedded by the new model. """ active = await resolve_embed_model(db) visible = int((await db.execute( select(func.count()).select_from(AuditorAnswerLibraryEntry).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ) )).scalar_one()) pending = await count_pending(db) last = (await db.execute( select(func.max(AuditorAnswerLibraryEntry.embedded_at)).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), ) )).scalar_one_or_none() if last is not None and last.tzinfo is not None: last = last.astimezone(UTC).replace(tzinfo=None) return { "model": active, "visible": visible, "indexed": max(0, visible - pending), "pending": pending, "last_embedded_at": last.isoformat() if last else None, } # ── Retrieval ───────────────────────────────────────────────────────── def cosine(a: list[float], b: list[float]) -> float: """Cosine similarity. 0.0 for a length mismatch or a zero vector — this is a ranking signal, it must never raise on a malformed stored row.""" 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) async def corpus_vectors(db: AsyncSession) -> dict[uuid.UUID, list[float]]: """The employee-visible 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 that could alter the corpus moves one of them: publishing, retracting, editing and re-embedding all touch ``updated_at`` (``onupdate``), and an added or removed row moves the count. Only ``(id, embedding)`` is selected. Hydrating whole ORM rows here would pull every answer body — megabytes of prose — on every question asked, to show at most five. """ probe = (await db.execute( select(func.count(), func.max(AuditorAnswerLibraryEntry.updated_at)) .select_from(AuditorAnswerLibraryEntry) .where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), AuditorAnswerLibraryEntry.embedding.is_not(None), ) )).one() signature = (int(probe[0] or 0), str(probe[1])) cached = _vector_cache.get("visible") if ( cached is not None and cached[0] == signature and time.monotonic() - cached[1] < _VECTOR_CACHE_TTL_SECONDS ): return cached[2] rows = (await db.execute( select(AuditorAnswerLibraryEntry.id, AuditorAnswerLibraryEntry.embedding).where( AuditorAnswerLibraryEntry.status == "active", AuditorAnswerLibraryEntry.employee_visible.is_(True), AuditorAnswerLibraryEntry.embedding.is_not(None), ) )).all() vectors = {r[0]: r[1] for r in rows if isinstance(r[1], list) and r[1]} _vector_cache["visible"] = (signature, time.monotonic(), vectors) logger.info( "security FAQ: cached %d entry vectors (signature=%s)", len(vectors), signature, ) return vectors async def semantic_ranked( db: AsyncSession, *, query: str, limit: int = _CHANNEL_DEPTH, ) -> list[tuple[uuid.UUID, float]] | None: """``[(entry_id, cosine)]`` above the score floor, best first. ``None`` means "semantic retrieval is not available" (no key, provider down, nothing embedded yet) and is different from ``[]``, which means "the corpus was searched and held nothing close enough". The caller treats the two the same way — fall through to lexical — but the distinction is what the console reports and what an admin needs to tell a dead index from a genuinely novel question. """ embedded = await embed_texts(db, [query], timeout=_QUERY_EMBED_TIMEOUT) if embedded is None: return None probe = embedded[0][0] vectors = await corpus_vectors(db) if not vectors: return None scored = [ (entry_id, cosine(probe, vec)) for entry_id, vec in vectors.items() ] scored.sort(key=lambda t: (-t[1], str(t[0]))) return [(i, s) for i, s in scored[:limit] if s >= _SEMANTIC_MIN_SCORE] def rrf_fuse( channels: dict[str, list[uuid.UUID]], *, limit: int, ) -> list[tuple[uuid.UUID, float, list[str]]]: """Reciprocal Rank Fusion of several ranked id lists. ``[(entry_id, score, [channels that found it])]``, best first. Score is ``Σ 1/(k + rank)`` over the channels that ranked the document. Rank, not score: a cosine of 0.62 and a term-weight of 14 are not comparable quantities, and every attempt to normalise them ends up encoding the shape of one particular corpus. RRF needs no calibration and has the property that matters here — a document BOTH channels found outranks one that only tops a single channel. """ scores: dict[uuid.UUID, float] = {} found_in: dict[uuid.UUID, list[str]] = {} for channel, ids in channels.items(): for rank, entry_id in enumerate(ids, start=1): scores[entry_id] = scores.get(entry_id, 0.0) + 1.0 / (_RRF_K + rank) found_in.setdefault(entry_id, []).append(channel) ordered = sorted( scores.items(), # Ties (the same document at the same rank in both channels is # impossible, but two documents CAN tie) break on the id so the order # is stable across processes — a result list that reshuffles between # two identical searches reads as a bug. key=lambda kv: (-kv[1], str(kv[0])), ) return [(i, s, found_in[i]) for i, s in ordered[:limit]] # ── Grounded composition ────────────────────────────────────────────── _COMPOSE_SYSTEM = ( "你是公司内部信息安全知识库的问答助手,面向本公司员工。" "你会收到一名员工的提问,以及若干条**信息安全团队撰写的标准问答条目**" "(每条含 id、标准问题、标准答案)。请只依据这些条目回答。\n" "硬性要求:\n" "① 只能使用条目中已有的事实。条目里没写的工具、流程、时限、联系人、" "例外情况一律不得补充,也不要用常识补全 —— 你写下的每一句都会被员工" "当作公司的正式规定。\n" "② 必须在 cited_ids 中列出你实际用到的条目 id;没用到的不要列。\n" "③ 检索是按字面与语义相似度召回的,里面**可能混进不对题的条目**。请在" "relevant_ids 中列出**确实与这个提问有关**的条目 id(可以多于 cited_ids:" "有关但你没用上的也算),把明显不对题的排除掉 —— 被排除的条目不会展示" "给员工。\n" "④ 如果这些条目**回答不了**这个提问,covered 置为 false、answer 置为空" "字符串 —— 宁可说答不了,也不要用相关但不对题的条目硬凑。\n" "⑤ 回答要直接、可执行:先给结论(能 / 不能 / 怎么做),再给必要的前提或" "步骤。不要复述提问,不要写「根据条目」之类的开场白,不要编号引用标记。\n" "⑥ 用与提问相同的语言作答。\n" "严格输出单个 JSON 对象。" ) #: Hard cap on the composed answer. The stored entries are shown in full below #: it; this is a lead, not a replacement for them. _ANSWER_MAX_CHARS = 900 async def compose_answer( db: AsyncSession, *, query: str, entries: list[AuditorAnswerLibraryEntry], lang: str, client: Any | None = None, ) -> dict[str, Any] | None: """A short grounded answer over ``entries``, or ``None``. Also the **rerank**: the model returns which of the retrieved entries are genuinely on topic (``relevant_ids``), and the caller drops the rest from what the reader sees. Retrieval is unsupervised — RRF fuses two similarity rankings and cannot tell "closest" from "correct" — so the list it produces reliably contains a tail of plausible-looking noise. Filtering it here costs nothing extra: the same call is already reading every one of these entries in order to write the lead. ``None`` covers every degradation — no key, provider outage, unparseable reply, a reply citing nothing we gave it — and means "show the entries without a lead paragraph", which is exactly the page's previous behaviour. ``{"covered": False}`` is different and deliberate: the model read the retrieved entries and said they do not answer the question. That is the single most useful thing this path can learn, because retrieval returning *something* is the normal case — the gap ledger would otherwise only ever see questions that matched literally nothing. The returned ``cited_ids`` are filtered to ids we actually supplied. A model that cites an id it invented is not grounded, and an id from another entry would render as a citation to an answer that never entered the prompt. """ if not entries: return None if client is None: try: from app.domains.auditor.inbound_ai import build_llm_client client = await build_llm_client(db) except Exception as exc: # noqa: BLE001 — degradation, not failure logger.info("security FAQ compose: LLM unavailable (%s)", exc) return None zh = lang == "zh" payload = [ { "id": str(e.id), "question": (e.question_zh if zh else e.question_en) or e.question_en, "answer": (e.answer_zh if zh else e.answer_en) or e.answer_en, } for e in entries ] allowed = {str(e.id) for e in entries} user_prompt = ( "标准问答条目(JSON 数组):\n" + _dump(payload) + f"\n\n员工提问:{query}\n\n" + '请输出 JSON:{"covered":true/false,"answer":"回答正文",' '"cited_ids":["用到的条目id", ...],' '"relevant_ids":["与提问有关的条目id", ...]}' ) try: resp = await client.chat_json( system=_COMPOSE_SYSTEM, user=user_prompt, max_tokens=800, ) except Exception as exc: # noqa: BLE001 — degradation, not failure logger.warning("security FAQ compose: LLM call failed (%s)", exc) return None if not isinstance(resp, dict): return None covered = bool(resp.get("covered")) text = str(resp.get("answer") or "").strip()[:_ANSWER_MAX_CHARS] def _ids(key: str) -> list[str]: raw = resp.get(key) picked = [ str(i).strip() for i in (raw if isinstance(raw, list) else []) if str(i).strip() in allowed ] # De-duplicate while keeping the model's order. ``dict.fromkeys`` # rather than the ``seen.add`` idiom: that one reads the return value # of a method that returns None, which is a real mypy error # (``func-returns-value``) and not merely a style opinion — CI's # ratchet caught it, and the same construct is already used a few # lines below for the same purpose. return list(dict.fromkeys(picked)) cited = _ids("cited_ids") # A model that answered but named nothing relevant is not reranking, it is # failing to fill a field. Fall back to "everything retrieved is kept" — # the pre-rerank behaviour — rather than blanking the result list. relevant = _ids("relevant_ids") or None if relevant is not None: # Anything cited is relevant by construction; a reply that cites an id # it left out of relevant_ids would otherwise have its own source # dropped out from under it. relevant = list(dict.fromkeys(relevant + cited)) if not covered: return { "covered": False, "text": "", "cited_ids": [], "relevant_ids": relevant, } if not text or not cited: # Prose with no traceable source is the one output this path must not # render: it looks exactly like a policy statement and nothing backs # it. Drop it and let the stored entries speak for themselves. logger.info( "security FAQ compose: dropping ungrounded answer (text=%s cited=%d)", bool(text), len(cited), ) return None return { "covered": True, "text": text, "cited_ids": cited, "relevant_ids": relevant, } def _dump(obj: Any) -> str: import json return json.dumps(obj, ensure_ascii=False) async def run_entry_embed(entry_id: uuid.UUID) -> None: """Background entrypoint — re-embed one entry after it was written. Fire-and-forget from the admin write path so creating or editing an answer is never blocked on (or failed by) an embeddings round trip. Opens its own session: the caller's has already been committed and closed by then. """ from app.core.database import async_session_factory async with async_session_factory() as db: entry = await db.get(AuditorAnswerLibraryEntry, entry_id) if entry is None or entry.status != "active" or not entry.employee_visible: return if await embed_entry(db, entry): await db.commit() reset_vector_cache() __all__ = [ "backfill", "compose_answer", "corpus_vectors", "cosine", "count_pending", "drain", "embed_entry", "embed_texts", "embedding_text", "index_stats", "pending_clause", "reset_vector_cache", "resolve_embed_model", "rrf_fuse", "run_entry_embed", "semantic_ranked", ]