第五章：数据安全与合规架构 —— 构建知识库前必读

三条绝对红线

[P0] 内部经营数据在没有完成数据映射、合法性依据、供应商条款、保留策略和书面审批前，不得发送到公网 LLM API。
[P0] 含 PII 数据未脱敏不得入向量库。
[P0] 知识库 API 暴露公网必须有认证 + 限流。

很多团队做知识库时，第一反应是“先把资料丢进去，再慢慢治理”。这在生产环境里是最危险的路径。知识库不是一个搜索盒子，而是一条数据处理链：采集、清洗、脱敏、嵌入、存储、查询、审计，每一步都可能形成合规责任。真正的顺序必须反过来：先定义安全边界，再决定技术栈。如果边界没画清，向量库、GraphRAG、本地 Agent 做得越快，风险暴露就越快。

司法辖区与责任边界

本章是工程控制清单，不构成法律意见。适用义务取决于组织角色、数据主体、部署地点、预期用途与行业规则；上线前必须由法务或隐私负责人确认适用辖区。NIST AI RMF 属于自愿风险管理框架，不能替代法律合规判断；EU AI Act 可能适用于在欧盟境内外、但把系统投放欧盟市场或在欧盟使用系统的主体。

本轮参考的基准截至 2026-08-01。NIST 官方页面明确提示 AI RMF 1.0 正在修订，因此这里不把版本号写成永久结论；欧盟规则按系统预期用途和风险分类适用，不能仅凭“用了大模型”判断风险级别。

5.1 数据分级体系：先分级，再决定模型和存储

建议把所有待入库数据先划为 L1-L4 四级。级别不是按“文件名看起来敏不敏感”，而是按泄露后的业务后果决定。

级别	数据定义	典型示例	允许工具	允许 LLM	存储要求	审计频率
L1 公开数据	对外公开、泄露无新增风险	Amazon 页面、品牌官网、行业报告	SaaS ETL / 云向量库 / 公网 API	公网 API 可用	常规加密存储即可	每月
L2 内部数据	内部运营资料，但不直接含敏感经营指标	产品目录、供应商列表、SOP 草稿	本地解析器、私有向量库	本地 LLM 优先，公网需审批	私网存储、最小权限	每两周
L3 敏感数据	含客户、员工、客服、订单侧敏感信息	客服记录、客户信息、售后工单	本地解析器、PII 检测器、私有向量库	仅本地 LLM，且必须先脱敏	加密 + 字段脱敏 + 审计日志	每周
L4 核心经营	直接影响经营决策与利润模型	销售额、毛利率、定价策略	物理隔离集群、本地推理、离线导入	仅本地 LLM，禁止公网	物理隔离、专库专网、禁外连	每日

图表接近视口时加载…

查看 Mermaid 源码

flowchart LR
    IN[待入库数据] --> C1{是否公开可披露}
    C1 -->|是| L1[L1 公开数据]
    C1 -->|否| C2{是否含内部运营信息}
    C2 -->|是| C3{是否含 PII 或客户记录}
    C2 -->|否| L1
    C3 -->|否| L2[L2 内部数据]
    C3 -->|是| C4{是否含销售/利润/定价策略}
    C4 -->|否| L3[L3 敏感数据]
    C4 -->|是| L4[L4 核心经营]
    L1 --> A1[公网 API 可处理]
    L2 --> A2[本地 LLM 优先]
    L3 --> A3[脱敏后本地处理]
    L4 --> A4[物理隔离仅本地]

    classDef entry fill:#e3f2fd,stroke:#1d4ed8,stroke-width:1.5px,color:#1e3a8a;
    classDef success fill:#c8e6c9,stroke:#15803d,stroke-width:1.5px,color:#14532d;
    classDef fail fill:#ffcdd2,stroke:#dc2626,stroke-width:1.5px,color:#991b1b;
    classDef store fill:#e0f2f1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a;

    class IN,C1,C2,C3,C4 entry;
    class L1,L2,L3,L4 store;
    class A1,A2,A3,A4 success;

一个实用原则：级别按最高风险字段上浮，不按平均值下调。一份客服会话里只要出现手机号和订单地址，就不再是 L2，而是 L3；一份经营周报里只要出现毛利率和价格策略，就直接按 L4 管理。

5.2 本地 LLM 安全方案：默认走本地，例外才走公网

5.2.1 Ollama 安装与安全启动

本地推理的核心目标不是“省钱”，而是把高风险数据留在可控边界内。Ollama 最低要求是只监听 127.0.0.1，绝不直接暴露到 0.0.0.0。

bash

brew install ollama

launchctl setenv OLLAMA_HOST 127.0.0.1:11434
launchctl setenv OLLAMA_ORIGINS http://127.0.0.1,http://localhost

ollama serve

# 另开终端拉取模型
ollama pull qwen2.5:14b
curl http://127.0.0.1:11434/api/tags

如果是 Linux systemd，可用以下服务文件：

bash

[Unit]
Description=Ollama Local Service
After=network.target

