"""安全问卷 · 乙方侧 AI 辅助作答 — match inbound questions to the answer library. For each unanswered inbound question, ask the Bailian (Qwen) model to pick the best-matching answer-library entry (or none) and draft an answer grounded in that entry. Gated on the platform LLM key — no key → ``AppException(BAILIAN_LLM_NO_KEY, 503)``, which the router surfaces so the UI degrades to fully-manual answering (the answer library + search still work without the LLM). The AI only DRAFTS: every answer stays ``status='drafted'`` until a human reviews it. We never auto-mark a drafted answer as reviewed. """ from __future__ import annotations import logging from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.domains.auditor.models import AuditorAnswerLibraryEntry from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) _KEY_API = "ai_spending.llm.api_key" _KEY_MODEL = "ai_spending.llm.model" _KEY_REGION = "ai_spending.llm.region" # Cap how many questions go in one LLM call — keeps the prompt bounded and # the JSON response parseable. Larger batches are chunked by the caller. _BATCH = 12 # Cap library entries in the prompt (most-used first) to bound tokens. _LIB_CAP = 120 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 build_llm_client(db: AsyncSession, *, timeout: float | None = None): """The platform LLM client, or a 503 when no key is configured. Public because the employee-facing 安全 FAQ ask path (``security_faq _service``) grounds itself on the same answer library and must reach the same provider configuration. That caller catches the no-key exception and degrades to keyword-only search rather than surfacing it — the gate is fail-closed here and softened there, deliberately. """ from app.domains.bailian.llm_client import BailianLLMClient api_key = (await _setting(db, _KEY_API) or "").strip() if not api_key: raise AppException( code="BAILIAN_LLM_NO_KEY", message=( "未配置百炼(DashScope) API Key,无法 AI 辅助作答。请在 管理 → " "连接器 → AI / LLM 配置后重试;在此之前可手动检索答案库作答。" ), status_code=503, ) model = await _setting(db, _KEY_MODEL) region = await _setting(db, _KEY_REGION) if timeout is not None: # A caller that runs INSIDE a request the user is waiting on passes # its own bound. The 90s default is right for questionnaire drafting # (a background-ish admin action) and wrong for the 安全 FAQ ask # path, where a slow provider must degrade to "no lead answer" well # before the browser gives up. return BailianLLMClient( api_key=api_key, model=model, region=region, timeout=timeout, purpose="auditor.inbound_questionnaire") return BailianLLMClient(api_key=api_key, model=model, region=region, purpose="auditor.inbound_questionnaire") _SYSTEM = ( "你是一名资深信息安全合规专家,负责代表本公司(乙方)回复客户发来的安全问卷。" "给定客户的若干问题和一份本公司的『答案库』(每条含 id、标准问题、标准答案)," "请为每个问题:① 从答案库中选出语义最匹配的一条(给出其 id 与 0-100 的置信度)," "若无合适条目则 matched_id 置为 null、置信度 0;② 基于匹配到的标准答案,起草一段" "面向该客户问题的、专业且准确的回答草稿(沿用答案库措辞,不要编造未提供的事实;" "语言与客户问题一致)。严格输出单个 JSON 对象。" ) def _lib_payload(entries: list[AuditorAnswerLibraryEntry]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for e in entries[:_LIB_CAP]: out.append({ "id": str(e.id), "category": e.category, "question": e.question_en or e.question_zh, "question_zh": e.question_zh, "answer": e.answer_en or e.answer_zh, "answer_zh": e.answer_zh, }) return out async def match_and_draft( db: AsyncSession, *, questions: list[dict[str, Any]], library: list[AuditorAnswerLibraryEntry], ) -> dict[str, dict[str, Any]]: """Return ``{question_id: {matched_id, confidence, draft}}``. ``questions`` = ``[{"id": str, "text": str}, ...]``. Chunks into batches; a batch that fails to parse is skipped (its questions simply get no draft — the caller leaves them pending). Never raises on a per-batch model hiccup; only the no-key gate raises (fail-closed at build time). If EVERY batch fails, raises ``AppException`` so the caller/UI sees a real error instead of a silent ``drafted:0`` success. """ client = await build_llm_client(db) lib = _lib_payload(library) valid_ids = {e["id"] for e in lib} results: dict[str, dict[str, Any]] = {} batches = 0 failed_batches = 0 last_error: Exception | None = None for start in range(0, len(questions), _BATCH): batches += 1 chunk = questions[start:start + _BATCH] user_prompt = ( "答案库(JSON 数组):\n" + _dump(lib) + "\n\n客户问题(JSON 数组,每条含 id 与 text):\n" + _dump([{"id": q["id"], "text": q["text"]} for q in chunk]) + "\n\n请输出 JSON:{\"answers\":[{\"id\":问题id," "\"matched_id\":答案库id或null,\"confidence\":0-100整数," "\"draft\":\"回答草稿\"}]}" ) try: resp = await client.chat_json(system=_SYSTEM, user=user_prompt) except Exception as exc: # noqa: BLE001 — one bad batch must not sink the run logger.warning("inbound AI batch failed: %s", exc) failed_batches += 1 last_error = exc continue for item in (resp or {}).get("answers", []) or []: qid = str(item.get("id", "")).strip() if not qid: continue matched = item.get("matched_id") matched = str(matched) if matched and str(matched) in valid_ids else None try: conf = int(item.get("confidence", 0) or 0) except (TypeError, ValueError): conf = 0 results[qid] = { "matched_id": matched, "confidence": max(0, min(conf, 100)), "draft": (item.get("draft") or "").strip(), } # Every batch blew up — that's a real failure (wrong call signature, # bad key, provider outage, unparseable JSON), not "the model matched # nothing". Surface it instead of returning {} as a silent success. if batches and failed_batches == batches: raise AppException( code="INBOUND_AI_DRAFT_FAILED", message=( "AI 辅助作答失败:所有批次均未成功。请稍后重试,或检查" " AI / LLM 连接器配置。" ), status_code=502, ) from last_error return results def _dump(obj: Any) -> str: import json return json.dumps(obj, ensure_ascii=False)