企业级 Claude API 治理工具包：PII 扫描 + 审计 + 成本控制

企业级 Claude API 治理工具包：PII 扫描 + 审计 + 成本控制 | 前端进阶之旅

Full translation

完整中文译文

Anthropic 的 Claude API 很优秀，SDK 干净、原生异步、文档完善。半天就能从零到一个可用的 AI 功能。

然后你们合规团队的人走进来了。

"我们是不是在向第三方 API 发送个人身份信息（PII）？""审计日志在哪里？""每天这块成本是多少？""怎么控制谁先访问？"

SDK 回答不了这些问题。它也不该回答——这不是它的工作。

所以我写了 enterprise-claude-kit：一个治理层，夹在应用和 Claude 之间。下面是它的功能以及我的设计思路。

pip install enterprise-claude-kit

一行安装搞定。没有外部服务、没有 Redis、没有边车容器。SQLite 做持久化，纯异步 Python，零基础设施。

一张图说明问题

每次调用都会经历五个阶段：

Your App
  → ① GovernanceLayer (pre-flight) — PII scan, blocked keywords, length guard
  → ② Claude API call
  → ③ GovernanceLayer (post-response) — PII in output, GxP citation check
  → ④ TokenMonitor — cost calculation, budget enforcement, alerts
  → ⑤ AuditLogger — append-only SHA-256 tamper-evident record
  → RunResult { content, cost_usd, tokens, governance_result }

单数字毫秒级开销。全程进程内完成。

快速上手 — 3 行配置

import asyncio
from enterprise_claude import AgentOrchestrator, GovernanceLayer, TokenMonitor

governance = GovernanceLayer(
    pii_filter=True,                          # email, SSN, phone, card, IP
    blocked_keywords=["MNPI", "classified"],
    max_prompt_length=50_000,
)
monitor = TokenMonitor(daily_budget_usd=100.0, alert_threshold_pct=0.80)

async def main():
    async with AgentOrchestrator(governance=governance, monitor=monitor) as orch:
        agent = await orch.create_agent(
            name="Analyst",
            system_prompt="You are a concise financial analyst.",
            persona="research",
        )
        result = await agent.run(
            "Summarise Q3 macro risks in 3 bullets.",
            user_id="alice@acme.com"
        )
        print(result.content)
        print(f"${result.cost_usd:.6f}")
        print(result.governance_result.passed)  # True

asyncio.run(main())

🛡️ GovernanceLayer — 正则，而非 AI

我做了一个审慎的选择：治理决策使用编译后的正则表达式，而不是再调用一个模型。

_PII_PATTERNS = {
    "email":       re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
    "phone":       re.compile(r'\b(\+1[-.\s]?)?\(?\d{3}\)?[-.\s]\d{3}[-.\s]\d{4}\b'),
    "ssn":         re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
    "credit_card": re.compile(r'\b(?:\d[ -]?){13,16}\b'),
    "ipv4":        re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b'),
}

为什么不用 spaCy 或 NER 模型？可审计性。正则表达式是一份你可以阅读、测试、展示给监管者的契约。神经模型是一个概率分布。当合规部门问"PII 检测是怎么工作的？"，你想要一个一行就能回答的答案。

在 prompt 中检测到 PII 会抛出 GovernanceViolation 并中止——API 根本不会被调用。响应中的 PII 会被标记（而不是抛出）——因为调用已经发生了。

try:
    result = await agent.run(prompt, user_id="alice")
except GovernanceViolation as e:
    print(e.violation_type)  # "pii_detected" | "blocked_keyword" | "prompt_too_long"

你也可以注册 pre/post 钩子——同步或异步，自动检测：

governance = GovernanceLayer(
    pii_filter=True,
    pre_hooks=[lambda prompt, ctx: logger.info(f"Prompt: {prompt[:100]}")],
    post_hooks=[async_verify_response],
)

💰 TokenMonitor — 重启后依然准确的预算

每次调用都会记录到 SQLite：

INSERT INTO usage_records (
    timestamp, agent_id, model, persona,
    input_tokens, output_tokens, cost_usd
) VALUES (?, ?, ?, ?, ?, ?, ?)

预算状态是从数据库重新计算的——而不是从内存计数器。重启进程后，预算计数仍然是准确的。

monitor = TokenMonitor(daily_budget_usd=100.0, alert_threshold_pct=0.80)

@monitor.on_alert
async def on_budget_alert(status):
    await slack.post(
        f"⚠️ Budget at {status.pct_used:.0%} — "
        f"${status.used_usd:.2f} of ${status.budget_usd:.2f} used today"
    )

# Query spend
summary = await monitor.get_cost_summary(start_date=yesterday, end_date=now)
print(summary.by_model)    # {"claude-sonnet-4-6": 0.042, "claude-haiku-4-5": 0.008}
print(summary.by_persona)  # {"research": 0.031, "engineering": 0.019}