[Service]
Environment="OLLAMA_HOST=127.0.0.1:11434"
Environment="OLLAMA_ORIGINS=http://127.0.0.1,http://localhost"
ExecStart=/usr/local/bin/ollama serve
Restart=always
User=ollama

[Install]
WantedBy=multi-user.target

5.2.2 数据敏感性检查中间件

下面的中间件把“能不能调用公网模型”收敛为显式规则：先看数据级别，再看是否命中 PII，再决定路由。

python

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from fastapi import FastAPI, HTTPException, Request
from typing import Any


class SensitivityLevel(str, Enum):
    L1 = "L1"
    L2 = "L2"
    L3 = "L3"
    L4 = "L4"


@dataclass
class RoutingDecision:
    allow_public_api: bool
    target_backend: str
    reason: str


def decide_llm_route(level: SensitivityLevel, pii_detected: bool) -> RoutingDecision:
    if level == SensitivityLevel.L4:
        return RoutingDecision(False, "local_ollama", "核心经营数据必须本地物理隔离")
    if level == SensitivityLevel.L3:
        if pii_detected:
            return RoutingDecision(False, "local_ollama", "敏感数据命中 PII，必须脱敏后本地处理")
        return RoutingDecision(False, "local_ollama", "敏感数据默认仅允许本地 LLM")
    if level == SensitivityLevel.L2:
        return RoutingDecision(False, "local_ollama", "内部数据默认本地优先，公网需专项审批")
    return RoutingDecision(True, "public_api", "公开数据可走公网 API")


app = FastAPI()


@app.middleware("http")
async def sensitivity_guard(request: Request, call_next):
    if request.url.path != "/ingest":
        return await call_next(request)

    payload: dict[str, Any] = await request.json()
    level = SensitivityLevel(payload["sensitivity_level"])
    pii_detected = bool(payload.get("pii_detected", False))
    decision = decide_llm_route(level, pii_detected)

    request.state.llm_backend = decision.target_backend
    request.state.routing_reason = decision.reason

    if not decision.allow_public_api and payload.get("requested_backend") == "public_api":
        raise HTTPException(status_code=403, detail=decision.reason)

    return await call_next(request)

5.2.3 一条数据能否走公网 API 的判定标准

只有同时满足下面三条，才可进入公网模型：

数据级别为 L1；
不含任何可识别自然人信息；
不包含内部经营指标、未发布策略、合同文本或客户上下文。

只要有一条不满足，就默认回退本地 LLM。 这比“业务方口头确认没问题”更可靠。

5.3 PII 检测与脱敏：未脱敏不入库

L3 以上数据进入知识库前，必须先过 PII 检测。推荐用 microsoft/presidio 做首轮识别，再根据场景选择替换或哈希。

5.3.1 替换策略 vs 哈希策略

策略	适用场景	优点	风险
替换策略	检索、总结、FAQ 生成	文本可读性高，便于 LLM 理解	无法跨文档追踪同一主体
哈希策略	行为分析、工单聚类、同人归并	可保留一致性标识	可读性差，需防止可逆推断

建议规则：面向回答生成用替换，面向统计关联用加盐哈希。不要把原文、替换文本、哈希文本混放在同一 collection。

5.3.2 完整脱敏示例：脱敏后才能入库

python

from __future__ import annotations

import hashlib
import json
import sqlite3
from datetime import datetime, timezone
from typing import Any

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig


analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()


def salted_hash(value: str, salt: str) -> str:
    digest = hashlib.sha256(f"{salt}:{value}".encode("utf-8")).hexdigest()
    return digest[:16]


def detect_and_mask(text: str, strategy: str = "replace", salt: str = "kb-salt") -> dict[str, Any]:
    results = analyzer.analyze(
        text=text,
        language="en",
        entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "LOCATION"],
    )

    pii_detected = len(results) > 0
    detected_entities = [item.entity_type for item in results]

    if strategy == "replace":
        operators = {
            "PERSON": OperatorConfig("replace", {"new_value": "<NAME>"}),
            "PHONE_NUMBER": OperatorConfig("replace", {"new_value": "<PHONE>"}),
            "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "<EMAIL>"}),
            "LOCATION": OperatorConfig("replace", {"new_value": "<ADDRESS>"}),
        }
        masked_text = anonymizer.anonymize(text=text, analyzer_results=results, operators=operators).text
    else:
        masked_text = text
        for item in sorted(results, key=lambda x: x.start, reverse=True):
            raw = text[item.start:item.end]
            token = f"<{item.entity_type}:{salted_hash(raw, salt)}>"
            masked_text = masked_text[:item.start] + token + masked_text[item.end:]

    return {
        "pii_detected": pii_detected,
        "detected_entities": detected_entities,
        "masked_text": masked_text,
    }


