221 lines
9.0 KiB
Python
221 lines
9.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""iAOP-Core · LLM 网关 —— 混合网关主编排（EPIC #6 主体交付）。
|
||
|
||
对应 PRD 5.4「④ LLM 网关 + RAG」与 EPIC #6（Issue #48 DLP / #46 RAG 模板化
|
||
已完成，本模块为其上层编排）：
|
||
|
||
用户提问 → 敏感度路由 → 本地/云端生成 → RAG 溯源校验 → 返回带引用答案
|
||
└──── DLP 出站检查（fail-closed，拦截即转本地/人工）────┘
|
||
|
||
`LLMGateway.ask()` 串起四个可插拔组件：
|
||
- `dlp`（DlpEngine）：出站防线，云端通道必经检查；
|
||
- `router`（SensitivityRouter）：敏感度分级路由（local / cloud / block）；
|
||
- `prompts`（PromptRegistry）：提示词模板版本绑定（可复现）；
|
||
- `guard`（HallucinationGuard）：引用溯源 + 信度阈值 → 人工确认；
|
||
- `backends`（LocalBackend / CloudBackend）：推理后端抽象（可注入）。
|
||
|
||
设计说明：
|
||
- 本版提供**编排闭环 + 后端抽象接口**，本地 70B / 云端 API 的具体接入
|
||
由子任务 #44 / #45 实现；`LocalBackend` / `CloudBackend` 默认内置一个
|
||
最小实现（返回固定占位答案 + 回显引用），供端到端测试与演示。
|
||
|
||
测试：`python -m unittest discover -s tests -v`（在 core/llm-gateway 目录下执行）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from typing import Callable, Dict, List, Optional, Sequence
|
||
|
||
from .dlp import DlpEngine
|
||
from .router import RouteDecision, RouteTarget, SensitivityRouter
|
||
from .prompts import PromptRegistry
|
||
from .hallucination import GuardVerdict, HallucinationGuard
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 推理后端抽象（Issue #44 / #45 将实现具体后端，业务代码只依赖本接口）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class InferenceBackend:
|
||
"""推理后端接口抽象（对齐 PRD 5.6 InferenceBackend 思想）。
|
||
|
||
业务代码只依赖本接口，不感知具体硬件/厂商；切换后端 = 换实现。
|
||
子任务 #44（本地 70B）、#45（云端 Qwen/DeepSeek）将各自实现本接口。
|
||
"""
|
||
|
||
name: str = "base"
|
||
|
||
def generate(self, prompt: str, context: Sequence[str]) -> str:
|
||
"""根据 prompt 与 RAG 上下文生成回答。子类实现。"""
|
||
raise NotImplementedError
|
||
|
||
|
||
class LocalBackend(InferenceBackend):
|
||
"""本地 70B 后端占位实现：数据不出厂（敏感/核心走此通道）。
|
||
|
||
子任务 #44 将替换为真实本地模型推理封装（vLLM/TGI 等）。
|
||
"""
|
||
|
||
name = "local-70b"
|
||
|
||
def __init__(self, echo_context: bool = True) -> None:
|
||
self.echo_context = echo_context
|
||
|
||
def generate(self, prompt: str, context: Sequence[str]) -> str:
|
||
head = f"[本地70B占位] {prompt[:40]}"
|
||
refs = ""
|
||
if self.echo_context:
|
||
for i, src in enumerate(context[:3], 1):
|
||
refs += f"\n[来源: {src}]"
|
||
return head + refs
|
||
|
||
|
||
class CloudBackend(InferenceBackend):
|
||
"""云端 API 后端占位实现：仅接收 DLP 放行的脱敏/通用内容。
|
||
|
||
子任务 #45 将替换为 Qwen/DeepSeek API 接入 + 安全网关。
|
||
"""
|
||
|
||
name = "cloud-api"
|
||
|
||
def __init__(self, echo_context: bool = True) -> None:
|
||
self.echo_context = echo_context
|
||
|
||
def generate(self, prompt: str, context: Sequence[str]) -> str:
|
||
head = f"[云端API占位] {prompt[:40]}"
|
||
refs = ""
|
||
if self.echo_context:
|
||
for i, src in enumerate(context[:3], 1):
|
||
refs += f"\n[来源: {src}]"
|
||
return head + refs
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 网关输出
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GatewayResult:
|
||
"""一次 ask() 的完整结果（含中间决策，便于审计与验收）。"""
|
||
|
||
query: str
|
||
answer: str
|
||
route: RouteDecision
|
||
verdict: GuardVerdict
|
||
answer_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
|
||
created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
||
|
||
@property
|
||
def needs_human(self) -> bool:
|
||
"""是否需转人工确认（路由 block 或校验 human_review/unsupported）。"""
|
||
return (self.route.target == RouteTarget.BLOCK
|
||
or self.verdict.action in ("human_review", "unsupported"))
|
||
|
||
def to_dict(self) -> Dict[str, object]:
|
||
return {
|
||
"answer_id": self.answer_id,
|
||
"created_at": self.created_at,
|
||
"query": self.query,
|
||
"answer": self.answer,
|
||
"route": self.route.to_dict(),
|
||
"verdict": self.verdict.to_dict(),
|
||
"needs_human": self.needs_human,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 混合网关主编排
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class LLMGateway:
|
||
"""混合 LLM 网关主编排：路由 → 生成 → 溯源校验 → DLP 出站防线。
|
||
|
||
构造参数均可注入（默认带内置 DLP 保底规则 + 空 router/prompts/guard）。
|
||
"""
|
||
|
||
def __init__(self,
|
||
dlp: Optional[DlpEngine] = None,
|
||
router: Optional[SensitivityRouter] = None,
|
||
prompts: Optional[PromptRegistry] = None,
|
||
guard: Optional[HallucinationGuard] = None,
|
||
local: Optional[InferenceBackend] = None,
|
||
cloud: Optional[InferenceBackend] = None,
|
||
prompt_name: str = "qa",
|
||
prompt_version: Optional[str] = None,
|
||
high_stakes_names: Optional[List[str]] = None) -> None:
|
||
self.dlp = dlp or DlpEngine()
|
||
self.router = router or SensitivityRouter()
|
||
self.prompts = prompts or PromptRegistry()
|
||
self.guard = guard or HallucinationGuard()
|
||
self.local = local or LocalBackend()
|
||
self.cloud = cloud or CloudBackend()
|
||
self.prompt_name = prompt_name
|
||
self.prompt_version = prompt_version
|
||
# 高利害提示词：命中即启用信度阈值（处置建议 / 报警解释等）
|
||
self.high_stakes_names = set(high_stakes_names or [])
|
||
|
||
# -- 主编排入口 --------------------------------------------------------
|
||
|
||
def ask(self, query: str,
|
||
rag_context: Optional[Sequence[str]] = None,
|
||
confidence: float = 1.0) -> GatewayResult:
|
||
"""完整处理一次用户提问。
|
||
|
||
`rag_context`：RAG 检索命中的文档标题列表（溯源校验用）；
|
||
`confidence`：模型输出信度（0~1，高利害场景低于阈值转人工）。
|
||
"""
|
||
rag_context = list(rag_context or [])
|
||
# 1) DLP 出站检查：query 敏感即拦截（fail-closed，云端不可达）
|
||
dlp_result = self.dlp.check_outbound({"query": query})
|
||
# 2) 敏感度路由（含 DLP 结果 → block）
|
||
decision = self.router.route(query, dlp_blocked=dlp_result.blocked)
|
||
|
||
# 3) 选择后端与提示词版本（运行时绑定，可复现）
|
||
prompt = self.prompts.get(self.prompt_name, self.prompt_version)
|
||
backend = self.local if decision.target != RouteTarget.CLOUD else self.cloud
|
||
|
||
# 4) 生成（block 时也不调用后端，直接给出人工确认占位答案）
|
||
if decision.target == RouteTarget.BLOCK:
|
||
answer = "该请求已拦截（敏感度路由/规则触发），请转人工确认处理。"
|
||
backend_name = "none"
|
||
else:
|
||
rendered = prompt.render(query=query)
|
||
answer = backend.generate(rendered, rag_context)
|
||
backend_name = backend.name
|
||
|
||
# 5) 幻觉/事实性校验（引用溯源 + 高利害信度阈值）
|
||
high_stakes = prompt.name in self.high_stakes_names
|
||
verdict = self.guard.check(
|
||
answer=answer, sources=rag_context,
|
||
confidence=confidence, high_stakes=high_stakes,
|
||
)
|
||
|
||
# 6) 出站前最终 DLP 防线（模型输出若含敏感内容：云端通道拦截）
|
||
if decision.target == RouteTarget.CLOUD:
|
||
outbound = self.dlp.check_outbound({"output": answer})
|
||
if outbound.blocked:
|
||
answer = "输出经 DLP 复查拦截，已转本地/人工处理。"
|
||
|
||
return GatewayResult(
|
||
query=query, answer=answer, route=decision, verdict=verdict,
|
||
)
|
||
|
||
# -- 审计汇总 ----------------------------------------------------------
|
||
|
||
def drain_audits(self) -> Dict[str, List[Dict[str, object]]]:
|
||
"""取走各组件审计记录（DLP / 路由 / Prompt / 幻觉校验）。"""
|
||
return {
|
||
"dlp": self.dlp.drain_audit(),
|
||
"router": self.router.drain_audit(),
|
||
"prompts": self.prompts.drain_audit(),
|
||
"guard": self.guard.drain_audit(),
|
||
}
|
||
|
||
def __repr__(self) -> str: # pragma: no cover - 调试辅助
|
||
return (f"<LLMGateway router={self.router!r} prompts={self.prompts!r} "
|
||
f"guard={self.guard!r}>")
|