ai-audit-trail

ai-audit-trail

证明 AI 做了什么、为什么这么做，以及记录未被篡改。

具备防篡改能力的决策凭证，采用 Ed25519 签名、SHA-256 哈希链， 并提供正式的合规映射。无需区块链、无需 SaaS、无需受制于供应商。 可自托管、可离线验证、原生支持 Python。

---

为什么需要它

**《欧盟 AI 法案》**将于 2026 年 8 月起对高风险 AI 系统强制实施。该法案要求提供防篡改日志，证明每项决策均被正确作出（第 12 条）。大多数团队使用常规日志来满足这一要求，但常规日志既不能防篡改，也无法在审计中作为有法律效力的证据。

ai-audit-trail 通过加密凭证弥补这一缺口，任何审计人员都能离线验证凭证，无需访问你的系统。其原理与区块链相同，但没有区块链的额外开销、SaaS 依赖或供应商锁定。

适用对象

• 受监管的 AI 团队（金融科技、健康科技、法律科技、保险科技），需要证明合规
• 企业平台团队，部署可访问工具的大模型 Agent
• 安全与合规负责人，需要可直接用于审计的证据包
• 开发者，希望使用 pip install 和 3 行代码，而非迁移平台

此库提供的内容

ai-audit-trail 提供支持《欧盟 AI 法案》、ISO 42001 和 NIST AI RMF 合规所需的技术构建模块。它本身并不能保证符合法规要求——合规是组织层面的义务，涉及的范围超出任何单一软件组件。请参阅下方的共同责任模型。

---

安装

pip install ai-audit-trail                    # Core (Ed25519 + SHA-256 + PII)
pip install "ai-audit-trail[redis]"           # + Redis persistence
pip install "ai-audit-trail[otel]"            # + OpenTelemetry metrics
pip install "ai-audit-trail[all]"             # Everything

要求： Python 3.11+ | 无需外部服务 | 可在隔离网络环境中运行

---

快速入门

from ai_audit import (
    AuditConfig, init_audit_config,
    ReceiptCollector, ReceiptStore, ReceiptAction,
    verify_chain, get_verify_key_hex,
)

# 1. Configure once at startup
init_audit_config(AuditConfig(is_production=False))
store = ReceiptStore()

# 2. Wrap every AI request
collector = ReceiptCollector(trace_id="req-1", tenant_id="acme")
collector.set_input("What is our GDPR policy?")
collector.add_check("safety", score=0.02, threshold=0.8, fired=False)
collector.set_output("Our GDPR policy states that...")
collector.set_action(ReceiptAction.ALLOW)
collector.emit(store)
collector.cleanup()

# 3. Verify tamper-evidence
result = verify_chain(store.get_by_tenant("acme"), get_verify_key_hex())
assert result.valid  # Ed25519 + SHA-256 + hash-chain verified

---

架构概览

Receipt Creation        Verification & Compliance       Agentic AI Audit
─────────────────       ──────────────────────────      ──────────────────
ReceiptCollector   ──>  verify_chain()                  ToolCallReceipt
  set_input()           build_compliance_summary()      TraceGraph (DAG)
  add_check()           build_crosswalk()               BehavioralContract
  set_output()          export_evidence_package()       ProvenanceChain
  set_action()          SPRTMonitor
  emit()                DriftMonitor
       │                EpochManager
       v
  ReceiptStore     ──>  StorageBackend ABC
  (in-memory LRU)       InMemoryBackend
  + Redis (optional)    (your custom backends)
  + AuditBuffer

---

核心功能

决策凭证（Ed25519 + SHA-256 + 哈希链）

AI 流水线中的每项决策都会生成一份决策凭证——以加密方式封装、通过哈希链串联的记录：

可证明内容	方式
输入完整性	对经过 NFKC 规范化并剥离 PII 的输入计算 SHA-256
输出完整性	对生成的输出计算 SHA-256
检查结果	包含评分和阈值的有序检查记录
决策	所采取的操作（ALLOW / REJECT / ESCALATE / ...）
模型来源	模型 ID + 配置摘要
不可否认性	Ed25519 签名（libsodium）
顺序	哈希链链接（prev_receipt_hash）

三级验证（每份凭证耗时 < 0.1 ms）：

Ed25519 signature  →  detects forgery
SHA-256 self-hash  →  detects corruption
Hash-chain link    →  detects insertions / deletions / reordering