def write_audit_log(record: dict[str, Any]) -> None:
    conn = sqlite3.connect("audit.db")
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS ingestion_audit (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            data_source TEXT NOT NULL,
            sensitivity_level TEXT NOT NULL,
            pii_detected INTEGER NOT NULL,
            detected_entities TEXT NOT NULL,
            created_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "INSERT INTO ingestion_audit (data_source, sensitivity_level, pii_detected, detected_entities, created_at) VALUES (?, ?, ?, ?, ?)",
        (
            record["data_source"],
            record["sensitivity_level"],
            int(record["pii_detected"]),
            json.dumps(record["detected_entities"], ensure_ascii=False),
            datetime.now(timezone.utc).isoformat(),
        ),
    )
    conn.commit()
    conn.close()


def sanitize_before_index(text: str, data_source: str, sensitivity_level: str) -> str:
    result = detect_and_mask(text, strategy="replace")
    write_audit_log(
        {
            "data_source": data_source,
            "sensitivity_level": sensitivity_level,
            "pii_detected": result["pii_detected"],
            "detected_entities": result["detected_entities"],
        }
    )
    if sensitivity_level in {"L3", "L4"} and result["pii_detected"]:
        return result["masked_text"]
    return text


if __name__ == "__main__":
    sample = "Alice lives at 1 Infinite Loop, email alice@example.com, phone 13800138000."
    print(sanitize_before_index(sample, data_source="crm_export.csv", sensitivity_level="L3"))

5.3.3 审计日志 schema

json

{
  "event_type": "ingestion",
  "data_source": "crm_export.csv",
  "sensitivity_level": "L3",
  "pii_detected": true,
  "detected_entities": ["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER"],
  "masked_strategy": "replace",
  "operator": "system",
  "created_at": "2026-07-27T10:00:00Z"
}

5.4 RBAC 权限模型：知识库不是所有人都能查

权限控制的关键不是“有没有登录”，而是谁可以看哪个 collection，能做什么操作。最小可用模型建议四种角色：管理员、分析师、只读用户、API 客户端。

图表接近视口时加载…

查看 Mermaid 源码

flowchart LR
    U1[管理员] --> P1[读/写/删/管理]
    U2[分析师] --> P2[读/写]
    U3[只读用户] --> P3[只读]
    U4[API 客户端] --> P4[受限读写]

    P1 --> R1[L1 Collection]
    P1 --> R2[L2 Collection]
    P1 --> R3[L3 Collection]
    P1 --> R4[L4 Collection]
    P2 --> R1
    P2 --> R2
    P2 --> R3
    P3 --> R1
    P3 --> R2
    P4 --> R1
    P4 --> R2

    classDef entry fill:#e3f2fd,stroke:#1d4ed8,stroke-width:1.5px,color:#1e3a8a;
    classDef success fill:#c8e6c9,stroke:#15803d,stroke-width:1.5px,color:#14532d;
    classDef store fill:#e0f2f1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a;
    classDef fail fill:#ffcdd2,stroke:#dc2626,stroke-width:1.5px,color:#991b1b;

    class U1,U2,U3,U4 entry;
    class P1,P2,P3,P4 success;
    class R1,R2,R3 store;
    class R4 fail;

5.4.1 FastAPI RBAC 中间件示例

python

from __future__ import annotations

from enum import Enum
from fastapi import Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel


class Role(str, Enum):
    ADMIN = "admin"
    ANALYST = "analyst"
    READER = "reader"
    API_CLIENT = "api_client"


class Permission(BaseModel):
    collection: str
    action: str


ROLE_MATRIX: dict[Role, list[Permission]] = {
    Role.ADMIN: [
        Permission(collection="*", action="read"),
        Permission(collection="*", action="write"),
        Permission(collection="*", action="delete"),
        Permission(collection="*", action="manage"),
    ],
    Role.ANALYST: [
        Permission(collection="L1", action="read"),
        Permission(collection="L2", action="read"),
        Permission(collection="L3", action="read"),
        Permission(collection="L2", action="write"),
        Permission(collection="L3", action="write"),
    ],
    Role.READER: [
        Permission(collection="L1", action="read"),
        Permission(collection="L2", action="read"),
    ],
    Role.API_CLIENT: [
        Permission(collection="L1", action="read"),
        Permission(collection="L2", action="read"),
        Permission(collection="L2", action="write"),
    ],
}


def has_permission(role: Role, collection: str, action: str) -> bool:
    for permission in ROLE_MATRIX[role]:
        same_collection = permission.collection in {collection, "*"}
        same_action = permission.action == action
        if same_collection and same_action:
            return True
    return False


def get_current_role(x_role: str = Header(...)) -> Role:
    try:
        return Role(x_role)
    except ValueError as exc:
        raise HTTPException(status_code=401, detail="invalid role") from exc


def require_permission(collection: str, action: str):
    def checker(role: Role = Depends(get_current_role)) -> Role:
        if not has_permission(role, collection, action):
            raise HTTPException(status_code=403, detail=f"{role} cannot {action} on {collection}")
        return role
    return checker


app = FastAPI()


@app.get("/collections/{collection}")
def read_collection(collection: str, role: Role = Depends(require_permission("L2", "read"))):
    return {"collection": collection, "role": role, "status": "ok"}


@app.post("/collections/{collection}")
def write_collection(collection: str, role: Role = Depends(require_permission("L2", "write"))):
    return {"collection": collection, "role": role, "status": "written"}