📋 AuditLogger — 设计上防篡改

只追加——代码库中没有任何 UPDATE 或 DELETE

每个事件都有 SHA-256 校验和（规范 JSON，sort_keys=True）

用户身份哈希存储——SHA-256(user_id) 存储，从不存原始邮箱

audit = AuditLogger(db_path="audit.db", gxp_mode=True)

# Verify the entire log — returns {event_id: bool}
checksums = await audit.verify_checksums()
tampered = [eid for eid, ok in checksums.items() if not ok]

if not tampered:
    print(f"✅ All {len(checksums)} events verified clean")

在 GxP 模式下（制药行业 21 CFR Part 11），还会检查响应中是否包含 [SOURCE:]、[REF:] 或 [CITATION:] 标记——模型必须引用来源，否则响应会被标记。

📈 AdoptionTracker — 带硬性关卡的波浪式推广

wave1 = await tracker.create_wave(name="Architects", target_count=50, order=1)
wave2 = await tracker.create_wave(
    name="Developers", target_count=200, order=2,
    gate_wave_id=wave1.wave_id,
    gate_threshold_pct=0.80,
)

await tracker.activate_wave(wave1.wave_id)

# Try to open Wave 2 too early:
try:
    await tracker.activate_wave(wave2.wave_id)
except WaveGateError as e:
    print(e)
    # "Wave 'Developers' requires 'Architects' to reach 80.0% (currently 12.0%)"

关卡是 WaveGateError，不是流程指南。你绕不过去。

使用率评分（0-100）是从调用频率在对数尺度上计算的——log10(calls + 1) / log10(101) * 100。这是一个有意义的代理指标，衡量的是真实参与度而非点击通过率。

🔌 MCP Registry — 无样板代码的连接器

from enterprise_claude import get_connector, list_connectors

github = get_connector("github")
print(github.required_env_vars)  # ["GITHUB_TOKEN"]

for c in list_connectors():
    status = "✅" if c["env_configured"] else "❌ missing env vars"
    print(f"{c['display_name']:<20} {status}")

10 个预置连接器：GitHub、Jira、Slack、Confluence、SharePoint、PostgreSQL、ServiceNow、Salesforce、Teams、本地文件系统。

⌨️ CLI — 无需写代码的运维

ecl cost summary --days 7        # spend breakdown by model and persona
ecl audit query --persona research --result pass
ecl audit export --format csv --out trail.csv
ecl waves list                   # all waves + progress bars
ecl waves activate <wave-id>
ecl connectors list              # env-var status for all 10 connectors
ecl connectors validate github

值得解释的设计决策

LangChain 是一个链式调用 LLM 的框架。这个库是生产部署的治理层。这是两个不同的问题。引入 LangChain 意味着：

50+ 个传递依赖
掩盖直接 API 调用的抽象层
框架版本与治理层版本绑定

这个库只有一件事：让每次 Claude 调用都受到治理、追踪和审计。它直接调用 anthropic.AsyncAnthropic.messages.create()。没有链、没有 agent 框架、没有向量存储。

为什么不选 PostgreSQL 而选 SQLite？

SQLite 零基础设施。你可以不用数据库服务器就开始使用这个库。 schema 足够简单，如果你需要生产级规模，向 asyncpg 的 PostgreSQL ATTACH DATABASE 迁移有文档记录。先简单起步，有问题了再扩展。

为什么用 frozen Pydantic 配置？

config = GovernanceConfig(pii_filter=True, blocked_keywords=["MNPI"])
config.pii_filter = False  # raises ValidationError

治理规则在运行过程中不能漂移。不可变配置是一种强制机制，不是风格偏好。

enterprise_claude/
├── governance.py        # GovernanceLayer — the policy engine
├── orchestrator.py      # AgentOrchestrator — agent lifecycle
├── token_monitor.py     # TokenMonitor — cost accounting
├── audit.py             # AuditLogger — immutable event log
├── adoption_tracker.py  # AdoptionTracker — wave rollout
├── mcp_connectors.py    # MCPConnectorRegistry — connector catalogue
└── cli.py               # ecl command — management CLI

tests/                   # 66 pytest tests
examples/
├── basic_governed_agent.py    # ← start here
├── clinical_trial_agent.py    # GxP + audit trail
└── sdlc_accelerator.py        # wave simulation (no API key needed)

pip install enterprise-claude-kit
cp .env.example .env     # add ANTHROPIC_API_KEY
python examples/basic_governed_agent.py

GitHub: github.com/sairajboddula/enterprise-claude-kit PyPI: pypi.org/project/enterprise-claude-kit

MIT 协议。欢迎 PR。如果你在受监管行业或大型企业中构建 Claude 集成，我很想知道你正在做什么。

你在 Claude 部署中解决了哪些治理问题？欢迎在下方留言。