PII 脱敏（GDPR 第 17 条）

在哈希处理前剥离个人数据——审计日志中永远不会包含原始 PII。

from ai_audit import PiiConfig, PiiMode, PiiType

config = PiiConfig(
    enabled_types=frozenset({PiiType.EMAIL, PiiType.PHONE, PiiType.IP}),
    mode=PiiMode.REDACT,  # or HASH (SHA-256) or MASK (a***m)
)
collector = ReceiptCollector(tenant_id="acme", pii_config=config)

模式	[email protected] 转换为
REDACT	[EMAIL]
HASH	3d4e5f8a...（确定性 SHA-256）
MASK	a***@c***.com

加密销毁（GDPR 删除权）

使用每个租户专属的 AES-256-GCM 密钥加密 PII 字段。销毁密钥 = 数据将永久无法读取，同时保持哈希链完整。

from ai_audit.shredding import AESGCMDEKStore, encrypt_field, shred_tenant

dek_store = AESGCMDEKStore()
dek_store.create_dek("tenant-acme")

field = encrypt_field("sensitive PII", dek_store, "tenant-acme")
shred_tenant("tenant-acme", dek_store)  # Key destroyed — data unrecoverable
# Hash-chain remains mathematically intact (hashes ciphertext, not plaintext)

---

合规与治理

ISO 42001 / NIST AI RMF 对照表

将收据数据直接映射到公认的管理控制，并提供证据指针。

from ai_audit.crosswalk import build_crosswalk, nist_function_map

crosswalk = build_crosswalk(receipts, chain_intact=True)
for entry in crosswalk:
    print(f"[{entry.status}] {entry.framework} {entry.control_id} — {entry.control_name}")

nist = nist_function_map(receipts)
print(nist["GOVERN"].coverage)   # 0.0–1.0
print(nist["MEASURE"].status)    # PASS / PARTIAL / FAIL

ISO 42001 控制项： A.6.2.8（日志记录）、A.7.5（溯源）、A.6.2.6（性能）、A.8.4（输出）、A.5.3（风险） NIST AI RMF： GOVERN、MAP、MEASURE、MANAGE——附带量化覆盖率评分

EU AI Act 合规报告

from ai_audit.report import ComplianceReportGenerator

gen = ComplianceReportGenerator(summary, verify_key_hex=get_verify_key_hex())
gen.to_markdown()   # Documentation portals
gen.to_json()       # Automated pipelines
gen.to_html()       # Air-gapped servers

涵盖第 9 条（风险）、第 12 条（记录保存）、第 13 条（透明度）、第 17 条（质量）、第 18 条（日志记录）。

证据包导出（离线验证）

提供供外部审计员使用的自包含签名 ZIP 文件——无需访问系统。

from ai_audit.export import export_evidence_package, verify_evidence_package

export_evidence_package(receipts, verify_key_hex, signing_key, "audit_2026.zip")
# Bundle: receipts.jsonl + chain_metadata.json + public_key.hex + manifest.json (signed) + verify.py

# Auditor verifies offline:
# python -m ai_audit verify audit_2026.zip

持续认证（SPRT）

序贯概率比检验——实时查看每个租户的合规状态。

from ai_audit.sprt import SPRTMonitor

monitor = SPRTMonitor(tenant_id="acme")
for receipt in receipts:
    state = monitor.update(is_reject=(receipt.action == "reject"))
print(state.status)      # CERTIFIED | MONITORING | FLAGGED
print(state.confidence)  # 0.0–1.0

漂移检测（Jensen-Shannon 散度）

实时检测行为变化——纯 Python 实现，无需 scipy。

from ai_audit.drift import DriftMonitor

monitor = DriftMonitor(window_size=100)
for receipt in receipts:
    state = monitor.update(receipt.action.value)
print(state.status)       # STABLE | DRIFTING | CRITICAL_DRIFT
print(state.drift_score)  # JSD 0.0–1.0

---

Agentic AI 审计（世界首创）

Agent 行为契约

形式化的（p、delta、k）满意度与可靠性指数 Theta——将 EU AI Act 要求转化为可测试的规范。

from ai_audit.contracts import BehavioralContract, Constraint, ContractMonitor