生产环境里应再叠加两层：

认证层：JWT / OAuth2 / mTLS，禁止仅靠自定义 Header；
配额层：API 客户端按 token、IP、租户三维限流。

5.5 审计日志系统：查过什么、写过什么，都要留痕

知识库最常见的安全盲点不是“被黑”，而是出了问题后无法追溯。最低要求是两类日志：查询日志和写入日志。

5.5.1 SQLite 轻量实现

python

from __future__ import annotations

import json
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Iterable


DB_PATH = "kb_audit.sqlite3"


@contextmanager
def get_conn():
    conn = sqlite3.connect(DB_PATH)
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()


def init_db() -> None:
    with get_conn() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS query_audit (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                query TEXT NOT NULL,
                retrieved_ids TEXT NOT NULL,
                timestamp TEXT NOT NULL,
                latency_ms INTEGER NOT NULL
            )
            """
        )
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS write_audit (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                data_source TEXT NOT NULL,
                sensitivity_level TEXT NOT NULL,
                pii_detected INTEGER NOT NULL,
                timestamp TEXT NOT NULL
            )
            """
        )


def log_query(user_id: str, query: str, retrieved_ids: Iterable[str], latency_ms: int) -> None:
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO query_audit (user_id, query, retrieved_ids, timestamp, latency_ms) VALUES (?, ?, ?, ?, ?)",
            (
                user_id,
                query,
                json.dumps(list(retrieved_ids), ensure_ascii=False),
                datetime.now(timezone.utc).isoformat(),
                latency_ms,
            ),
        )


def log_write(data_source: str, sensitivity_level: str, pii_detected: bool) -> None:
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO write_audit (data_source, sensitivity_level, pii_detected, timestamp) VALUES (?, ?, ?, ?)",
            (
                data_source,
                sensitivity_level,
                int(pii_detected),
                datetime.now(timezone.utc).isoformat(),
            ),
        )


if __name__ == "__main__":
    init_db()
    log_query("u_001", "查 7 月退款原因", ["doc_12", "doc_18"], 143)
    log_write("crm_export.csv", "L3", True)

5.5.2 最低审计要求

每次查询记录：user / query / retrieved_ids / timestamp / latency；
每次写入记录：data_source / sensitivity_level / pii_detected；
日志默认不可被普通业务用户删除；
日志保留周期至少 180 天，L4 建议 365 天以上。

示例期限不是法定义务

本地控制 fixture 将 180/365 天、48 小时和 30 天固定标记为 illustrative；缺少辖区、数据类别、合同依据、批准引用和复核日期时，校验器会拒绝形成 production-baseline。这只能证明本地规则不会静默晋级，不是跨司法辖区的通用法定期限，也不是已经批准的生产基线。

5.6 供应商风险评估清单：先看边界，再看价格

公网模型是否可用，不应只看效果，要把数据留存、合规认证和适用数据级别一起比较。下表按 2026 年常见企业采购方式给出保守决策口径；具体条款以上线时签署的企业协议为准。

供应商	数据留存政策（ZDRA）	合规认证	适用数据级别	月度成本估算
Claude API	默认受平台策略约束；企业版可谈零数据留存协议，需合同确认	SOC 2、ISO 27001、GDPR 支持	L1；L2 仅限审批后脱敏场景	中高，视 token 量约数百到数千美元
OpenAI API	默认按平台策略处理；企业场景可申请零留存/不训练条款，需合同确认	SOC 2、ISO 27001、GDPR 支持	L1；L2 仅限审批后脱敏场景	中高，视模型与调用量约数百到数千美元
Azure OpenAI	走 Azure 企业合规边界，留存与地域控制通常更强，适合企业托管	SOC 2、ISO 27001、GDPR、区域合规能力较强	L1-L2；L3 需配合脱敏与专属网络	中高到高，外加 Azure 基础设施成本
本地 Ollama	无第三方留存，数据留在本地或私有网络	取决于自建环境与组织制度	L1-L4；L3/L4 推荐默认方案	低到中，主要为机器与运维成本

结论很简单：

L1 可以比较自由地选供应商。
L2 不看“能不能”，看“有没有审批与脱敏”。
L3/L4 不再是采购问题，而是架构问题，本地或私有化是默认前提。

5.7 上线前安全检查清单

下面这份清单可直接作为上线前自查项。

P0（阻塞上线）

☐ 已完成所有数据源的 L1-L4 分级，且有负责人签字。
☐ L3/L4 数据未走任何公网 LLM API。
☐ 所有入库文本均经过 PII 检测；命中记录有审计日志。
☐ 向量库与知识库 API 暴露公网时已启用认证。
☐ 所有公网接口已启用限流、超时与异常告警。
☐ Ollama 或本地推理服务仅绑定 127.0.0.1 或私网地址。
☐ L4 数据所在存储与应用部署在隔离网络，不与公网直连。

P1（核心回归）

