nujovich/tracepath

# Tracepath **用于 AI 治理的可审计多代理中间件。** 只需一次 `docker compose up`，即可让任何代理框架（LangChain、CrewAI、AutoGen、LangGraph）符合 EU AI Act、FINRA 和 SOC2 标准。 ## 目录 - [快速开始](#quickstart) - [Tracepath 的功能](#what-tracepath-does) - [架构](#architecture) - [仪表盘](#dashboard) - [API 参考](#api-reference) - [Python SDK](#python-sdk) - [策略引擎](#policy-engine) - [事件检测](#incident-detection) - [Gemini 语义分类器](#gemini-semantic-classifier) - [合规报告](#compliance-reports) - [可观测性](#observability) - [路线图](#roadmap) - [许可证](#license) ## 快速开始 ### 前置条件 - Docker + Docker Compose - Python 3.10+（用于 SDK） - Rust 1.80+（用于网关开发） ### 1. 启动技术栈 ``` git clone https://github.com/nujovich/tracepath.git cd tracepath/docker AUDIT_SIGNING_KEY=$(openssl rand -hex 32) docker compose up -d ``` ### 2. 打开仪表盘 ``` http://localhost:3000 ``` 五个标签页：**Audit**、**Incidents**、**Policies**、**Reports** 和 **Gemini**。 ### 3. 记录您的第一个审计步骤 ``` curl -X POST http://localhost:9001/audit/step \ -H "Content-Type: application/json" \ -d '{ "session_id":"demo","agent_id":"test","agent_type":"coder", "step_number":1,"tool_name":"read_file", "tool_input":{"path":"/tmp/test"},"tool_output":{"lines":10}, "timestamp":"2026-07-12T12:00:00Z" }' # → {"status":"recorded","signature":"","policy_decision":{"allowed":true,"denials":[]}} ``` ### 4. 使用 Python SDK ``` from tracepath_sdk import AsyncAuditClient, audit client = AsyncAuditClient(agent_type="coder") @audit(client) async def read_file(path: str) -> dict: return {"lines": 42} async with client: result = await read_file(path="/tmp/x") events = await client.query_events(limit=10) incidents = await client.get_incidents() ``` ## Tracepath 的功能 Tracepath 位于您的 AI 代理与其调用的工具之间。每次工具调用都会被拦截、签名、根据策略进行检查，并存储在不可篡改的审计追踪记录中。实时事件检测器会监控异常情况——例如拒绝激增、预算超支、可疑模式和速率限制违规——并在仪表盘中展示出来。可选的 Gemini 分类器可利用语义分析来细化严重程度。 | 功能 | 工作原理 | |---|---| | **对每次调用进行签名** | 每个事件使用 Ed25519 签名 → 加密级别的不可否认性 | | **执行策略** | OPA WASM 引擎在 <1ms 内评估允许列表、预算和速率限制 | | **不可篡改的审计日志** | PostgreSQL（可查询）+ MinIO S3 Object Lock（WORM，365 天保留期） | | **检测事件** | NATS JetStream → 实时检测器 → 仪表盘警报 | | **严重程度分类** | 通过 OpenRouter 调用 Gemini 2.5 Flash → 对基于阈值的警报进行语义细化 | | **策略版本控制** | 基于 Git 的策略版本控制，支持 diff、回滚和重放 | | **生成报告** | 一键生成 FINRA Rule 4511 和 EU AI Act Article 50 合规报告 | | **多 SDK 支持** | Python（异步 + 同步，`@audit` 装饰器）、TypeScript、Java | ## 架构 ``` ┌─────────────────────────────────────────────────────────┐ │ Agent (LangChain / CrewAI / custom) │ │ │ │ │ ▼ │ │ Tracepath SDK (Python / TypeScript / Java) │ │ │ │ │ ▼ │ │ Audit Gateway (Rust + actix-web, :9001) │ │ ├─ Ed25519 signing ───────────────► PostgreSQL │ │ ├─ OPA WASM policy engine ────────► allow / deny │ │ ├─ Allowed events ────────────────► MinIO S3 (WORM) │ │ │ │ │ ├─ NATS JetStream ────────────────► Incident Detector│ │ │ ├─ Denial spike│ │ │ ├─ Budget │ │ │ ├─ Suspicious │ │ │ ├─ Rate limit │ │ │ └─ Gemini 2.5 │ │ │ (semantic) │ │ │ │ │ ├─ Proxy ─────────────────────────► Policy API (:9003)│ │ │ ├─ Versions │ │ │ ├─ Diff │ │ │ └─ Rollback │ │ │ │ │ └─ API ───────────────────────────► Dashboard (:3000)│ │ ├─ Audit trail│ │ ├─ Incidents │ │ ├─ Policies │ │ ├─ Reports │ │ └─ Gemini │ └─────────────────────────────────────────────────────────┘ ``` ## 仪表盘  *Gemini 标签页，显示启用状态、模型信息和缓存的语义分类。*  *Incidents 标签页，包含实时检测时间线：CRITICAL 拒绝激增，WARNING 预算超支。*  *Policies 标签页，包含 Git 版本历史记录、并排差异查看器以及一键回滚功能。*  *Reports 标签页，支持一键生成 FINRA 和 EU AI Act 合规报告。*  *Audit 标签页，包含事件追踪、策略决策明细和工具使用统计信息。* ## API 参考 | 方法 | 路径 | 描述 | |---|---|---| | `GET` | `/health` | 网关健康状态 | | `GET` | `/health/policy` | OPA 策略引擎冒烟测试 | | `POST` | `/audit/step` | 记录审计步骤（已签名 + 经过策略检查） | | `GET` | `/audit/events` | 查询审计事件（支持过滤：`session_id`、`agent_id`、`tool_name`；支持分页） | | `GET` | `/incidents` | 查询由事件服务检测到的事件 | | `GET` | `/reports` | 列出生成的合规报告 | | `POST` | `/reports/generate` | 生成合规报告（`type: "finra" \| "eu-ai-act"`） | | `GET` | `/reports/{name}` | 下载特定的报告文件 | | `GET` | `/gemini` | Gemini 分类器状态和缓存分类 | | `GET` | `/policies/versions` | 列出策略版本（Git 历史记录） | | `GET` | `/policies/diff?old=&new=` | 两个策略版本之间的统一差异对比 | | `POST` | `/policies/rollback` | 回滚到先前的策略版本 | ## Python SDK ``` from tracepath_sdk import AsyncAuditClient, SyncAuditClient, audit, PolicyDenied # ── 带 decorator 的 Async client ────────────────── client = AsyncAuditClient(agent_type="coder") @audit(client) async def read_file(path: str) -> dict: return {"lines": 42} async with client: # Auto-audited call result = await read_file(path="/tmp/x") # Manual audit resp = await client.record_step("terminal", {"cmd": "ls"}, {"exit": 0}) # Query the audit trail events = await client.query_events(session_id=client.session_id) print(f"{events.count} events recorded") # Check for incidents for inc in await client.get_incidents(): print(f"[{inc.severity}] {inc.type}: {inc.message}") # ── Sync client ────────────────────────────────── with SyncAuditClient(agent_type="researcher") as sync: sync.record_step("web_search", {"q": "test"}, {"results": 3}) print(sync.health()) ``` ### SDK 功能 | 功能 | API | |---|---| | 异步客户端 | `AsyncAuditClient`，支持 `async with` 上下文管理器 | | 同步客户端 | `SyncAuditClient`，用于脚本的轻量级包装器 | | `@audit` 装饰器 | 包装任何异步/同步函数；工具名称 = 函数名称 | | 策略拒绝检测 | 带有 `.denials` 列表和 `.signature` 的 `PolicyDenied` 异常 | | 审计追踪查询 | `query_events(session_id, agent_id, tool_name, limit, offset)` | | 事件时间线 | `get_incidents(limit)` | | Pydantic 模型 | `AuditResponse`、`AuditQueryResult`、`Incident`、`PolicyDecision` | | 测试 | 14 项测试（7 项单元测试 + 7 项集成测试） | ## 策略引擎 三种基础策略，编译为 OPA WASM 并在网关处进行评估： | 策略 | 规则 | |---|---| | **允许列表** | 拒绝不在代理类型允许集合中的工具 | | **预算** | 如果累计工具成本超过会话预算，则拒绝 | | **速率限制** | 如果每个会话超过 60 次调用/分钟，则拒绝 | ### 代理类型允许列表 | 代理类型 | 允许的工具 | |---|---| | `coder` | `read_file`, `write_file`, `terminal`, `search_files`, `patch`, `execute_code` | | `researcher` | `web_search`, `web_extract`, `browser_navigate`, `browser_snapshot` | | `assistant` | `read_file`, `web_search`, `web_extract`, `terminal` | | `default` | `read_file`, `web_search`, `web_extract` | ### 重新编译策略 ``` cd policies opa build -t wasm -e tracepath/main/decision -o bundle.tar.gz rules/ # 重启 gateway 以加载新 bundle ``` ## 事件检测 利用 NATS JetStream 流式传输进行实时异常检测。 | 检测器 | 触发条件 | 严重程度 | |---|---|---| | **拒绝激增** | 单个会话中出现 >5 次策略拒绝 | CRITICAL | | **预算超支** | 每个会话累计成本超过 1000 美分 | WARNING | | **可疑模式** | 连续调用同一工具 ≥10 次 | WARNING | | **速率限制违规** | 每个会话超过 60 次调用/分钟 | WARNING | ### 检测管道 ``` Audit event → Gateway → NATS JetStream → Incident Detector (Python) │ ├─ Threshold pass (Rego-like rules) ├─ Gemini refinement (semantic pass) └─ Incident persisted → Dashboard API ``` ## Gemini 语义分类器 由通过 OpenRouter 调用的 **Google Gemini 2.5 Flash** 提供支持。它接收由阈值触发的事件，并通过分析语义上下文来细化其严重程度。 | 事件 | 初始严重程度 | Gemini 推理 | 最终严重程度 | |---|---|---| | 单个会话中出现 6 次 `image_generate` 拒绝 | CRITICAL | *"所有拒绝都是针对图像生成的，这表明是配置错误而非恶意尝试"* | WARNING | | 预算超支 2 倍 | WARNING | *"使用的工具符合合法的研究工作流"* | INFO | ### 后端支持 - **OpenRouter** — 设置 `OPENROUTER_API_KEY`（推荐） - **Google AI 原生** — 设置 `GOOGLE_API_KEY` ``` # 使用 OpenRouter OPENROUTER_API_KEY=sk-or-v1-... docker compose up -d # 或者显式设置 model OPENROUTER_MODEL=google/gemini-2.5-flash docker compose up -d ``` ### 持久化缓存 所有 Gemini 分类都会缓存到磁盘（`/data/gemini-cache.json`），并在容器重启后依然存在。仪表盘会显示分类历史记录以及每个事件的推理过程。 ## 合规报告 直接在仪表盘中一键生成监管报告。 | 报告 | 标准 | 状态 | |---|---|---| | **FINRA** | Rule 4511（账簿与记录） | ✅ 合规 | | **EU AI Act** | Article 50（透明度与监督） | ✅ 合规 | 每份报告都会验证： - **数据完整性** — 每个事件上的 Ed25519 签名 - **记录保留** — WORM 存储，最低保留 365 天 - **访问控制** — API 级别的身份验证 - **可搜索的记录** — 完整的 PostgreSQL 查询功能 - **人工监督** — 基于仪表盘的监控和干预 ## 可观测性 网关通过 OTLP (HTTP) 将追踪导出到 **LangFuse**。默认处于禁用状态，可通过以下方式启用： ``` export OTEL_EXPORTER_OTLP_ENDPOINT="https://cloud.langfuse.com/api/public/otel" export OTEL_SERVICE_NAME="tracepath-gateway" export LANGFUSE_PUBLIC_KEY="pk-lf-..." export LANGFUSE_SECRET_KEY="sk-lf-..." ``` 未设置时，网关会将日志输出到 stdout（JSON 格式），且没有 OTLP 开销。 ## 路线图 ### 当前（周末构建 — 已完成 ✅） - [x] 每个事件的 Ed25519 签名 - [x] OPA WASM 策略引擎（允许列表、预算、速率限制） - [x] 带查询 API 的 PostgreSQL 审计日志 - [x] WORM 存储（MinIO S3 Object Lock） - [x] NATS JetStream 事件总线 - [x] 实时事件检测（4 个检测器） - [x] 基于 Git 的策略版本控制（差异对比 + 回滚 + 重放） - [x] FINRA + EU AI Act 合规报告 - [x] Gemini 2.5 Flash 语义分类器（通过 OpenRouter） - [x] React 仪表盘（5 个标签页） - [x] Python SDK（异步 + 同步，`@audit` 装饰器） - [x] TypeScript SDK - [x] Java SDK ### 接下来（第 1-4 周） - [ ] **Helm chart** — 部署到任何 Kubernetes 集群 - [ ] **PDF 报告** — 将 FINRA/EU AI Act 报告导出为可下载的 PDF - [ ] **Webhook 警报** — 针对严重事件的 Slack、Discord、PagerDuty 警报 - [ ] **自定义策略 UI** — 直接在仪表盘中编写和测试 Rego 规则 - [ ] **TypeScript SDK 功能对齐** — `@audit` 装饰器 + 与 Python SDK 的全面功能对齐 ### 不久后（第 1-3 个月） - [ ] **SOC2 准备就绪包** — 为 SOC2 审计员预先构建的证据收集 - [ ] **AWS Marketplace 上架** — 从 AWS 控制台一键部署 - [ ] **多租户仪表盘** — 组织级别的范围界定和 RBAC - [ ] **端到端重放引擎** — 带有对比报告的完整历史重放 - [ ] **LangFuse 集成** — 原生 OpenTelemetry 导出及预构建的仪表盘 ### 愿景（6 个月以上） - [ ] **策略市场** — 分享和导入社区 OPA 策略 - [ ] **联邦审计** — 利用零知识证明进行跨组织的审计追踪共享 - [ ] **代理评分** — 每个代理/会话的合规评分，在仪表盘中可见 - [ ] **实时干预** — 检测到严重事件时自动暂停代理 ## 许可证 Apache 2.0 *由 [Nadia Ujovich](https://github.com/nujovich) 用 ❤️‍🔥 打造*

标签：AI治理, API集成, DLL 劫持, 中间件, 可观测性, 可视化界面, 大语言模型, 政策引擎, 测试用例, 请求拦截, 逆向工具