contract = BehavioralContract(
    contract_id="safety-v1",
    constraints=[
        Constraint(name="no_pii_leak", kind="hard", field="action", operator="!=", value="reject"),
        Constraint(name="quality", kind="soft", field="checks.safety.score", operator="<=", value=0.1, delta=0.5),
    ],
)
monitor = ContractMonitor(contract)
for receipt in receipts:
    state = monitor.evaluate(receipt)

print(state.p)                   # Hard constraint satisfaction probability
print(state.delta)               # Maximum soft deviation observed
print(state.k)                   # Recovery steps after last violation
print(state.reliability_index)   # Theta: single compliance metric (0.0–1.0)
print(state.status)              # COMPLIANT | DEGRADED | VIOLATED

加密工具调用收据

每次 Agent API 调用均使用 Ed25519 签名——目前没有其他框架提供此功能。

from ai_audit.toolcall import seal_tool_call, verify_tool_call_chain

receipt = seal_tool_call(
    agent_id="researcher",
    tool_name="web_search",
    tool_args={"query": "EU AI Act compliance"},
    tool_result="Found 5 relevant documents...",
    private_key=signing_key,
    tenant_id="acme",
)
assert receipt.verify(signing_key.verify_key)

多 Agent 跟踪图（DAG）

审计委派、交接和并行编排——不只是线性日志。

from ai_audit.tracegraph import TraceGraph

graph = TraceGraph(trace_id="workflow-1", tenant_id="acme")
root = graph.add_node(agent_id="orchestrator", action="plan")
graph.add_node(agent_id="researcher", action="search", parent_id=root.node_id)
graph.add_node(agent_id="writer", action="draft", parent_id=root.node_id)

assert graph.verify_integrity()  # Hash-based tamper detection
assert not graph.has_cycles()    # DAG validation
lineage = graph.get_agent_lineage(leaf_node.node_id)  # Root-to-leaf trace

认知完整性 / 不可伪造的溯源

跟踪每条信息的来源——证明决策未受提示词注入影响。

from ai_audit.provenance import ProvenanceChain, ProvenanceRecord, SourceType

chain = ProvenanceChain(receipt_id="r1", tenant_id="acme")
chain.add(ProvenanceRecord(source_type=SourceType.SYSTEM, source_id="prompt", trust_level=1.0, content_hash="..."))
chain.add(ProvenanceRecord(source_type=SourceType.DOCUMENT, source_id="doc-123", trust_level=0.8, content_hash="..."))
chain.add(ProvenanceRecord(source_type=SourceType.UNKNOWN, source_id="???", trust_level=0.0, content_hash="..."))

summary = chain.trust_summary()
print(summary.system_grounded)       # True — has SYSTEM source
print(summary.potentially_injected)  # True — has UNKNOWN source
print(summary.min_trust)             # 0.0 — weakest link

---

高吞吐架构

Merkle 树批次封存（RFC 6962）

使用根哈希链替代收据链——每批次的验证复杂度为 O(log N)。

from ai_audit.batch import MerkleBatcher

batcher = MerkleBatcher(tenant_id="acme", private_key=key, max_batch_size=2048)
for receipt in receipts:
    seal = batcher.add(receipt.receipt_id, receipt.seal_payload())
    if seal:  # Auto-flushed at 2048 receipts
        print(f"Batch sealed: {seal.merkle_root[:16]}...")

assert batcher.verify_chain_of_roots(key.verify_key)

链纪元 / 轮换

防止链无限增长。旧纪元可以归档或删除。

from ai_audit.epochs import EpochManager

mgr = EpochManager(tenant_id="acme", private_key=key, max_epoch_size=10_000)
for receipt in receipts:
    seal = mgr.add_receipt(receipt)  # Auto-seals at 10k
mgr.seal_epoch()                     # Or explicit rollover
assert mgr.verify_epoch_chain(key.verify_key)

带背压的环形缓冲区

面向高吞吐量数据摄取的有界缓冲区——失败时关闭，不会静默丢失数据。

from ai_audit.buffer import AuditBuffer, AuditBufferFullError

buffer = AuditBuffer(maxsize=50_000)  # ~5 seconds at 10k req/s
try:
    buffer.put(receipt)
except AuditBufferFullError:
    # Backpressure — reject the request rather than lose audit data
    pass
batch = buffer.drain(max_items=2048)

存储后端抽象基类

可插拔持久化——可使用自选数据库。

from ai_audit.storage import StorageBackend, InMemoryBackend