☐ RBAC 已按 collection 和 action 生效，而不是只有登录态。
☐ 查询日志可追溯到用户、时间、检索文档 ID 和延迟。
☐ 写入日志可追溯到数据源、敏感级别、PII 检测结果。
☐ 脱敏策略已区分替换与哈希，不混用在同一 collection。
☐ 管理员、分析师、API 客户端权限已做最小化收敛。

P2（一般治理）

☐ 审计日志保留周期已配置并验证可导出。
☐ 厂商协议、ZDRA 与地域合规条款已归档。
☐ 已建立季度安全复审机制，至少复核数据分级、权限和供应商边界。

5.8 推断性隐私：被忽视的安全盲区

RBAC 控制"谁能查"，但无法控制"Agent 查完后能推断出什么"。这是传统权限模型的根本局限。

什么是推断性隐私（Inference Privacy）

当 Agent 把多条独立的 L1/L2 数据拼接后，可以推断出原本属于 L3/L4 的敏感信息——即便每条数据单独看都是合规的。

典型场景：

公开的员工姓名（L1）+ 公开的组织架构图（L1）+ 公开的薪酬区间（L2）→ Agent 可以推断出具体个人的薪酬范围（L3）
公开的产品价格（L1）+ 公开的市场份额数据（L1）+ 内部的毛利率区间（L3）→ 可反推出供应链成本（L4）

防护策略

策略一：查询意图过滤 在 Agent 生成 SQL/向量查询之前，加入意图分类器。当查询意图涉及"综合多类信息推断个体属性"时，触发人工审核。

策略二：输出范围约束 对 Agent 的最终回答做后处理扫描，检测输出中是否包含超出输入数据分级的推断结论。

策略三：数据混合限制 在知识库架构层面，明确禁止特定数据组合的同时检索。例如：个人信息类 collection 与财务数据 collection，不允许在同一个 Agent session 内跨 collection 联合查询。

python

FORBIDDEN_COMBINATIONS = [
    ("employee_personal", "salary_ranges"),
    ("product_pricing", "cost_structure"),
    ("user_behavior", "health_records"),
]

def check_query_scope(collections_accessed: list[str]) -> bool:
    for a, b in FORBIDDEN_COMBINATIONS:
        if a in collections_accessed and b in collections_accessed:
            raise PrivacyViolation(f"禁止同时访问 {a} 和 {b}")
    return True

5.9 数据毒化攻击：自动采集系统的隐藏风险

当知识库允许从公开网络自动采集内容时，存在被外部攻击者定向污染的风险。

攻击原理

攻击者在目标网页植入精心设计的内容，使 Agent 从该网页采集后，知识库中特定领域的 Skill 或知识条目被悄悄修改。典型目标：

篡改竞品分析中的参数对比（让你误判竞争对手）
注入错误的安全操作规程（让 Agent 指导用户做危险操作）
污染价格参考数据（影响定价决策）

防护清单

☐ 内容溯源哈希：每条入库内容保存原始 URL + 采集时间戳 + 内容哈希，任何后续更新必须与原始哈希对比

☐ 高权重来源白名单：只有白名单域名的内容可以自动入库；其他来源需人工审核后才能进入 L2 以上的知识层

☐ 异常变更检测：建立知识库内容的基线，对超过 20% 变化率的批量更新触发告警

☐ 隔离沙箱测试：新采集的内容先在隔离环境运行 48 小时，通过 Golden Set 验证后才合并到生产库

自动化程度越高，受攻击面越大

完全自动化的知识库采集 pipeline，一旦被定向攻击，可以在几天内系统性地改变 Agent 的行为，且极难被发现。建议：完全自动化仅适用于高度可信的封闭数据源（内部系统、白名单 API）；公开网络采集必须保留人工抽检环节。

本章结论

知识库建设真正的第一步，不是选 Qdrant、Neo4j、LightRAG 还是 GraphRAG，而是先回答四个问题：数据是什么级别、是否含 PII、谁能访问、出了事能否追溯。 只要这四件事没定清，任何"先做 MVP 再补安全"的方案，最后都会变成返工。

工程上最稳妥的默认值只有一句话：数据先分级，敏感默认本地，入库必须脱敏，访问全程留痕。

→ 下一章

安全到位后，按需深入GraphRAG图谱构建 → 05-graphrag

5.10 推断攻击演练 SOP

演练目标：在上线前模拟攻击者视角，验证系统是否会通过合法查询推断出超权限信息。

演练步骤

步骤 1：构造组合查询序列 设计三轮递进查询，每轮单独合法，组合后可推断敏感信息：

python

INFERENCE_ATTACK_SCENARIOS = [
    {
        "name": "员工薪酬推断",
        "queries": [
            "公司各职级的薪酬区间是什么？",          # L2 内部数据
            "张三目前是什么职级？",                   # L2 内部数据
            "张三的薪酬大概是多少？",                  # 组合后变L3
        ],
        "expected_block_at": 3,   # 第3轮应被拦截
        "risk_if_leaked": "违反劳动法保密义务"
    },
    {
        "name": "供应链成本推断",
        "queries": [
            "产品A的建议零售价是多少？",              # L1 公开
            "产品A的毛利率区间是多少？",              # L3 内部
            "产品A的供应商采购成本大概是多少？",       # 推断出L4
        ],
        "expected_block_at": 2,
        "risk_if_leaked": "商业机密泄露"
    }
]

