尧图网站建设 
创业公司在全球扩张时最容易被忽视却又最致命的问题往往不是产品功能而是合规。不同国家的数据保护法、劳动法、财税要求交织在一起人工追踪效率低、易遗漏等真正面对审计时才补文档已经来不及了。Veritas 就是一个面向全球初创公司的 AI 合规追踪器它把“合规”从一堆分散的 PDF 和表格里抽出来变成可分配、可追踪、可验证的任务流。这篇文章会以 Veritas 项目为原型完整拆解一个 AI 驱动的合规追踪系统应该怎么做。内容覆盖需求分析、系统设计、数据建模、AI 能力接入、最小可运行项目代码以及常见问题和工程建议。无论你是后端开发者还是刚接触 AI 应用开发的初学者都可以照着本文思路搭建一个自己的合规追踪 MVP再逐步扩展成生产级系统。1. 背景与核心概念1.1 什么是合规追踪器合规追踪器简单说就是一套用来管理“企业必须遵守哪些规则、需要完成哪些合规动作、每件事做到什么程度”的系统。它不是在帮企业规避法律而是帮企业把法律义务转变成具体的任务清单。传统做法是法务或运营负责人手工维护一张表格记录法规名称、适用范围、责任部门、到期时间、执行状态。对于早期项目可能够用但一旦业务扩展到多个地区规则数量会成倍增长。比如在欧盟运营要关注 GDPR在美国做消费者业务可能涉及 CCPA/CPRA在东南亚开展业务又各有本地隐私法。每一个法规又可能拆出十几条具体义务。合规追踪器要解决的核心问题有三个规则分散法律文本散落在不同政府网站和专业数据库中难以集中管理。责任不清一条合规义务没有明确的 owner最后往往没人跟进。时效难控哪些义务需要年度复审哪些需要 30 天内响应靠人记不现实。把这些问题系统化就是合规追踪器存在的价值。1.2 Veritas 项目的定位Veritas 的名字来源于拉丁语“真理”。这个项目的核心理念是用 AI 辅助人类处理海量法规文本但最终决策和审计责任仍然由人来承担。它不是要做成一个自动判断公司是否违法的“审判系统”而是做成一个“追踪提醒辅助理解”的平台。具体来说Veritas 会做以下几件事把导入的法规文本通过 AI 进行结构化抽取提取义务主体、动作、时间要求、处罚风险。将不同的义务自动分类到对应部门比如工程团队负责隐私保护、财务团队负责税务申报。根据义务的紧急性、影响范围、处罚金额自动计算风险等级。生成一条条可执行任务支持负责人分配、截止日期提醒、完成度跟踪。所有操作记录审计日志方便合规审计时出具证据链。这里要特别强调AI 在合规场景里只能做“辅助”不能替代专业判断。因为法规解读涉及到具体业务上下文同一个条款在不同行业、不同规模的公司里含义可能完全不同。所以 Veritas 的设计原则是“AI 建议人工确认”。1.3 为什么初创团队需要重点关注合规很多初创团队觉得合规是大公司的事自己还小先跑起来再说。但实际情况恰恰相反越早期越容易因为某个基础动作不规范在融资尽调或客户安全审计时被动。举一个很常见的场景初创公司做了一款 SaaS 工具拿到了第一个海外客户。客户发来一份安全合规问卷要求说明数据存储位置、保留策略、访问日志、信息安全标准。如果公司内部没有一套追踪机制问卷里的问题很难回答甚至可能因为答不上来而丢掉订单。Veritas 这类系统在早期阶段的价值不是用来自动完成合规认证而是让团队始终保持“知道自己该做什么”的状态。这正是技术工程中最有价值的部分把模糊的法律条款变成清晰的项目管理动作。2. 系统设计与功能拆分2.1 需求分析在动手写代码之前先梳理一下系统需要哪些角色和流程。Veritas 的最小可用版本需要支持三类角色管理员负责维护合规规则库、配置 AI 模型、管理用户。合规负责人负责导入法规、审核 AI 抽取结果、分派任务。团队成员查看分配给自己的义务提交完成证明更新状态。核心业务流如下合规负责人导入一份法规文本可以是 PDF、Word 或纯文本。后端调用 AI 服务把文本拆成结构化条款。AI 对每条条款进行归类识别出“义务动作”“适用对象”“时间期限”“风险等级”。合规负责人逐条确认或修改 AI 结果。确认后的数据生成一条义务记录并自动派生一个或多个任务。任务分配负责人后系统在截止日期前发送提醒。负责人完成任务后系统记录完成人、完成时间、证明材料。所有关键操作写入审计日志。2.2 模块拆分按照上面的流程Veritas 可以拆成以下模块模块职责规则管理法规文本上传、版本管理、条款结构维护AI 抽取服务调用大模型提取结构化信息义务管理管理合规义务维护风险等级、责任方、到期时间任务管理任务的创建、分配、完成、延期提醒通知按截止时间发送邮件或站内通知审计日志记录用户操作和 AI 行为支持追溯用户与权限角色管理控制不同用户可见的数据范围为了控制 MVP 的复杂度暂时不做复杂的权限矩阵只实现管理员和普通用户两个角色。2.3 技术栈选型对于 MVP我推荐以下技术栈后端框架Python FastAPI开发效率高天然支持异步自带 OpenAPI 文档。数据库PostgreSQL事务支持好适合业务系统也可以先用 SQLite 快速验证。ORMSQLAlchemy 2.0兼容主流数据库。任务调度APScheduler负责循环检查到期任务并触发通知。AI 服务OpenAI 兼容接口也可以用本地模型或各类 API 网关。本文代码以兼容接口为例模型只要支持对话补全即可。部署Docker Compose方便本地启动 PostgreSQL 和 Redis。版本方面Python 建议 3.10 以上FastAPI 使用最新稳定版即可。因为依赖版本更新较快本文示例以常规写法为主实际运行时按你的环境锁定版本。3. 数据模型设计数据模型是整个系统的地基。Veritas 的核心实体包括用户、合规规则、合规义务、任务、通知日志和审计日志。3.1 核心表结构我们先用自然语言描述再转成 SQLAlchemy 模型。用户表id、用户名、邮箱、角色、创建时间。合规规则表法规名称、法规编号、发布日期、原文存储路径、当前版本。条款表属于哪个规则条款编号原始文本AI 抽取结果人工确认状态。义务表从条款中提炼出来的具体义务包含适用对象、动作描述、截止类型、风险等级、负责人。任务表某个义务派生的执行任务包含截止日期、状态、负责人、完成时间。通知日志表记录每次提醒通知的类型、接收人、发送时间、关联任务。审计日志表记录操作者、操作类型、操作详情、IP、时间。3.2 SQLAlchemy 模型示例下面是一份简化后的模型代码。文件路径backend/app/models.py。# backend/app/models.py from datetime import datetime from sqlalchemy import ( String, Integer, Boolean, DateTime, Text, ForeignKey, Enum, Float, JSON, LargeBinary ) from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.dialects.postgresql import UUID import uuid class Base: id: Mapped[str] mapped_column( UUID(as_uuidTrue), primary_keyTrue, defaultuuid.uuid4 ) created_at: Mapped[datetime] mapped_column( DateTime, defaultdatetime.utcnow ) class User(Base): __tablename__ users username: Mapped[str] mapped_column(String(50), uniqueTrue, indexTrue) email: Mapped[str] mapped_column(String(120), uniqueTrue, indexTrue) role: Mapped[str] mapped_column(String(20), defaultuser) is_active: Mapped[bool] mapped_column(Boolean, defaultTrue) class ComplianceRule(Base): __tablename__ compliance_rules title: Mapped[str] mapped_column(String(200)) rule_code: Mapped[str] mapped_column(String(100), uniqueTrue, indexTrue) jurisdiction: Mapped[str] mapped_column(String(50), indexTrue) published_date: Mapped[datetime] mapped_column(DateTime, nullableTrue) source_url: Mapped[str] mapped_column(String(500), default) version: Mapped[str] mapped_column(String(20), default1.0) class ComplianceClause(Base): __tablename__ compliance_clauses rule_id: Mapped[str] mapped_column(ForeignKey(compliance_rules.id)) clause_number: Mapped[str] mapped_column(String(50)) original_text: Mapped[str] mapped_column(Text) ai_extracted: Mapped[dict] mapped_column(JSON, defaultdict) human_confirmed: Mapped[bool] mapped_column(Boolean, defaultFalse) class ComplianceObligation(Base): __tablename__ compliance_obligations clause_id: Mapped[str] mapped_column(ForeignKey(compliance_clauses.id)) title: Mapped[str] mapped_column(String(200)) description: Mapped[str] mapped_column(Text) required_action: Mapped[str] mapped_column(String(200)) applicable_object: Mapped[str] mapped_column(String(200)) risk_level: Mapped[str] mapped_column(String(20), defaultmedium) due_type: Mapped[str] mapped_column(String(20), defaultone_time) due_days: Mapped[int] mapped_column(Integer, default30) owner_id: Mapped[str] mapped_column(ForeignKey(users.id), nullableTrue) class ComplianceTask(Base): __tablename__ compliance_tasks obligation_id: Mapped[str] mapped_column(ForeignKey(compliance_obligations.id)) title: Mapped[str] mapped_column(String(200)) assignee_id: Mapped[str] mapped_column(ForeignKey(users.id), nullableTrue) due_date: Mapped[datetime] mapped_column(DateTime) status: Mapped[str] mapped_column(String(20), defaultpending) completed_at: Mapped[datetime] mapped_column(DateTime, nullableTrue) evidence: Mapped[str] mapped_column(Text, default) class NotificationLog(Base): __tablename__ notification_logs task_id: Mapped[str] mapped_column(ForeignKey(compliance_tasks.id)) receiver_email: Mapped[str] mapped_column(String(120)) notification_type: Mapped[str] mapped_column(String(30)) sent_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class AuditLog(Base): __tablename__ audit_logs user_id: Mapped[str] mapped_column(ForeignKey(users.id), nullableTrue) action: Mapped[str] mapped_column(String(50)) entity_type: Mapped[str] mapped_column(String(50)) entity_id: Mapped[str] mapped_column(String(50)) detail: Mapped[dict] mapped_column(JSON, defaultdict) ip_address: Mapped[str] mapped_column(String(50), default)这个模型覆盖了核心业务链路。其中ai_extracted字段用 JSON 类型保存大模型抽取出的结构化结果方便后期字段演进不用频繁改表结构。4. AI 辅助能力拆解合规追踪器接入 AI 的好处不是让 AI 替人做决定而是把“读法规、找要点、初步归类”这种重复劳动自动化。下面我们拆解四个核心 AI 能力。4.1 法规文本结构化抽取原始法规是一整篇长文本AI 要做的事情是把它按条款拆分并提取关键信息。一个典型的 prompt 结构如下你是合规分析助手。请从提供的法规文本中提取条款列表。 每个条款需要包含 - clause_number: 条款编号 - summary: 条款摘要 - required_action: 条款要求企业采取的动作 - applicable_object: 适用的业务对象如 网站、移动应用、员工数据 - due_type: 义务类型one_time 一次性 / recurring 周期性 / event_driven 事件驱动 - due_days: 合规动作应在多少天内完成 - risk_level: high/medium/low 请只输出 JSON 数组不要包含其他文字。这里的关键是给模型一个非常明确的“输出格式”。否则模型会生成一大段解释不利于下游解析。4.2 义务自动分类抽取出的条款很多直接把每条都变成任务会给团队带来负担。所以 Veritas 会让 AI 对义务做一次“初步分类”比如工程安全类数据隐私类财务会计类人力资源类商业模式合规类分类合理之后系统可以自动把任务分配给对应的业务线。4.3 风险等级评估风险等级不能完全依赖模型拍脑袋。我们在 prompt 里可以要求模型基于三个维度打分违反后可能的罚款金额监管关注程度对业务运营的影响程度再让模型输出一个综合等级。不过风险等级更适合作为“建议”后续由人工复核。4.4 生成执行建议AI 还可以根据义务内容生成建议步骤。比如某条义务要求“用户删除账户后 30 天内彻底清除数据”AI 可以建议工程团队设计定时任务、数据库级联删除逻辑、以及删除结果留存证明。这部分输出适合放在义务详情页让负责人知道“下一步该做什么”。4.5 AI 能力实现注意事项使用大模型处理合规文本时有几点必须注意不要上传不必要的敏感数据生产环境建议对接可信的私有化模型或合规数据网关。prompt 中必须要求模型输出 JSON并且在代码里做异常兜底。所有 AI 输出必须留痕和原始条款文本绑定方便溯源。AI 抽取结果只能作为草稿必须有一个人工确认的环节。下面给出一个简单的 AI 客户端示例使用 requests 调用 OpenAI 兼容接口。# backend/app/ai_client.py import json import requests class AIClient: def __init__(self, api_base, api_key, model): self.api_base api_base.rstrip(/) self.api_key api_key self.model model def analyze_text(self, text: str) - list: prompt 你是合规分析助手。请从法规文本中提取条款列表。 每个条款需要包含clause_number, summary, required_action, applicable_object, due_type(one_time/recurring/event_driven), due_days, risk_level(high/medium/low)。 请只输出 JSON 数组不要包含其他文字。 payload { model: self.model, messages: [ {role: system, content: prompt}, {role: user, content: text[:8000]} ], temperature: 0.2 } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } resp requests.post( f{self.api_base}/chat/completions, headersheaders, jsonpayload, timeout60 ) resp.raise_for_status() content resp.json()[choices][0][message][content] # 解析模型返回的 JSON try: return json.loads(content) except json.JSONDecodeError: # 如果模型输出包含 json 代码块做一次清理 cleaned content.strip().removeprefix(json).removesuffix() return json.loads(cleaned)这段代码不绑定某个特定厂商只要你的模型服务提供/chat/completions接口就能用。5. 完整实战案例最小可运行 MVP接下来我们实现一个可运行的 MVP。为了不让文章过长这里只实现核心链路导入法规文本 → AI 抽取 → 任务创建 → 查询任务列表。5.1 项目结构veritas-demo/ ├── backend/ │ ├── app/ │ │ ├── __init__.py │ │ ├── main.py │ │ ├── config.py │ │ ├── database.py │ │ ├── models.py │ │ ├── schemas.py │ │ └── ai_client.py │ └── requirements.txt ├── docker-compose.yml └── README.md5.2 requirements.txtfastapi0.111.0 uvicorn[standard]0.30.1 sqlalchemy2.0.30 psycopg2-binary2.9.9 python-dotenv1.0.1 requests2.32.3 pydantic2.7.4如果你本地还没有 PostgreSQL可以直接使用 Docker Compose 启动一个。5.3 docker-compose.ymlversion: 3.9 services: db: image: postgres:15 container_name: veritas_db environment: POSTGRES_USER: veritas POSTGRES_PASSWORD: veritas POSTGRES_DB: veritas ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data redis: image: redis:7 container_name: veritas_redis ports: - 6379:6379 volumes: pgdata:在真实项目中 Redis 可以用于缓存和异步任务队列MVP 阶段暂时不接入但先预留。5.4 配置文件# backend/app/config.py import os from dotenv import load_dotenv load_dotenv() class Settings: DATABASE_URL os.getenv( DATABASE_URL, postgresqlpsycopg2://veritas:veritaslocalhost:5432/veritas ) AI_API_BASE os.getenv(AI_API_BASE, https://your-ai-gateway.example.com) AI_API_KEY os.getenv(AI_API_KEY, your-api-key) AI_MODEL os.getenv(AI_MODEL, compliance-assistant) settings Settings()这里要注意AI_API_BASE只是一个示例占位地址实际部署时请改成你自己的模型服务地址。不要把你自己的真实密钥提交到代码库。5.5 数据库连接# backend/app/database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase from .config import settings engine create_engine(settings.DATABASE_URL, echoFalse) SessionLocal sessionmaker(bindengine, autocommitFalse, autoflushFalse) class Base(DeclarativeBase): pass5.6 Pydantic 模型# backend/app/schemas.py from pydantic import BaseModel, Field from datetime import datetime class ClauseCreate(BaseModel): rule_id: str clause_number: str original_text: str class ClauseOut(ClauseCreate): id: str ai_extracted: dict human_confirmed: bool class Config: from_attributes True class ObligationOut(BaseModel): id: str title: str risk_level: str owner_id: str | None due_type: str due_days: int class TaskCreate(BaseModel): obligation_id: str title: str assignee_id: str | None None due_date: datetime class TaskOut(TaskCreate): id: str status: str completed_at: datetime | None None class Config: from_attributes True5.7 FastAPI 主程序# backend/app/main.py from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from .database import engine, SessionLocal, Base from .models import ( User, ComplianceRule, ComplianceClause, ComplianceObligation, ComplianceTask ) from .schemas import ( TaskCreate, TaskOut, ObligationOut, ClauseOut ) from .ai_client import AIClient from .config import settings Base.metadata.create_all(bindengine) app FastAPI(titleVeritas Compliance Tracker, version0.1.0) def get_db(): db SessionLocal() try: yield db finally: db.close() app.post(/rules/import) def import_rule(rule_title: str, rule_code: str, jurisdiction: str, raw_text: str, db: Session Depends(get_db)): 模拟一个简单的规则导入接口。 实际项目中应该上传文件并解析文本这里为了演示直接传入原文。 rule ComplianceRule( titlerule_title, rule_coderule_code, jurisdictionjurisdiction, ) db.add(rule) db.flush() client AIClient( api_basesettings.AI_API_BASE, api_keysettings.AI_API_KEY, modelsettings.AI_MODEL ) try: clauses client.analyze_text(raw_text) except Exception as e: raise HTTPException(status_code500, detailfAI 抽取失败: {str(e)}) created_clauses [] for item in clauses: clause ComplianceClause( rule_idrule.id, clause_numberitem.get(clause_number, 0), original_textraw_text[:2000], ai_extracteditem, human_confirmedFalse, ) db.add(clause) db.flush() obligation ComplianceObligation( clause_idclause.id, titleitem.get(summary, )[:200], descriptionitem.get(required_action, ), required_actionitem.get(required_action, ), applicable_objectitem.get(applicable_object, ), risk_levelitem.get(risk_level, medium), due_typeitem.get(due_type, one_time), due_daysitem.get(due_days, 30), ) db.add(obligation) created_clauses.append(clause) db.commit() return {rule_id: rule.id, clause_count: len(created_clauses)} app.get(/clauses, response_modellist[ClauseOut]) def list_clauses(db: Session Depends(get_db)): return db.query(ComplianceClause).all() app.get(/obligations, response_modellist[ObligationOut]) def list_obligations(db: Session Depends(get_db)): return db.query(ComplianceObligation).all() app.post(/tasks, response_modelTaskOut) def create_task(task: TaskCreate, db: Session Depends(get_db)): db_task ComplianceTask(**task.model_dump()) db.add(db_task) db.commit() db.refresh(db_task) return db_task app.get(/tasks, response_modellist[TaskOut]) def list_tasks(db: Session Depends(get_db)): return db.query(ComplianceTask).all()这个 MVP 没有写复杂权限控制核心目的是走通“规则导入 → AI 提取 → 义务生成 → 任务创建”的完整链路。实际生产项目里需要把 import 接口改成异步任务并验证用户权限。5.8 运行与验证后端服务启动命令cd veritas-demo/backend pip install -r requirements.txt uvicorn app.main:app --reload --port 8000接着用 curl 模拟一次规则导入curl -X POST http://localhost:8000/rules/import \ -H Content-Type: application/json \ -d { rule_title: GDPR Data Retention Policy, rule_code: GDPR-ART-5, jurisdiction: EU, raw_text: Personal data shall be kept in a form which permits identification of data subjects for no longer than is necessary for the purposes for which the personal data are processed. }导入成功后查看义务列表curl http://localhost:8000/obligations预期返回一个 JSON 数组里面至少有一条risk_level和due_type字段。这些字段来自 AI 抽取结果。6. 常见问题与排查思路问题现象常见原因解决思路导入规则时提示 500AI 服务地址不可达或 API Key 错误检查AI_API_BASE和AI_API_KEY用 curl 测试模型接口连通性AI 返回内容无法解析成 JSON模型输出包含额外文字或 markdown 代码块在代码里做代码块清理或者调整 prompt 要求“只输出 JSON”数据库连接失败PostgreSQL 未启动或 DATABASE_URL 配置错误先运行docker compose up -d db再检查连接串任务创建失败缺少 obligation_id外键约束失败先查询/obligations获取合法的义务 ID数据库表结构变更后启动报错没有做迁移MVP 阶段可用Base.metadata.create_all生产环境建议引入 Alembic模型响应太慢上游模型处理长文本耗时MVP 阶段可以限制 raw_text 长度生产环境改用异步任务和消息队列排查时建议按“配置 → 网络 → 数据 → 代码”的顺序来。先确认环境变量是否正确再验证模型服务是否连通然后检查数据库数据最后看代码日志。7. 最佳实践与工程建议7.1 AI 输出必须有“人审”环节合规场景中AI 输出不能直接变成正式义务。最稳妥的做法是增加一个人工确认工作流AI 先产出草稿合规负责人确认后才生成任务。除了用户手动确认外还可以用置信度分数筛选出低置信度记录优先让人工复核。7.2 全文留痕与审计日志生产系统里每一次 AI 调用、每一条规则导入、每一次人工修改都应该写入审计日志。不仅要记录操作结果还要记录原始输入、模型输出、用户确认信息。这样在合规审计时才能说清楚“这条数据是怎么来的、谁改过、为什么会这样”。7.3 权限与数据隔离如果系统里同时管理多个公司或品牌的合规事务需要在设计阶段就考虑多租户数据隔离。最简单的方式是在所有核心表上增加tenant_id字段查询时强制过滤。不要把多租户隔离放到应用层之外否则很容易出现数据越权。7.4 任务调度和通知策略MVP 里没有写调度器但生产环境需要定时扫描即将到期的任务。建议通过 APScheduler 或者 Celery Beat 每天扫描一次到期前 7 天、3 天、1 天分别触发提醒。通知渠道除站内信外可以接入邮件、企业微信、钉钉等。7.5 安全与敏感信息合规业务经常会涉及企业内部数据。在开发阶段就要遵守最小权限原则数据库账号只授权需要的库表和权限。API 密钥不要写在代码里使用环境变量或密钥管理服务。AI 请求日志中不要明文记录大段敏感文本。对外接口必须做身份认证不能裸奔。7.6 数据库迁移Base.metadata.create_all只适合开发环境。项目要进入测试或生产一定要引入 Alembic 或类似工具管理表结构变更否则后续改动字段成本会非常高。8. 总结与下一步本文围绕 Veritas 项目梳理了 AI 驱动的合规追踪系统从需求到实现的完整链路。我们从合规追踪器的概念出发拆解了规则管理、义务管理、任务管理、AI 辅助抽取、通知和审计日志等模块设计了核心数据模型并提供一个可以本地运行的最小 MVP。通过这个项目至少可以体会到两点第一合规系统不是靠一堆法律文书的“资料库”就能解决问题关键是把文本变成可执行的任务。 第二AI 在这个场景里的角色是效率工具而不是决策者。必须有结构化输出、异常兜底、人工审核和完整日志才能安全地引入到生产环境。接下来你可以继续扩展的方向包括接入更稳定的异步任务队列、增加用户认证与 RBAC 权限、引入 Alembic 管理数据库迁移、接入真实的通知渠道以及把 AI 抽取结果做成可视化对照表方便人工快速确认。如果你正在规划类似的合规中台或 AI 辅助业务系统建议先从这个 MVP 跑起来再逐步迭代。现在就可以打开终端把代码下载下来试一遍亲手体验一次法规文本到任务清单的转化过程。 AI驱动合规追踪系统：从法规文本到任务清单的工程实践
创业公司在全球扩张时最容易被忽视却又最致命的问题往往不是产品功能而是合规。不同国家的数据保护法、劳动法、财税要求交织在一起人工追踪效率低、易遗漏等真正面对审计时才补文档已经来不及了。Veritas 就是一个面向全球初创公司的 AI 合规追踪器它把“合规”从一堆分散的 PDF 和表格里抽出来变成可分配、可追踪、可验证的任务流。这篇文章会以 Veritas 项目为原型完整拆解一个 AI 驱动的合规追踪系统应该怎么做。内容覆盖需求分析、系统设计、数据建模、AI 能力接入、最小可运行项目代码以及常见问题和工程建议。无论你是后端开发者还是刚接触 AI 应用开发的初学者都可以照着本文思路搭建一个自己的合规追踪 MVP再逐步扩展成生产级系统。1. 背景与核心概念1.1 什么是合规追踪器合规追踪器简单说就是一套用来管理“企业必须遵守哪些规则、需要完成哪些合规动作、每件事做到什么程度”的系统。它不是在帮企业规避法律而是帮企业把法律义务转变成具体的任务清单。传统做法是法务或运营负责人手工维护一张表格记录法规名称、适用范围、责任部门、到期时间、执行状态。对于早期项目可能够用但一旦业务扩展到多个地区规则数量会成倍增长。比如在欧盟运营要关注 GDPR在美国做消费者业务可能涉及 CCPA/CPRA在东南亚开展业务又各有本地隐私法。每一个法规又可能拆出十几条具体义务。合规追踪器要解决的核心问题有三个规则分散法律文本散落在不同政府网站和专业数据库中难以集中管理。责任不清一条合规义务没有明确的 owner最后往往没人跟进。时效难控哪些义务需要年度复审哪些需要 30 天内响应靠人记不现实。把这些问题系统化就是合规追踪器存在的价值。1.2 Veritas 项目的定位Veritas 的名字来源于拉丁语“真理”。这个项目的核心理念是用 AI 辅助人类处理海量法规文本但最终决策和审计责任仍然由人来承担。它不是要做成一个自动判断公司是否违法的“审判系统”而是做成一个“追踪提醒辅助理解”的平台。具体来说Veritas 会做以下几件事把导入的法规文本通过 AI 进行结构化抽取提取义务主体、动作、时间要求、处罚风险。将不同的义务自动分类到对应部门比如工程团队负责隐私保护、财务团队负责税务申报。根据义务的紧急性、影响范围、处罚金额自动计算风险等级。生成一条条可执行任务支持负责人分配、截止日期提醒、完成度跟踪。所有操作记录审计日志方便合规审计时出具证据链。这里要特别强调AI 在合规场景里只能做“辅助”不能替代专业判断。因为法规解读涉及到具体业务上下文同一个条款在不同行业、不同规模的公司里含义可能完全不同。所以 Veritas 的设计原则是“AI 建议人工确认”。1.3 为什么初创团队需要重点关注合规很多初创团队觉得合规是大公司的事自己还小先跑起来再说。但实际情况恰恰相反越早期越容易因为某个基础动作不规范在融资尽调或客户安全审计时被动。举一个很常见的场景初创公司做了一款 SaaS 工具拿到了第一个海外客户。客户发来一份安全合规问卷要求说明数据存储位置、保留策略、访问日志、信息安全标准。如果公司内部没有一套追踪机制问卷里的问题很难回答甚至可能因为答不上来而丢掉订单。Veritas 这类系统在早期阶段的价值不是用来自动完成合规认证而是让团队始终保持“知道自己该做什么”的状态。这正是技术工程中最有价值的部分把模糊的法律条款变成清晰的项目管理动作。2. 系统设计与功能拆分2.1 需求分析在动手写代码之前先梳理一下系统需要哪些角色和流程。Veritas 的最小可用版本需要支持三类角色管理员负责维护合规规则库、配置 AI 模型、管理用户。合规负责人负责导入法规、审核 AI 抽取结果、分派任务。团队成员查看分配给自己的义务提交完成证明更新状态。核心业务流如下合规负责人导入一份法规文本可以是 PDF、Word 或纯文本。后端调用 AI 服务把文本拆成结构化条款。AI 对每条条款进行归类识别出“义务动作”“适用对象”“时间期限”“风险等级”。合规负责人逐条确认或修改 AI 结果。确认后的数据生成一条义务记录并自动派生一个或多个任务。任务分配负责人后系统在截止日期前发送提醒。负责人完成任务后系统记录完成人、完成时间、证明材料。所有关键操作写入审计日志。2.2 模块拆分按照上面的流程Veritas 可以拆成以下模块模块职责规则管理法规文本上传、版本管理、条款结构维护AI 抽取服务调用大模型提取结构化信息义务管理管理合规义务维护风险等级、责任方、到期时间任务管理任务的创建、分配、完成、延期提醒通知按截止时间发送邮件或站内通知审计日志记录用户操作和 AI 行为支持追溯用户与权限角色管理控制不同用户可见的数据范围为了控制 MVP 的复杂度暂时不做复杂的权限矩阵只实现管理员和普通用户两个角色。2.3 技术栈选型对于 MVP我推荐以下技术栈后端框架Python FastAPI开发效率高天然支持异步自带 OpenAPI 文档。数据库PostgreSQL事务支持好适合业务系统也可以先用 SQLite 快速验证。ORMSQLAlchemy 2.0兼容主流数据库。任务调度APScheduler负责循环检查到期任务并触发通知。AI 服务OpenAI 兼容接口也可以用本地模型或各类 API 网关。本文代码以兼容接口为例模型只要支持对话补全即可。部署Docker Compose方便本地启动 PostgreSQL 和 Redis。版本方面Python 建议 3.10 以上FastAPI 使用最新稳定版即可。因为依赖版本更新较快本文示例以常规写法为主实际运行时按你的环境锁定版本。3. 数据模型设计数据模型是整个系统的地基。Veritas 的核心实体包括用户、合规规则、合规义务、任务、通知日志和审计日志。3.1 核心表结构我们先用自然语言描述再转成 SQLAlchemy 模型。用户表id、用户名、邮箱、角色、创建时间。合规规则表法规名称、法规编号、发布日期、原文存储路径、当前版本。条款表属于哪个规则条款编号原始文本AI 抽取结果人工确认状态。义务表从条款中提炼出来的具体义务包含适用对象、动作描述、截止类型、风险等级、负责人。任务表某个义务派生的执行任务包含截止日期、状态、负责人、完成时间。通知日志表记录每次提醒通知的类型、接收人、发送时间、关联任务。审计日志表记录操作者、操作类型、操作详情、IP、时间。3.2 SQLAlchemy 模型示例下面是一份简化后的模型代码。文件路径backend/app/models.py。# backend/app/models.py from datetime import datetime from sqlalchemy import ( String, Integer, Boolean, DateTime, Text, ForeignKey, Enum, Float, JSON, LargeBinary ) from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.dialects.postgresql import UUID import uuid class Base: id: Mapped[str] mapped_column( UUID(as_uuidTrue), primary_keyTrue, defaultuuid.uuid4 ) created_at: Mapped[datetime] mapped_column( DateTime, defaultdatetime.utcnow ) class User(Base): __tablename__ users username: Mapped[str] mapped_column(String(50), uniqueTrue, indexTrue) email: Mapped[str] mapped_column(String(120), uniqueTrue, indexTrue) role: Mapped[str] mapped_column(String(20), defaultuser) is_active: Mapped[bool] mapped_column(Boolean, defaultTrue) class ComplianceRule(Base): __tablename__ compliance_rules title: Mapped[str] mapped_column(String(200)) rule_code: Mapped[str] mapped_column(String(100), uniqueTrue, indexTrue) jurisdiction: Mapped[str] mapped_column(String(50), indexTrue) published_date: Mapped[datetime] mapped_column(DateTime, nullableTrue) source_url: Mapped[str] mapped_column(String(500), default) version: Mapped[str] mapped_column(String(20), default1.0) class ComplianceClause(Base): __tablename__ compliance_clauses rule_id: Mapped[str] mapped_column(ForeignKey(compliance_rules.id)) clause_number: Mapped[str] mapped_column(String(50)) original_text: Mapped[str] mapped_column(Text) ai_extracted: Mapped[dict] mapped_column(JSON, defaultdict) human_confirmed: Mapped[bool] mapped_column(Boolean, defaultFalse) class ComplianceObligation(Base): __tablename__ compliance_obligations clause_id: Mapped[str] mapped_column(ForeignKey(compliance_clauses.id)) title: Mapped[str] mapped_column(String(200)) description: Mapped[str] mapped_column(Text) required_action: Mapped[str] mapped_column(String(200)) applicable_object: Mapped[str] mapped_column(String(200)) risk_level: Mapped[str] mapped_column(String(20), defaultmedium) due_type: Mapped[str] mapped_column(String(20), defaultone_time) due_days: Mapped[int] mapped_column(Integer, default30) owner_id: Mapped[str] mapped_column(ForeignKey(users.id), nullableTrue) class ComplianceTask(Base): __tablename__ compliance_tasks obligation_id: Mapped[str] mapped_column(ForeignKey(compliance_obligations.id)) title: Mapped[str] mapped_column(String(200)) assignee_id: Mapped[str] mapped_column(ForeignKey(users.id), nullableTrue) due_date: Mapped[datetime] mapped_column(DateTime) status: Mapped[str] mapped_column(String(20), defaultpending) completed_at: Mapped[datetime] mapped_column(DateTime, nullableTrue) evidence: Mapped[str] mapped_column(Text, default) class NotificationLog(Base): __tablename__ notification_logs task_id: Mapped[str] mapped_column(ForeignKey(compliance_tasks.id)) receiver_email: Mapped[str] mapped_column(String(120)) notification_type: Mapped[str] mapped_column(String(30)) sent_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class AuditLog(Base): __tablename__ audit_logs user_id: Mapped[str] mapped_column(ForeignKey(users.id), nullableTrue) action: Mapped[str] mapped_column(String(50)) entity_type: Mapped[str] mapped_column(String(50)) entity_id: Mapped[str] mapped_column(String(50)) detail: Mapped[dict] mapped_column(JSON, defaultdict) ip_address: Mapped[str] mapped_column(String(50), default)这个模型覆盖了核心业务链路。其中ai_extracted字段用 JSON 类型保存大模型抽取出的结构化结果方便后期字段演进不用频繁改表结构。4. AI 辅助能力拆解合规追踪器接入 AI 的好处不是让 AI 替人做决定而是把“读法规、找要点、初步归类”这种重复劳动自动化。下面我们拆解四个核心 AI 能力。4.1 法规文本结构化抽取原始法规是一整篇长文本AI 要做的事情是把它按条款拆分并提取关键信息。一个典型的 prompt 结构如下你是合规分析助手。请从提供的法规文本中提取条款列表。 每个条款需要包含 - clause_number: 条款编号 - summary: 条款摘要 - required_action: 条款要求企业采取的动作 - applicable_object: 适用的业务对象如 网站、移动应用、员工数据 - due_type: 义务类型one_time 一次性 / recurring 周期性 / event_driven 事件驱动 - due_days: 合规动作应在多少天内完成 - risk_level: high/medium/low 请只输出 JSON 数组不要包含其他文字。这里的关键是给模型一个非常明确的“输出格式”。否则模型会生成一大段解释不利于下游解析。4.2 义务自动分类抽取出的条款很多直接把每条都变成任务会给团队带来负担。所以 Veritas 会让 AI 对义务做一次“初步分类”比如工程安全类数据隐私类财务会计类人力资源类商业模式合规类分类合理之后系统可以自动把任务分配给对应的业务线。4.3 风险等级评估风险等级不能完全依赖模型拍脑袋。我们在 prompt 里可以要求模型基于三个维度打分违反后可能的罚款金额监管关注程度对业务运营的影响程度再让模型输出一个综合等级。不过风险等级更适合作为“建议”后续由人工复核。4.4 生成执行建议AI 还可以根据义务内容生成建议步骤。比如某条义务要求“用户删除账户后 30 天内彻底清除数据”AI 可以建议工程团队设计定时任务、数据库级联删除逻辑、以及删除结果留存证明。这部分输出适合放在义务详情页让负责人知道“下一步该做什么”。4.5 AI 能力实现注意事项使用大模型处理合规文本时有几点必须注意不要上传不必要的敏感数据生产环境建议对接可信的私有化模型或合规数据网关。prompt 中必须要求模型输出 JSON并且在代码里做异常兜底。所有 AI 输出必须留痕和原始条款文本绑定方便溯源。AI 抽取结果只能作为草稿必须有一个人工确认的环节。下面给出一个简单的 AI 客户端示例使用 requests 调用 OpenAI 兼容接口。# backend/app/ai_client.py import json import requests class AIClient: def __init__(self, api_base, api_key, model): self.api_base api_base.rstrip(/) self.api_key api_key self.model model def analyze_text(self, text: str) - list: prompt 你是合规分析助手。请从法规文本中提取条款列表。 每个条款需要包含clause_number, summary, required_action, applicable_object, due_type(one_time/recurring/event_driven), due_days, risk_level(high/medium/low)。 请只输出 JSON 数组不要包含其他文字。 payload { model: self.model, messages: [ {role: system, content: prompt}, {role: user, content: text[:8000]} ], temperature: 0.2 } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } resp requests.post( f{self.api_base}/chat/completions, headersheaders, jsonpayload, timeout60 ) resp.raise_for_status() content resp.json()[choices][0][message][content] # 解析模型返回的 JSON try: return json.loads(content) except json.JSONDecodeError: # 如果模型输出包含 json 代码块做一次清理 cleaned content.strip().removeprefix(json).removesuffix() return json.loads(cleaned)这段代码不绑定某个特定厂商只要你的模型服务提供/chat/completions接口就能用。5. 完整实战案例最小可运行 MVP接下来我们实现一个可运行的 MVP。为了不让文章过长这里只实现核心链路导入法规文本 → AI 抽取 → 任务创建 → 查询任务列表。5.1 项目结构veritas-demo/ ├── backend/ │ ├── app/ │ │ ├── __init__.py │ │ ├── main.py │ │ ├── config.py │ │ ├── database.py │ │ ├── models.py │ │ ├── schemas.py │ │ └── ai_client.py │ └── requirements.txt ├── docker-compose.yml └── README.md5.2 requirements.txtfastapi0.111.0 uvicorn[standard]0.30.1 sqlalchemy2.0.30 psycopg2-binary2.9.9 python-dotenv1.0.1 requests2.32.3 pydantic2.7.4如果你本地还没有 PostgreSQL可以直接使用 Docker Compose 启动一个。5.3 docker-compose.ymlversion: 3.9 services: db: image: postgres:15 container_name: veritas_db environment: POSTGRES_USER: veritas POSTGRES_PASSWORD: veritas POSTGRES_DB: veritas ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data redis: image: redis:7 container_name: veritas_redis ports: - 6379:6379 volumes: pgdata:在真实项目中 Redis 可以用于缓存和异步任务队列MVP 阶段暂时不接入但先预留。5.4 配置文件# backend/app/config.py import os from dotenv import load_dotenv load_dotenv() class Settings: DATABASE_URL os.getenv( DATABASE_URL, postgresqlpsycopg2://veritas:veritaslocalhost:5432/veritas ) AI_API_BASE os.getenv(AI_API_BASE, https://your-ai-gateway.example.com) AI_API_KEY os.getenv(AI_API_KEY, your-api-key) AI_MODEL os.getenv(AI_MODEL, compliance-assistant) settings Settings()这里要注意AI_API_BASE只是一个示例占位地址实际部署时请改成你自己的模型服务地址。不要把你自己的真实密钥提交到代码库。5.5 数据库连接# backend/app/database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase from .config import settings engine create_engine(settings.DATABASE_URL, echoFalse) SessionLocal sessionmaker(bindengine, autocommitFalse, autoflushFalse) class Base(DeclarativeBase): pass5.6 Pydantic 模型# backend/app/schemas.py from pydantic import BaseModel, Field from datetime import datetime class ClauseCreate(BaseModel): rule_id: str clause_number: str original_text: str class ClauseOut(ClauseCreate): id: str ai_extracted: dict human_confirmed: bool class Config: from_attributes True class ObligationOut(BaseModel): id: str title: str risk_level: str owner_id: str | None due_type: str due_days: int class TaskCreate(BaseModel): obligation_id: str title: str assignee_id: str | None None due_date: datetime class TaskOut(TaskCreate): id: str status: str completed_at: datetime | None None class Config: from_attributes True5.7 FastAPI 主程序# backend/app/main.py from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from .database import engine, SessionLocal, Base from .models import ( User, ComplianceRule, ComplianceClause, ComplianceObligation, ComplianceTask ) from .schemas import ( TaskCreate, TaskOut, ObligationOut, ClauseOut ) from .ai_client import AIClient from .config import settings Base.metadata.create_all(bindengine) app FastAPI(titleVeritas Compliance Tracker, version0.1.0) def get_db(): db SessionLocal() try: yield db finally: db.close() app.post(/rules/import) def import_rule(rule_title: str, rule_code: str, jurisdiction: str, raw_text: str, db: Session Depends(get_db)): 模拟一个简单的规则导入接口。 实际项目中应该上传文件并解析文本这里为了演示直接传入原文。 rule ComplianceRule( titlerule_title, rule_coderule_code, jurisdictionjurisdiction, ) db.add(rule) db.flush() client AIClient( api_basesettings.AI_API_BASE, api_keysettings.AI_API_KEY, modelsettings.AI_MODEL ) try: clauses client.analyze_text(raw_text) except Exception as e: raise HTTPException(status_code500, detailfAI 抽取失败: {str(e)}) created_clauses [] for item in clauses: clause ComplianceClause( rule_idrule.id, clause_numberitem.get(clause_number, 0), original_textraw_text[:2000], ai_extracteditem, human_confirmedFalse, ) db.add(clause) db.flush() obligation ComplianceObligation( clause_idclause.id, titleitem.get(summary, )[:200], descriptionitem.get(required_action, ), required_actionitem.get(required_action, ), applicable_objectitem.get(applicable_object, ), risk_levelitem.get(risk_level, medium), due_typeitem.get(due_type, one_time), due_daysitem.get(due_days, 30), ) db.add(obligation) created_clauses.append(clause) db.commit() return {rule_id: rule.id, clause_count: len(created_clauses)} app.get(/clauses, response_modellist[ClauseOut]) def list_clauses(db: Session Depends(get_db)): return db.query(ComplianceClause).all() app.get(/obligations, response_modellist[ObligationOut]) def list_obligations(db: Session Depends(get_db)): return db.query(ComplianceObligation).all() app.post(/tasks, response_modelTaskOut) def create_task(task: TaskCreate, db: Session Depends(get_db)): db_task ComplianceTask(**task.model_dump()) db.add(db_task) db.commit() db.refresh(db_task) return db_task app.get(/tasks, response_modellist[TaskOut]) def list_tasks(db: Session Depends(get_db)): return db.query(ComplianceTask).all()这个 MVP 没有写复杂权限控制核心目的是走通“规则导入 → AI 提取 → 义务生成 → 任务创建”的完整链路。实际生产项目里需要把 import 接口改成异步任务并验证用户权限。5.8 运行与验证后端服务启动命令cd veritas-demo/backend pip install -r requirements.txt uvicorn app.main:app --reload --port 8000接着用 curl 模拟一次规则导入curl -X POST http://localhost:8000/rules/import \ -H Content-Type: application/json \ -d { rule_title: GDPR Data Retention Policy, rule_code: GDPR-ART-5, jurisdiction: EU, raw_text: Personal data shall be kept in a form which permits identification of data subjects for no longer than is necessary for the purposes for which the personal data are processed. }导入成功后查看义务列表curl http://localhost:8000/obligations预期返回一个 JSON 数组里面至少有一条risk_level和due_type字段。这些字段来自 AI 抽取结果。6. 常见问题与排查思路问题现象常见原因解决思路导入规则时提示 500AI 服务地址不可达或 API Key 错误检查AI_API_BASE和AI_API_KEY用 curl 测试模型接口连通性AI 返回内容无法解析成 JSON模型输出包含额外文字或 markdown 代码块在代码里做代码块清理或者调整 prompt 要求“只输出 JSON”数据库连接失败PostgreSQL 未启动或 DATABASE_URL 配置错误先运行docker compose up -d db再检查连接串任务创建失败缺少 obligation_id外键约束失败先查询/obligations获取合法的义务 ID数据库表结构变更后启动报错没有做迁移MVP 阶段可用Base.metadata.create_all生产环境建议引入 Alembic模型响应太慢上游模型处理长文本耗时MVP 阶段可以限制 raw_text 长度生产环境改用异步任务和消息队列排查时建议按“配置 → 网络 → 数据 → 代码”的顺序来。先确认环境变量是否正确再验证模型服务是否连通然后检查数据库数据最后看代码日志。7. 最佳实践与工程建议7.1 AI 输出必须有“人审”环节合规场景中AI 输出不能直接变成正式义务。最稳妥的做法是增加一个人工确认工作流AI 先产出草稿合规负责人确认后才生成任务。除了用户手动确认外还可以用置信度分数筛选出低置信度记录优先让人工复核。7.2 全文留痕与审计日志生产系统里每一次 AI 调用、每一条规则导入、每一次人工修改都应该写入审计日志。不仅要记录操作结果还要记录原始输入、模型输出、用户确认信息。这样在合规审计时才能说清楚“这条数据是怎么来的、谁改过、为什么会这样”。7.3 权限与数据隔离如果系统里同时管理多个公司或品牌的合规事务需要在设计阶段就考虑多租户数据隔离。最简单的方式是在所有核心表上增加tenant_id字段查询时强制过滤。不要把多租户隔离放到应用层之外否则很容易出现数据越权。7.4 任务调度和通知策略MVP 里没有写调度器但生产环境需要定时扫描即将到期的任务。建议通过 APScheduler 或者 Celery Beat 每天扫描一次到期前 7 天、3 天、1 天分别触发提醒。通知渠道除站内信外可以接入邮件、企业微信、钉钉等。7.5 安全与敏感信息合规业务经常会涉及企业内部数据。在开发阶段就要遵守最小权限原则数据库账号只授权需要的库表和权限。API 密钥不要写在代码里使用环境变量或密钥管理服务。AI 请求日志中不要明文记录大段敏感文本。对外接口必须做身份认证不能裸奔。7.6 数据库迁移Base.metadata.create_all只适合开发环境。项目要进入测试或生产一定要引入 Alembic 或类似工具管理表结构变更否则后续改动字段成本会非常高。8. 总结与下一步本文围绕 Veritas 项目梳理了 AI 驱动的合规追踪系统从需求到实现的完整链路。我们从合规追踪器的概念出发拆解了规则管理、义务管理、任务管理、AI 辅助抽取、通知和审计日志等模块设计了核心数据模型并提供一个可以本地运行的最小 MVP。通过这个项目至少可以体会到两点第一合规系统不是靠一堆法律文书的“资料库”就能解决问题关键是把文本变成可执行的任务。 第二AI 在这个场景里的角色是效率工具而不是决策者。必须有结构化输出、异常兜底、人工审核和完整日志才能安全地引入到生产环境。接下来你可以继续扩展的方向包括接入更稳定的异步任务队列、增加用户认证与 RBAC 权限、引入 Alembic 管理数据库迁移、接入真实的通知渠道以及把 AI 抽取结果做成可视化对照表方便人工快速确认。如果你正在规划类似的合规中台或 AI 辅助业务系统建议先从这个 MVP 跑起来再逐步迭代。现在就可以打开终端把代码下载下来试一遍亲手体验一次法规文本到任务清单的转化过程。