# Use the reference implementation for dev/test
backend = InMemoryBackend(max_receipts=50_000)

# Or implement your own:
class PostgresBackend(StorageBackend):
    def write_receipt(self, receipt): ...
    def read_receipt(self, receipt_id): ...
    def query_by_tenant(self, tenant_id, limit=100): ...
    def healthcheck(self) -> bool: ...

OpenTelemetry 监测

为 SRE 仪表板提供原生指标——未安装 OTel SDK 时可平滑退化为空操作。

# pip install "ai-audit-trail[otel]"
from ai_audit.telemetry import record_seal, record_append, record_drift

record_seal(duration_seconds=0.000045, tenant_id="acme")
record_append(tenant_id="acme", async_mode=True)
record_drift(score=0.03, tenant_id="acme")

指标： ai_audit.seal_duration_seconds、ai_audit.append_total、ai_audit.redis_fallback_total、ai_audit.chain_break_total、ai_audit.drift_score、ai_audit.buffer_size、ai_audit.epoch_sealed_total

---

生产环境配置

持久化签名密钥

python -c "import nacl.signing; print(nacl.signing.SigningKey.generate().encode().hex())"

init_audit_config(AuditConfig(is_production=True, signing_key_hex="your-64-char-hex-key"))

KMS 集成

from ai_audit import KeyProvider, init_key_provider

class VaultKeyProvider(KeyProvider):
    def get_signing_key(self) -> nacl.signing.SigningKey:
        secret = vault_client.secrets.kv.read_secret("secret/ai-audit/key")
        return nacl.signing.SigningKey(bytes.fromhex(secret["data"]["key"]))
    def get_verify_key_hex(self) -> str:
        return self.get_signing_key().verify_key.encode().hex()

init_key_provider(VaultKeyProvider())

Redis 持久化

import redis
store = ReceiptStore(redis_client=redis.Redis(), use_lua=True)  # Lua mode: 10k+ req/s

---

共同责任模型

责任	库	部署方
Ed25519 + SHA-256 签名与哈希	X
哈希链完整性	X
PII 脱敏（REDACT/HASH/MASK）	X
Merkle 树批次封存（RFC 6962）	X
SPRT 合规认证	X
ISO 42001 / NIST AI RMF 映射	X
证据包导出与验证	X
加密销毁（AES-256-GCM）	X
Agent 行为契约	X
OpenTelemetry 指标	X
安全密钥存储（HSM/Vault）		X
PII 类型配置		X
持久化存储后端		X
访问控制 / RBAC		X
人工监督（EU AI Act 第 14 条）		X
时钟同步（NTP）		X
事件响应		X
监管合规认证		X

---

示例

请参阅 examples/ 目录：

• end_to_end_audit.py — 完整生命周期：收据、验证、对照表、证据导出
• fastapi_middleware.py — FastAPI 审计中间件模式
• langchain_callback.py — LangChain 回调处理器

---

性能

操作	典型延迟	备注
seal() (哈希 + 签名)	< 100 us	通过 libsodium C 实现 Ed25519
verify_chain(1000)	< 50 ms	线性扩展
merkle_root(2048)	< 5 ms	RFC 6962 SHA-256
每份收据的内存占用	~1 KB	Pydantic V2 + orjson

运行基准测试：pytest tests/test_benchmark.py -v -s

---

项目统计

指标	数值
测试	196
源码模块	26
__all__ 导出项	60
类型检查	mypy --strict，0 个错误
代码检查	ruff，0 个错误
Python 版本	3.11、3.12、3.13
安全扫描	已完成 3 次，应用 6 项修复
咨询的 NB 验证人员	5 位（架构、企业、性能、Agentic、品牌）

---

安全

完整的威胁模型、漏洞报告流程和受支持版本，请参阅 SECURITY.md。

架构不变量、共同责任模型和贡献指南，请参阅 CONTRIBUTING.md。

---

许可证

MIT — 可免费用于商业用途。

---

创建并维护者

S&S Connect — 为自主 AI 系统构建信任基础设施。

• 维护者： Fabrice（@sundsoffice-tech）
• 代码仓库： github.com/sundsoffice-tech/ai-audit-trail
• PyPI： pypi.org/project/ai-audit-trail
• 问题与反馈： GitHub Issues

如果你在生产环境或研究中使用 ai-audit-trail，我们很乐意听取你的反馈。