步骤 2：记录实际拦截情况

python

def run_inference_drill(kb_client, scenarios: list[dict]) -> dict:
    results = []
    for scenario in scenarios:
        drill_result = {"name": scenario["name"], "steps": []}
        session_context = []

        for i, query in enumerate(scenario["queries"]):
            response = kb_client.query(query, context=session_context)
            blocked = response.get("privacy_blocked", False)
            drill_result["steps"].append({
                "step": i + 1,
                "query": query,
                "blocked": blocked,
                "should_block": i + 1 >= scenario["expected_block_at"]
            })
            session_context.append({"query": query, "response": response["answer"]})
            if blocked:
                break

        drill_result["passed"] = all(
            s["blocked"] == s["should_block"]
            for s in drill_result["steps"]
        )
        results.append(drill_result)

    return {
        "total": len(results),
        "passed": sum(1 for r in results if r["passed"]),
        "details": results
    }

步骤 3：修复未拦截的场景 对每个未通过的推断路径，在 FORBIDDEN_COMBINATIONS 规则里新增对应的 collection 组合限制（见 §5.8）。

演练必须使用已授权的代表性数据

本地安全演练 fixture 会拒绝任何标记为 production 的样本，并要求隔离环境、受支持的非生产数据分类、保留时长和销毁回执模式。Fixture 通过只证明失败关闭逻辑；真实演练仍须优先使用合成数据、脱敏样本或经数据负责人批准的最小化快照，并记录授权范围、保留期限与销毁回执。

5.11 数据毒化隔离与回滚 Runbook

当检测到知识库内容可能已被外部定向污染时，按以下流程处置。

毒化检测信号

某个来源域名下的内容在短期内集中出现与历史风格差异显著的新内容
特定主题的 Agent 输出开始系统性偏向某个方向（用漂移检测发现）
安全团队或用户报告"AI建议了不合理操作"且可追溯到特定知识源

隔离步骤（4小时内完成）

bash

# Step 1: 识别可疑来源
python3 - << 'EOF'
import chromadb
from collections import Counter

client = chromadb.PersistentClient("./chroma_db")
col = client.get_collection("product_kb_v2")
all_data = col.get(include=["metadatas"])

# 找出近7天新增且来自可疑域名的内容
suspicious = [
    (m["id"] if "id" in m else all_data["ids"][i], m["source_url"])
    for i, m in enumerate(all_data["metadatas"])
    if "suspicious-domain.com" in m.get("source_url", "")
]
print(f"可疑条目：{len(suspicious)}")
for item in suspicious[:10]:
    print(item)
EOF

# Step 2: 立即隔离（移出活跃索引，不删除）
python3 - << 'EOF'
# 把可疑条目的 metadata 标记为 quarantined=True
# ChromaDB 不支持直接隐藏，改为降低其检索权重
# 实际操作：迁移到隔离 collection
client.create_collection("quarantine_kb")
# ... 移动逻辑
print("隔离完成，可疑内容已移出活跃索引")
EOF

回滚步骤（若确认污染）

bash

# Step 3: 从快照恢复（需要定期备份）
SNAPSHOT_DATE="2026-07-01"
cp -r "./chroma_db_backup_${SNAPSHOT_DATE}" "./chroma_db_restored"

# Step 4: 增量补录快照日期之后的合法内容
python3 main.py ingest \
  --source "./verified_sources/" \
  --since "${SNAPSHOT_DATE}" \
  --skip-domains "suspicious-domain.com"

# Step 5: 重新运行黄金问题集验收
python3 main.py eval --golden-set golden_questions.json

# Step 6: 记录事故复盘
echo "毒化事故记录到 incidents/$(date +%Y%m%d)-poisoning.md"

备份是前提

没有定期快照，就没有可回滚的版本。建议：每日增量备份 + 每周全量快照，至少保留 30 天。这是整个回滚 Runbook 的基础，没有它，其他步骤都是空话。

验收契约

ACC-SECURITY-001 已锁定 5 个本地合成用例，其中 3 个是失败关闭负例。当前复放为 5/5，但数据集不是获授权业务样本、阈值仍是本地示意门槛、四类角色均未具名接受，也没有最终验收回执，因此本章继续保持 solution / pending。

Acceptance gate · schema v1.0

本地复放通过，不等于最终接受

本地可复放: 1/1
已验收: 0/1

本注册表只裁决知识页面能否升级为本仓库的可验收内容，不证明生产部署、法律合规、预算批准或真实模型效果。当前复核日期 2026-08-02。

05 · 数据安全与合规架构 —— 构建知识库前必读ACC-SECURITY-001审批受阻

固定集合5 例3 个负例

本地复放100%L2 · 零外部调用

基线差异0.0 pp未观察到退化

责任接受0/4最终回执缺失

固定输入与回归身份

数据集: ADS-SECURITY-LOCAL-V1 · v1.0.0
数据边界: 合成 fixture，未获业务授权
基线回执: ACR-SECURITY-LOCAL-V1
验收范围: 仅仓库内容 · productionReady=false

阈值复放与批准状态

用例通过率 实测 1 · ≥ 1 本地通过 · 示意阈值
外部调用 实测 0 · = 0 本地通过 · 示意阈值
外部副作用 实测 0 · = 0 本地通过 · 示意阈值

当前阻断

DATASET_NOT_BUSINESS_AUTHORIZED数据集尚非获授权业务样本
THRESHOLDS_NOT_APPROVED阈值仍是示意值，未获批准
OWNER_ACCEPTANCE_INCOMPLETE四类具名责任尚未全部接受
FINAL_RECEIPT_MISSING最终仓库内容验收回执缺失

下一证据绑定获授权业务样本与辖区控制表，由四类具名责任人接受职责并批准阈值，再签发仓库内容验收回执。

查看 05 章验收上下文

关键断言与证据

4 条试点断言 · 页面状态与断言证据分开计算

L0/L1/L2 只说明登记范围；不自动代表法律合规、生产可用或整章验收。

CLM-SEC-001来源复核

截至 2026-08-01，NIST 将 AI RMF 描述为自愿使用的框架，并提示 AI RMF 1.0 正在修订。

证据等级: L1 · 一手公开 / 只读观察
类型 / 风险: 外部事实 · 高
截至日期: 2026-08-01
责任链: 4/4 角色已映射 · 0/4 已接受

查看适用范围、限制与证据

适用范围: 用于工程风险管理参考，不是法律合规结论。
限制: 一手网页复核只能证明公开表述；不能证明组织已实施框架，也不能替代适用法律。
下一动作: 由安全负责人选定采用版本、记录控制映射，并保存审批回执。

责任链（角色映射）

内容维护安全合规内容负责人角色已映射 · 待具名认领
证据复核安全、法务与隐私证据复核人角色已映射 · 待具名认领
测试维护安全控制测试负责人角色已映射 · 待具名认领
最终批准安全与合规最终批准人角色已映射 · 待具名认领

角色映射只解决职责归属；具名责任人接受并留下回执前，不能据此宣称已审批。

来源

NIST AI Risk Management Framework复核于 2026-08-01

CLM-SEC-002来源复核

EU AI Act 的适用范围可能覆盖欧盟境内外向欧盟市场投放或在欧盟使用 AI 系统的主体，具体义务取决于角色和预期用途。

证据等级: L1 · 一手公开 / 只读观察
类型 / 风险: 法律语境 · 关键
截至日期: 2026-08-01
责任链: 4/4 角色已映射 · 0/4 已接受

查看适用范围、限制与证据

适用范围: 仅用于触发法务与隐私审查，不用于自动作出辖区或风险分类结论。
限制: 官方 FAQ 不是针对本项目的法律意见；角色、豁免、行业规则和实施时间仍需逐案确认。
下一动作: 由法务或隐私负责人签署角色、辖区、预期用途和风险分类记录。

责任链（角色映射）

内容维护安全合规内容负责人角色已映射 · 待具名认领
证据复核安全、法务与隐私证据复核人角色已映射 · 待具名认领
测试维护安全控制测试负责人角色已映射 · 待具名认领
最终批准安全与合规最终批准人角色已映射 · 待具名认领

角色映射只解决职责归属；具名责任人接受并留下回执前，不能据此宣称已审批。

来源

European Commission - Navigating the AI Act复核于 2026-08-01

CLM-SEC-003Fixture 验证

仓库内安全演练 fixture 会拒绝标记为 production 的样本，并要求使用隔离环境、受支持的非生产数据分类和销毁回执模式。

证据等级: L2 · Fixture / Dry-run
类型 / 风险: 控制要求 · 关键
截至日期: 2026-08-01
责任链: 4/4 角色已映射 · 0/4 已接受

查看适用范围、限制与证据

适用范围: 适用于本指南建议的安全演练流程。
限制: 该 fixture 只验证本地失败关闭逻辑；没有接触真实数据，也没有产生已批准威胁模型、真实演练或销毁回执。
下一动作: 由具名安全与数据责任人认领角色后，在隔离环境使用获授权样本执行一次真实演练并留存回执。

责任链（角色映射）

内容维护安全合规内容负责人角色已映射 · 待具名认领
证据复核安全、法务与隐私证据复核人角色已映射 · 待具名认领
测试维护安全控制测试负责人角色已映射 · 待具名认领
最终批准安全与合规最终批准人角色已映射 · 待具名认领

角色映射只解决职责归属；具名责任人接受并留下回执前，不能据此宣称已审批。

本地证据

Fail-closed security governance fixture · fixtures/security-governance.mjsL2 · Fixture / Dry-run
Security governance Node tests · tests/content/security-governance.test.mjsL2 · Fixture / Dry-run

CLM-SEC-004Fixture 验证

仓库内安全治理 fixture 将 180/365 天、48 小时和 30 天标记为 illustrative，并在缺少辖区、数据类别、合同依据、批准引用和复核日期时拒绝形成 production-baseline。

证据等级: L2 · Fixture / Dry-run
类型 / 风险: 控制要求 · 关键
截至日期: 2026-08-01
责任链: 4/4 角色已映射 · 0/4 已接受

查看适用范围、限制与证据

适用范围: 仅可作为设计讨论的起始参数。
限制: Fixture 只能证明示例值不会被本地规则静默提升；它没有绑定真实辖区、合同义务、恢复目标或责任人批准。
下一动作: 建立真实辖区与数据类别控制表，由具名法务、安全和业务责任人批准每个期限并保存接受回执。

责任链（角色映射）

内容维护安全合规内容负责人角色已映射 · 待具名认领
证据复核安全、法务与隐私证据复核人角色已映射 · 待具名认领
测试维护安全控制测试负责人角色已映射 · 待具名认领
最终批准安全与合规最终批准人角色已映射 · 待具名认领

角色映射只解决职责归属；具名责任人接受并留下回执前，不能据此宣称已审批。

本地证据

Fail-closed security governance fixture · fixtures/security-governance.mjsL2 · Fixture / Dry-run
Security governance Node tests · tests/content/security-governance.test.mjsL2 · Fixture / Dry-run

来源与复核

一手基准（截至 2026-08-01）：NIST AI Risk Management Framework（自愿框架，1.0 正在修订）与 European Commission: Navigating the AI Act（适用范围、风险分类与高风险义务）。
复核状态：待复核。任何易漂移的版本、价格、法律或性能结论，采用前都必须回到一手来源再次确认。
代码状态：混合边界。fixtures/security-governance.mjs 及其测试为 L2 本地 fixture；本章其他代码仍是示意代码。
证据边界：本页成熟度只描述内容形态，不代表部署、上线或生产验收已经完成。
下一验收动作：由具名角色接受责任后，绑定真实辖区与授权样本，在隔离环境完成演练和销毁回执。

第五章：数据安全与合规架构 —— 构建知识库前必读 ​

5.1 数据分级体系：先分级，再决定模型和存储 ​

5.2 本地 LLM 安全方案：默认走本地，例外才走公网 ​

5.2.1 Ollama 安装与安全启动 ​

5.2.2 数据敏感性检查中间件 ​

5.2.3 一条数据能否走公网 API 的判定标准 ​

5.3 PII 检测与脱敏：未脱敏不入库 ​

5.3.1 替换策略 vs 哈希策略 ​

5.3.2 完整脱敏示例：脱敏后才能入库 ​

5.3.3 审计日志 schema ​

5.4 RBAC 权限模型：知识库不是所有人都能查 ​

5.4.1 FastAPI RBAC 中间件示例 ​

5.5 审计日志系统：查过什么、写过什么，都要留痕 ​

5.5.1 SQLite 轻量实现 ​

5.5.2 最低审计要求 ​

5.6 供应商风险评估清单：先看边界，再看价格 ​

5.7 上线前安全检查清单 ​

P0（阻塞上线） ​

P1（核心回归） ​

P2（一般治理） ​

5.8 推断性隐私：被忽视的安全盲区 ​

什么是推断性隐私（Inference Privacy） ​

防护策略 ​

5.9 数据毒化攻击：自动采集系统的隐藏风险 ​

攻击原理 ​

防护清单 ​

本章结论 ​

5.10 推断攻击演练 SOP ​

演练步骤 ​

5.11 数据毒化隔离与回滚 Runbook ​

毒化检测信号 ​

隔离步骤（4小时内完成） ​

回滚步骤（若确认污染） ​

验收契约 ​

关键断言与证据 ​

来源与复核 ​

第五章：数据安全与合规架构 —— 构建知识库前必读

5.1 数据分级体系：先分级，再决定模型和存储

5.2 本地 LLM 安全方案：默认走本地，例外才走公网

5.2.1 Ollama 安装与安全启动

5.2.2 数据敏感性检查中间件

5.2.3 一条数据能否走公网 API 的判定标准

5.3 PII 检测与脱敏：未脱敏不入库

5.3.1 替换策略 vs 哈希策略

5.3.2 完整脱敏示例：脱敏后才能入库

5.3.3 审计日志 schema

5.4 RBAC 权限模型：知识库不是所有人都能查

5.4.1 FastAPI RBAC 中间件示例

5.5 审计日志系统：查过什么、写过什么，都要留痕

5.5.1 SQLite 轻量实现

5.5.2 最低审计要求

5.6 供应商风险评估清单：先看边界，再看价格

5.7 上线前安全检查清单

P0（阻塞上线）

P1（核心回归）

P2（一般治理）

5.8 推断性隐私：被忽视的安全盲区

什么是推断性隐私（Inference Privacy）

防护策略

5.9 数据毒化攻击：自动采集系统的隐藏风险

攻击原理

防护清单

本章结论

5.10 推断攻击演练 SOP

演练步骤

5.11 数据毒化隔离与回滚 Runbook

毒化检测信号

隔离步骤（4小时内完成）

回滚步骤（若确认污染）

验收契约

关键断言与证据

来源与复核