做国际化产品的第一步不是把功能上线而是先搞清楚产品在哪些市场受哪些规则约束。很多初创团队会把“合规”理解成找律师审一遍合同或者等到融资尽调才临时补文档但在实际运营里GDPR 的数据处理记录、CCPA 的消费者删除请求、SOC 2 的审计证据、出口管制产品清单每一项都对应明确的负责人和截止日期。Veritas 就是为这类场景设计的 AI 驱动的合规追踪器目标是让全球初创公司在不同法域下把散落在法规文本、合同条款和内部邮件里的合规义务自动整理成可追踪、可提醒、可审计的任务。这套系统听起来像一个“待办事项工具”但真正实现起来要比普通待办复杂不少。合规文本不是结构化数据法律条款中的义务主体、适用实体、履行期限和验证标准经常混在一段长文本里同一件事在不同法域可能又有不同定义。下面我会从核心设计、环境准备、代码实现到生产落地完整拆解一个可运行的 Veritas 最小系统。你可以把它当作学习项目逐步搭建也可以把它当成自建合规平台的初始骨架。1. 为什么全球初创公司需要 AI 驱动的合规追踪器1.1 合规催办不只是一个待办清单普通任务管理工具的核心对象是“某个员工要做某件事”合规追踪的核心对象是“某个法律义务是否被覆盖并落实”。这两者最大的区别在于合规义务会随着法规版本、业务范围、公司实体所在地的变化而变化而且每一件合规事项都必须能说清楚依据是什么、谁负责、什么时候之前必须完成。例如一家面向欧盟用户提供云服务的初创公司可能需要同时追踪 GDPR 下的数据处理记录义务、Cookie 同意机制、数据泄露通知时限以及客户合同中关于数据处理者的条款。这些事项分布在多个文档里靠 Excel 表格手工维护时常常出现三类问题责任人离职后事项无人接手法规更新后旧任务没有重跑审计时拿不出证据链。Veritas 要解决的正是“合规文本到可执行任务”之间的转换和持续追踪问题。1.2 传统工具在跨法域场景下的难点面向全球初创公司的合规追踪比单一国家内部合规更麻烦。每一个目标市场都有自己的监管机构和语言甚至同一个法律术语在不同法域的含义也不同。传统工具大致有几类但都有明显短板工具类型代表方式主要短板电子表格Excel、Google Sheets缺少任务状态、责任人、过期时间之间的强关系通用项目管理工具Jira、Trello只适合管理工程任务缺少法规依据和证据字段合同管理系统共享网盘、合同库能存档但不能从条款里自动提取义务外部合规咨询律师事务所、咨询顾问准确但成本高难以覆盖所有法域的日常更新真正的合规追踪应同时具备三个能力从非结构化文本中抽取结构化义务把义务转化为带状态和截止日期的任务在法规更新时能重新评估旧任务是否仍然成立。传统工具无法自动完成第一项和第三项所以“AI 驱动”在这里不是噱头而是解决核心问题的技术路径。1.3 Veritas 的系统定位和技术目标Veritas 不是要替代律师而是要做“合规运营的中控台”。律师或者合规负责人可以把法规原文、监管通知、内部政策文档导入系统系统借助 AI 生成义务草稿再由人工审核确认最后按照组织归属生成任务。技术目标可以拆成四条文本抽取从法规和合同条款中识别义务类型、适用实体、截止日期、风险等级。职责闭环每个义务都有负责人、状态、截止日期和审计记录。规则与模型结合AI 处理灵活文本规则引擎处理日期、关键词和已知模板。可观测所有抽取结果都有置信度所有状态变更都有日志。这套设计不是把全部判断交给模型而是让人工审核成为必要环节。后面章节的代码都会围绕这些目标展开。2. 核心设计如何把合规文本变成可追踪的任务2.1 合规数据的结构化模型在写代码之前先定义清楚数据边界。合规追踪系统最核心的表不是“任务表”而是“义务表”。一个合规义务是从某份文档中抽取出来的、对某个实体具有约束力的待办事项例如“在 72 小时内向监管机构报告个人数据泄露”。为了保持模型简洁又能支撑业务我建议至少设计四张表jurisdictions法域表记录国家或地区、监管机构。compliance_documents合规文档表保存原文和来源信息。compliance_obligations义务表保存抽取出的结构化义务。compliance_tasks任务表把义务分配给具体责任人并跟踪执行状态。义务表是 AI 抽取的产物任务表是人工协同的结果。不要把 AI 抽取结果直接当成任务否则一旦抽取错误追责和修正会非常困难。人工审核应该发生在“义务”和“任务”之间。2.2 规则引擎加 LLM 的混合抽取思路合规文本类型很多有的是结构化的法规条文有的是长合同条款有的甚至是一封监管邮件。LLM 很擅长理解自然语言但直接让 LLM 输出 JSON 并插入数据库存在幻觉和格式不稳定问题。更稳妥的做法是“规则引擎兜底LLM 做语义补充”。规则引擎负责两类事情日期识别例如“30日内”“within 30 days”“72 hours”转换为具体截止时间。关键词触发出现“shall”“必须”“应当”“is required to”时初步判断这里存在义务。LLM 负责更难的部分判断义务主体是谁、适用条件是什么、风险等级如何。最终结果通过 Pydantic Schema 校验只有通过校验的数据才能进入数据库。这样即使 LLM 输出不稳定数据库层面也不会写入格式错误的数据。2.3 任务状态机和风险评分模型义务和任务都要有明确状态但状态不宜在数据库里散落一地。建议义务状态使用以下流转draftAI 刚抽取等待人工审核。active人工审核通过已生成对应任务。suppressed人工判断不适用或错误。closed义务已经完成或失效。任务状态相对独立可以使用todo、in_progress、completed、cancelled。过期是一个计算字段而不是状态。判断逻辑是任务过期 当前时间 due_date AND status ! completed AND status ! cancelled风险评分建议使用 0 到 1 的小数。0.8 以上视为高风险0.4 到 0.8 为中风险0.4 以下为低风险。评分不是最终结论而是用于排序和提醒优先级。AI 给出的risk_score必须有人工复核入口。3. 环境准备与最小项目骨架3.1 技术选型与版本对齐实现 Veritas 最小系统我选用以下技术栈核心原则是结构清晰、便于替换 AI 服务Python 3.11FastAPISQLAlchemy 2.0PostgreSQLPydantic v2httpxpytest这里不需要把 AI 服务绑定在某一家厂商上。通过环境变量配置 OpenAI-compatible 的接口地址既可以接云端模型也可以接本地 Ollama。关键依赖版本如下fastapi0.115.6 uvicorn[standard]0.30.6 sqlalchemy2.0.36 psycopg[binary]3.2.3 pydantic2.9.2 pydantic-settings2.6.0 httpx0.27.2 python-dotenv1.0.1 pytest8.3.4实际项目落地前要确认这些版本已经兼容不要把版本号当作长期固定值。如果依赖冲突优先以 SQLAlchemy 和 FastAPI 的官方兼容说明为准。3.2 初始化项目结构和依赖项目目录建议按服务边界拆分避免把所有代码堆在一个文件里compliance_tracker/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── database.py │ ├── models.py │ ├── schemas.py │ ├── services/ │ │ ├── __init__.py │ │ ├── extractor.py │ │ └── rules.py │ └── routers/ │ ├── __init__.py │ ├── documents.py │ └── dashboard.py ├── tests/ ├── docker-compose.yml └── requirements.txt本地使用 Docker Compose 启动 PostgreSQL 最省事version: 3.9 services: postgres: image: postgres:15-alpine container_name: veritas-postgres environment: POSTGRES_USER: veritas POSTGRES_PASSWORD: veritas POSTGRES_DB: veritas ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata:启动数据库docker compose up -d这里要注意本地开发用这个默认密码没问题但生产环境必须通过环境变量或 Secret 管理工具注入凭证不能写在 Docker Compose 文件里。3.3 数据库模型和迁移SQLAlchemy 2.0 的声明式模型可以把 XML 映射彻底藏起来代码更直观。下面是核心模型的简化版本from datetime import datetime from sqlalchemy import String, Text, DateTime, Float, ForeignKey from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class Jurisdiction(Base): __tablename__ jurisdictions id: Mapped[int] mapped_column(primary_keyTrue) code: Mapped[str] mapped_column(String(16), uniqueTrue) name: Mapped[str] mapped_column(String(255)) regulatory_body: Mapped[str | None] mapped_column(String(255), nullableTrue) documents: Mapped[list[ComplianceDocument]] relationship(back_populatesjurisdiction) class ComplianceDocument(Base): __tablename__ compliance_documents id: Mapped[int] mapped_column(primary_keyTrue) jurisdiction_id: Mapped[int | None] mapped_column(ForeignKey(jurisdictions.id), nullableTrue) title: Mapped[str] mapped_column(String(500)) content_text: Mapped[str] mapped_column(Text) document_type: Mapped[str] mapped_column(String(64), defaultregulation) source_url: Mapped[str | None] mapped_column(String(1000), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) jurisdiction: Mapped[Jurisdiction | None] relationship(back_populatesdocuments) obligations: Mapped[list[ComplianceObligation]] relationship(back_populatesdocument)义务表是关键一定要把“从哪里来”和“归谁管”都记录清楚class ComplianceObligation(Base): __tablename__ compliance_obligations id: Mapped[int] mapped_column(primary_keyTrue) document_id: Mapped[int] mapped_column(ForeignKey(compliance_documents.id)) obligation_type: Mapped[str] mapped_column(String(128)) summary: Mapped[str] mapped_column(Text) regulation_reference: Mapped[str | None] mapped_column(String(500), nullableTrue) applicable_entity: Mapped[str | None] mapped_column(String(255), nullableTrue) due_date: Mapped[datetime | None] mapped_column(DateTime, nullableTrue) risk_score: Mapped[float] mapped_column(Float, default0.5) confidence_score: Mapped[float] mapped_column(Float, default0.0) status: Mapped[str] mapped_column(String(32), defaultdraft) owner: Mapped[str | None] mapped_column(String(255), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) updated_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) document: Mapped[ComplianceDocument] relationship(back_populatesobligations) tasks: Mapped[list[ComplianceTask]] relationship(back_populatesobligation)这里没有用 Alembic因为最小系统可以直接使用Base.metadata.create_all。实际项目建议从第一天就引入 Alembic否则数据库字段变更时没法做可回滚迁移。4. 实现核心模块AI 抽取、任务跟踪、仪表盘 API4.1 依赖注入与配置管理配置要做到“代码里不出现密钥”。使用pydantic-settings读取环境变量配置对象如下from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str postgresqlpsycopg://veritas:veritaslocalhost:5432/veritas llm_api_key: str llm_base_url: str https://api.openai.com/v1 llm_model: str gpt-4o-mini llm_timeout_seconds: float 30.0 default_owner: str complianceexample.com class Config: env_file .env env_prefix VERITAS_ settings Settings()env_prefix VERITAS_意味着本地环境变量使用VERITAS_DATABASE_URL、VERITAS_LLM_API_KEY等。如果设置了.env文件也要采用相同前缀。FastAPI 的依赖注入可以通过Depends获取数据库会话from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from fastapi import Depends engine create_engine(settings.database_url, pool_pre_pingTrue) SessionLocal sessionmaker(bindengine, autoflushFalse) def get_db(): db SessionLocal() try: yield db finally: db.close()4.2 实现 LLM 抽取服务与 JSON 校验LLM 抽取服务要承担两件事拼装提示词、解析并校验模型输出。先定义 Pydantic 输出结构from pydantic import BaseModel, Field class ExtractedObligation(BaseModel): obligation_type: str Field(description义务类型例如 data_protection、consumer_rights) summary: str Field(description义务内容摘要) regulation_reference: str | None Field(defaultNone, description法规或条款编号) applicable_entity: str | None Field(defaultNone, description适用实体例如 EEA 用户数据处理者) due_date: str | None Field(defaultNone, description如文本中有明确时间给出 ISO 格式日期) risk_score: float Field(default0.5, ge0, le1) confidence_score: float Field(default0.6, ge0, le1)抽取服务使用httpx请求 OpenAI-compatible 接口。为了方便本地测试这里把 base_url 设计成可以指向 Ollama 的http://localhost:11434/v1import json import httpx from pydantic import ValidationError from app.config import settings from app.schemas import ExtractedObligation class LLMExtractor: def __init__(self, client: httpx.AsyncClient | None None): self.client client async def _complete(self, prompt: str, system_prompt: str) - str: headers { Authorization: fBearer {settings.llm_api_key}, Content-Type: application/json, } payload { model: settings.llm_model, messages: [ {role: system, content: system_prompt}, {role: user, content: prompt}, ], temperature: 0.2, response_format: {type: json_object}, } url f{settings.llm_base_url}/chat/completions async with self.client or httpx.AsyncClient(timeoutsettings.llm_timeout_seconds) as client: response await client.post(url, headersheaders, jsonpayload) response.raise_for_status() data response.json() return data[choices][0][message][content] async def extract_many(self, text: str) - list[ExtractedObligation]: system_prompt ( 你是一个合规文本分析助手。请从用户提供的文本中抽取合规义务。 所有输出必须是合法的 JSON 对象格式为 {\obligations\: [...]}。 不要添加任何解释文字。 ) prompt f请从以下文本中抽取合规义务\n{text[:8000]} raw await self._complete(prompt, system_prompt) try: payload json.loads(raw) obligations [ExtractedObligation(**item) for item in payload[obligations]] return obligations except (json.JSONDecodeError, KeyError, ValidationError) as exc: raise ValueError(fLLM 输出无法解析为合规义务列表: {exc}) from exc这里有几个工程点必须说明温度设置 0.2 是为了让输出更稳定合规场景不需要创造性表达。response_format是 OpenAI-compatible 接口的通用参数但并不是所有本地模型都支持。如果本地模型不支持就要在提示词里强制要求 JSON 输出并做多一次格式修复。文本截取 8000 字符只是为了示例。实际项目应该按章节切片而不是直接截断避免丢失关键义务。4.3 实现规则兜底和日期识别如果 LLM 服务不可用或者模型输出为空规则引擎仍然要能给出一个“待人工审核”的草稿。下面是一个最小规则函数import re from datetime import datetime, timedelta def apply_rules(text: str) - list[dict]: results [] keyword_pattern re.compile(r(shall|must|应当|必须|应在|required to), re.IGNORECASE) for match in keyword_pattern.finditer(text): start max(0, match.start() - 120) end min(len(text), match.end() 180) snippet text[start:end] due_date find_due_date(snippet) results.append({ obligation_type: general_obligation, summary: snippet.strip(), regulation_reference: None, applicable_entity: None, due_date: due_date.isoformat() if due_date else None, risk_score: 0.5, confidence_score: 0.3, }) return results def find_due_date(text: str) - datetime | None: patterns [ (r(\d)\s*days?, timedelta(days1)), (r(\d)\s*hours?, timedelta(hours1)), (r(\d)\s*weeks?, timedelta(weeks1)), ] for pattern, unit in patterns: match re.search(pattern, text, re.IGNORECASE) if match: return datetime.utcnow() int(match.group(1)) * unit return None规则结果置信度设置为 0.3意味着“仅供参考”。真正使用时要让前端把这个草稿标记为“待人工校验”不能直接生成任务。4.4 实现文档导入、义务审核和仪表盘 APIFastAPI 路由可以分成三个模块但为了便于阅读这里给出核心方法。文档导入接口保存原文from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceDocument from app.schemas import DocumentCreate router APIRouter(prefix/api/v1/documents, tags[documents]) router.post() def create_document(payload: DocumentCreate, db: Session Depends(get_db)): doc ComplianceDocument(**payload.model_dump()) db.add(doc) db.commit() db.refresh(doc) return {id: doc.id, title: doc.title, created_at: doc.created_at}触发 AI 抽取的接口需要异步处理。最简实现是直接await抽取但生产环境建议把任务提交到 Celery 或 Arq避免接口请求超时from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceDocument, ComplianceObligation from app.services.extractor import LLMExtractor from app.services.rules import apply_rules router APIRouter(prefix/api/v1/documents, tags[documents]) router.post(/{document_id}/extract) async def extract_obligations(document_id: int, db: Session Depends(get_db)): document db.get(ComplianceDocument, document_id) if not document: raise HTTPException(status_code404, detaildocument not found) extractor LLMExtractor() obligations [] try: obligations await extractor.extract_many(document.content_text) except Exception: obligations apply_rules(document.content_text) created [] for item in obligations: obligation ComplianceObligation( document_iddocument.id, obligation_typeitem.obligation_type, summaryitem.summary, regulation_referenceitem.regulation_reference, applicable_entityitem.applicable_entity, due_dateitem.due_date, risk_scoreitem.risk_score, confidence_scoreitem.confidence_score, statusdraft, ) db.add(obligation) created.append(obligation) db.commit() return {created: len(created), source: llm if created else rule}人工审核接口。这里的关键是在审核通过时同步生成任务from datetime import datetime from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceObligation, ComplianceTask router APIRouter(prefix/api/v1/obligations, tags[obligations]) router.post(/{obligation_id}/approve) def approve_obligation(obligation_id: int, owner: str, db: Session Depends(get_db)): obligation db.get(ComplianceObligation, obligation_id) if not obligation: raise HTTPException(status_code404, detailobligation not found) obligation.status active obligation.owner owner db.add( ComplianceTask( obligation_idobligation.id, assigneeowner, due_dateobligation.due_date, statustodo, ) ) db.commit() return {ok: True, obligation_id: obligation.id, status: obligation.status}仪表盘接口提供风险分布和过期统计from sqlalchemy import func, and_ from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceObligation def get_dashboard_summary(db: Session): status_counts ( db.query(ComplianceObligation.status, func.count()) .group_by(ComplianceObligation.status) .all() ) risk_counts ( db.query(ComplianceObligation.risk_score, func.count()) .group_by(ComplianceObligation.risk_score) .all() ) now datetime.utcnow() overdue_obligations ( db.query(ComplianceObligation) .filter( ComplianceObligation.due_date.isnot(None), ComplianceObligation.due_date now, ComplianceObligation.status.notin_([closed, suppressed]), ) .count() ) return { status_counts: dict(status_counts), risk_counts: dict(risk_counts), overdue_obligation_count: overdue_obligations, }注意risk_counts直接用浮点数分组并不合适实际项目建议把风险分为高、中、低三档后再统计。这里只是为了演示概念。5. 验证方式与排查链路5.1 用本地模型和 Mock 数据跑通全流程在本地跑通流程时不一定要有云端模型。先在.env中配置VERITAS_DATABASE_URLpostgresqlpsycopg://veritas:veritaslocalhost:5432/veritas VERITAS_LLM_BASE_URLhttp://localhost:11434/v1 VERITAS_LLM_MODELqwen2.5:7b VERITAS_LLM_API_KEYollama如果已经启动 Ollama直接拉取模型ollama pull qwen2.5:7b然后启动 FastAPIuvicorn app.main:app --reload --port 8000准备一段示例法规文本通过接口导入curl -X POST http://localhost:8000/api/v1/documents \ -H Content-Type: application/json \ -d { title: Sample Data Protection Policy, content_text: The data controller must notify the relevant authority within 72 hours of a personal data breach. The controller shall also maintain a record of all processing activities., document_type: regulation }然后触发抽取curl -X POST http://localhost:8000/api/v1/documents/1/extract预期返回created大于 0并且生成的义务记录 status 为draft。如果要快速测试接口而模型尚未准备好最简单的方式是临时把LLMExtractor替换成返回固定结构的 Mock。不要在代码里写死 Mock而是通过环境变量控制if settings.llm_model mock: obligations [ExtractedObligation(...)] else: obligations await extractor.extract_many(...)这种设计能让前端开发和后端模型联调解耦。5.2 常见错误现象与排查路径合规追踪系统是一个典型的“数据流型”系统错误往往发生在输入、模型调用、校验和数据库写入四层。下面整理了一份高频排错表问题现象常见原因检查方式处理建议文档导入成功但抽取返回 0LLM 输出格式不符合预期查看模型返回原始内容和日志检查response_format是否被模型支持或改用规则兜底模型调用超时法规文本过长或模型响应慢检查接口耗时和超时配置增加切片逻辑分批请求提高超时时间但不超过 60 秒义务表出现 null 日期原文没有明确时间描述查看summary是否包含相对时间设置due_date为空并标记“人工补充截止日期”审核通过后任务未生成事务没有提交或接口报错查看接口返回和数据库任务表检查/approve接口的异常处理确认提交事务本地 Ollama 不支持 JSON mode模型配置与 OpenAI 参数不一致查看模型日志确认是否忽略response_format在提示词中强调只输出 JSON并在解析失败时做二次清洗数据库连接失败PostgreSQL 未启动或连接串错误执行docker compose ps检查端口和连接串启动容器确认.env中数据库 URL 正确排查顺序很重要。先确认输入内容有没有进库再确认模型有没有被调用之后看解析是否通过最后看数据库写入是否成功。不要一上来就去查业务代码。5.3 日志、监控和评估指标合规系统必须有审计日志。每次 AI 抽取、人工审核、任务状态变更都要记录操作人、时间和变更前后值。在最小系统中可以在关键函数里加统一的日志装饰器生产环境建议把日志发送到集中式平台。需要监控的指标包括抽取成功率LLM 返回可解析 JSON 的比例。校验通过率通过 Pydantic 校验的义务数量。人工修改率用户修改 AI 字段的比例。义务活跃时长从 draft 到 active 的平均时间。过期义务数每日过期未完成数量。人工修改率尤其重要。如果某个法域的文档经常被人工修改due_date说明 LLM 对该类文本的日期识别能力不足需要针对性调整提示词或增加规则。6. 生产环境落地与最佳实践6.1 学习环境与生产环境的差异本地跑通只是第一步。生产环境至少有五个方面要加固配置外置不要使用.env提交到仓库使用云环境变量服务或密钥管理工具。数据库迁移使用 Alembic 管理 schema 变更禁止手工改表结构。权限控制按公司、部门、法域划分数据权限避免一个员工看到其他地区的敏感合规记录。异步化AI 抽取耗时可能较长不能阻塞 HTTP 线程应该放入任务队列。审计记录每一次 AI 输出和人工修改的前后差异满足审计和监管要求。举例来说本地直接调用response_format很方便但在生产环境要评估这个参数是否在你选择的模型供应商中稳定。如果模型不支持 JSON mode你要在代码中多写一个“JSON 修复”函数把模型可能输出的 Markdown 代码块剥离出来。6.2 避免 AI 幻觉导致合规信息错误AI 在合规领域的最大风险不是“不会做”而是“一本正经地编造”。避免幻觉的核心不是换一个更大的模型而是设计流程约束。我的建议是所有 AI 抽取结果必须是draft状态不允许直接进入active。AI 输出必须携带confidence_score低于阈值的义务要标红提醒。法律条文原文必须保存在文档表中义务记录里的regulation_reference要能链接回原文片段。当法规文档更新时旧义务不能被静默删除而要进入suppressed并记录原因。重要法域可以要求双人复核避免单人误判。这五条比任何提示词工程都重要因为它们从流程上兜住了错误。6.3 可复用检查清单上线一个 AI 驱动的合规追踪系统前建议逐项核对以下清单检查项完成标准数据来源清晰每个文档都有来源 URL 或上传人文本切片策略长文档按章节或条款切片不直接截断模型提示词稳定同一文档重复抽取 3 次结构基本一致输出 schema 正确Pydantic 校验失败率低于 1%人工审核闭环没有人工审批的义务不会生成任务日期和时区统一所有 due_date 使用 UTC 保存展示时再转时区审计日志完整每个关键操作都有操作人、时间、前后值常规报告可用能按法域、风险等级、过期时间生成报告备份和恢复数据库每日备份且恢复演练通过这份清单可以直接放进项目文档也可以转化为 CI 检查项。6.4 扩展方向Veritas 的最小系统跑通后下一步可以沿着四个方向扩展。第一法规版本追踪。订阅监管机构更新源当法律文本变化时重新抽取义务并与旧义务做差异比对。这是合规系统最有价值的能力。第二多语言。通过翻译预处理或调用多语言模型让欧洲、东南亚、中东的法规都能进入同一套数据模型。第三自动提醒和证据收集。义务变为active后系统可以自动创建日历提醒并生成证据收集任务。例如“数据泄露通知义务”要求保存通知发送时间、发送对象和发送内容这些证据字段需要提前建模。第四AI Agent 化。当“抽取、审核、派单、催办、收集证据”都可以被可靠执行后可以设计成多个 Agent 协作的流程一个 Agent 负责法规监控一个负责任务分发一个负责证据链维护。但 Agent 化必须在人工审核闭环稳定之后再做否则错误会以更快的速度扩散。对刚开始学习这套系统的开发者最有效的练习不是追求复杂模型而是把一个真实法规文档跑通“导入、抽取、审核、生成任务、过期提醒”整个链路并尝试在没有 LLM 的情况下用规则引擎完成同样的事。只有同时理解规则和模型的边界才能真正把 AI 用进合规追踪这个严肃场景。 AI驱动的合规追踪器：从法规文本到可执行任务的最小系统实现
做国际化产品的第一步不是把功能上线而是先搞清楚产品在哪些市场受哪些规则约束。很多初创团队会把“合规”理解成找律师审一遍合同或者等到融资尽调才临时补文档但在实际运营里GDPR 的数据处理记录、CCPA 的消费者删除请求、SOC 2 的审计证据、出口管制产品清单每一项都对应明确的负责人和截止日期。Veritas 就是为这类场景设计的 AI 驱动的合规追踪器目标是让全球初创公司在不同法域下把散落在法规文本、合同条款和内部邮件里的合规义务自动整理成可追踪、可提醒、可审计的任务。这套系统听起来像一个“待办事项工具”但真正实现起来要比普通待办复杂不少。合规文本不是结构化数据法律条款中的义务主体、适用实体、履行期限和验证标准经常混在一段长文本里同一件事在不同法域可能又有不同定义。下面我会从核心设计、环境准备、代码实现到生产落地完整拆解一个可运行的 Veritas 最小系统。你可以把它当作学习项目逐步搭建也可以把它当成自建合规平台的初始骨架。1. 为什么全球初创公司需要 AI 驱动的合规追踪器1.1 合规催办不只是一个待办清单普通任务管理工具的核心对象是“某个员工要做某件事”合规追踪的核心对象是“某个法律义务是否被覆盖并落实”。这两者最大的区别在于合规义务会随着法规版本、业务范围、公司实体所在地的变化而变化而且每一件合规事项都必须能说清楚依据是什么、谁负责、什么时候之前必须完成。例如一家面向欧盟用户提供云服务的初创公司可能需要同时追踪 GDPR 下的数据处理记录义务、Cookie 同意机制、数据泄露通知时限以及客户合同中关于数据处理者的条款。这些事项分布在多个文档里靠 Excel 表格手工维护时常常出现三类问题责任人离职后事项无人接手法规更新后旧任务没有重跑审计时拿不出证据链。Veritas 要解决的正是“合规文本到可执行任务”之间的转换和持续追踪问题。1.2 传统工具在跨法域场景下的难点面向全球初创公司的合规追踪比单一国家内部合规更麻烦。每一个目标市场都有自己的监管机构和语言甚至同一个法律术语在不同法域的含义也不同。传统工具大致有几类但都有明显短板工具类型代表方式主要短板电子表格Excel、Google Sheets缺少任务状态、责任人、过期时间之间的强关系通用项目管理工具Jira、Trello只适合管理工程任务缺少法规依据和证据字段合同管理系统共享网盘、合同库能存档但不能从条款里自动提取义务外部合规咨询律师事务所、咨询顾问准确但成本高难以覆盖所有法域的日常更新真正的合规追踪应同时具备三个能力从非结构化文本中抽取结构化义务把义务转化为带状态和截止日期的任务在法规更新时能重新评估旧任务是否仍然成立。传统工具无法自动完成第一项和第三项所以“AI 驱动”在这里不是噱头而是解决核心问题的技术路径。1.3 Veritas 的系统定位和技术目标Veritas 不是要替代律师而是要做“合规运营的中控台”。律师或者合规负责人可以把法规原文、监管通知、内部政策文档导入系统系统借助 AI 生成义务草稿再由人工审核确认最后按照组织归属生成任务。技术目标可以拆成四条文本抽取从法规和合同条款中识别义务类型、适用实体、截止日期、风险等级。职责闭环每个义务都有负责人、状态、截止日期和审计记录。规则与模型结合AI 处理灵活文本规则引擎处理日期、关键词和已知模板。可观测所有抽取结果都有置信度所有状态变更都有日志。这套设计不是把全部判断交给模型而是让人工审核成为必要环节。后面章节的代码都会围绕这些目标展开。2. 核心设计如何把合规文本变成可追踪的任务2.1 合规数据的结构化模型在写代码之前先定义清楚数据边界。合规追踪系统最核心的表不是“任务表”而是“义务表”。一个合规义务是从某份文档中抽取出来的、对某个实体具有约束力的待办事项例如“在 72 小时内向监管机构报告个人数据泄露”。为了保持模型简洁又能支撑业务我建议至少设计四张表jurisdictions法域表记录国家或地区、监管机构。compliance_documents合规文档表保存原文和来源信息。compliance_obligations义务表保存抽取出的结构化义务。compliance_tasks任务表把义务分配给具体责任人并跟踪执行状态。义务表是 AI 抽取的产物任务表是人工协同的结果。不要把 AI 抽取结果直接当成任务否则一旦抽取错误追责和修正会非常困难。人工审核应该发生在“义务”和“任务”之间。2.2 规则引擎加 LLM 的混合抽取思路合规文本类型很多有的是结构化的法规条文有的是长合同条款有的甚至是一封监管邮件。LLM 很擅长理解自然语言但直接让 LLM 输出 JSON 并插入数据库存在幻觉和格式不稳定问题。更稳妥的做法是“规则引擎兜底LLM 做语义补充”。规则引擎负责两类事情日期识别例如“30日内”“within 30 days”“72 hours”转换为具体截止时间。关键词触发出现“shall”“必须”“应当”“is required to”时初步判断这里存在义务。LLM 负责更难的部分判断义务主体是谁、适用条件是什么、风险等级如何。最终结果通过 Pydantic Schema 校验只有通过校验的数据才能进入数据库。这样即使 LLM 输出不稳定数据库层面也不会写入格式错误的数据。2.3 任务状态机和风险评分模型义务和任务都要有明确状态但状态不宜在数据库里散落一地。建议义务状态使用以下流转draftAI 刚抽取等待人工审核。active人工审核通过已生成对应任务。suppressed人工判断不适用或错误。closed义务已经完成或失效。任务状态相对独立可以使用todo、in_progress、completed、cancelled。过期是一个计算字段而不是状态。判断逻辑是任务过期 当前时间 due_date AND status ! completed AND status ! cancelled风险评分建议使用 0 到 1 的小数。0.8 以上视为高风险0.4 到 0.8 为中风险0.4 以下为低风险。评分不是最终结论而是用于排序和提醒优先级。AI 给出的risk_score必须有人工复核入口。3. 环境准备与最小项目骨架3.1 技术选型与版本对齐实现 Veritas 最小系统我选用以下技术栈核心原则是结构清晰、便于替换 AI 服务Python 3.11FastAPISQLAlchemy 2.0PostgreSQLPydantic v2httpxpytest这里不需要把 AI 服务绑定在某一家厂商上。通过环境变量配置 OpenAI-compatible 的接口地址既可以接云端模型也可以接本地 Ollama。关键依赖版本如下fastapi0.115.6 uvicorn[standard]0.30.6 sqlalchemy2.0.36 psycopg[binary]3.2.3 pydantic2.9.2 pydantic-settings2.6.0 httpx0.27.2 python-dotenv1.0.1 pytest8.3.4实际项目落地前要确认这些版本已经兼容不要把版本号当作长期固定值。如果依赖冲突优先以 SQLAlchemy 和 FastAPI 的官方兼容说明为准。3.2 初始化项目结构和依赖项目目录建议按服务边界拆分避免把所有代码堆在一个文件里compliance_tracker/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── database.py │ ├── models.py │ ├── schemas.py │ ├── services/ │ │ ├── __init__.py │ │ ├── extractor.py │ │ └── rules.py │ └── routers/ │ ├── __init__.py │ ├── documents.py │ └── dashboard.py ├── tests/ ├── docker-compose.yml └── requirements.txt本地使用 Docker Compose 启动 PostgreSQL 最省事version: 3.9 services: postgres: image: postgres:15-alpine container_name: veritas-postgres environment: POSTGRES_USER: veritas POSTGRES_PASSWORD: veritas POSTGRES_DB: veritas ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata:启动数据库docker compose up -d这里要注意本地开发用这个默认密码没问题但生产环境必须通过环境变量或 Secret 管理工具注入凭证不能写在 Docker Compose 文件里。3.3 数据库模型和迁移SQLAlchemy 2.0 的声明式模型可以把 XML 映射彻底藏起来代码更直观。下面是核心模型的简化版本from datetime import datetime from sqlalchemy import String, Text, DateTime, Float, ForeignKey from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class Jurisdiction(Base): __tablename__ jurisdictions id: Mapped[int] mapped_column(primary_keyTrue) code: Mapped[str] mapped_column(String(16), uniqueTrue) name: Mapped[str] mapped_column(String(255)) regulatory_body: Mapped[str | None] mapped_column(String(255), nullableTrue) documents: Mapped[list[ComplianceDocument]] relationship(back_populatesjurisdiction) class ComplianceDocument(Base): __tablename__ compliance_documents id: Mapped[int] mapped_column(primary_keyTrue) jurisdiction_id: Mapped[int | None] mapped_column(ForeignKey(jurisdictions.id), nullableTrue) title: Mapped[str] mapped_column(String(500)) content_text: Mapped[str] mapped_column(Text) document_type: Mapped[str] mapped_column(String(64), defaultregulation) source_url: Mapped[str | None] mapped_column(String(1000), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) jurisdiction: Mapped[Jurisdiction | None] relationship(back_populatesdocuments) obligations: Mapped[list[ComplianceObligation]] relationship(back_populatesdocument)义务表是关键一定要把“从哪里来”和“归谁管”都记录清楚class ComplianceObligation(Base): __tablename__ compliance_obligations id: Mapped[int] mapped_column(primary_keyTrue) document_id: Mapped[int] mapped_column(ForeignKey(compliance_documents.id)) obligation_type: Mapped[str] mapped_column(String(128)) summary: Mapped[str] mapped_column(Text) regulation_reference: Mapped[str | None] mapped_column(String(500), nullableTrue) applicable_entity: Mapped[str | None] mapped_column(String(255), nullableTrue) due_date: Mapped[datetime | None] mapped_column(DateTime, nullableTrue) risk_score: Mapped[float] mapped_column(Float, default0.5) confidence_score: Mapped[float] mapped_column(Float, default0.0) status: Mapped[str] mapped_column(String(32), defaultdraft) owner: Mapped[str | None] mapped_column(String(255), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) updated_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) document: Mapped[ComplianceDocument] relationship(back_populatesobligations) tasks: Mapped[list[ComplianceTask]] relationship(back_populatesobligation)这里没有用 Alembic因为最小系统可以直接使用Base.metadata.create_all。实际项目建议从第一天就引入 Alembic否则数据库字段变更时没法做可回滚迁移。4. 实现核心模块AI 抽取、任务跟踪、仪表盘 API4.1 依赖注入与配置管理配置要做到“代码里不出现密钥”。使用pydantic-settings读取环境变量配置对象如下from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str postgresqlpsycopg://veritas:veritaslocalhost:5432/veritas llm_api_key: str llm_base_url: str https://api.openai.com/v1 llm_model: str gpt-4o-mini llm_timeout_seconds: float 30.0 default_owner: str complianceexample.com class Config: env_file .env env_prefix VERITAS_ settings Settings()env_prefix VERITAS_意味着本地环境变量使用VERITAS_DATABASE_URL、VERITAS_LLM_API_KEY等。如果设置了.env文件也要采用相同前缀。FastAPI 的依赖注入可以通过Depends获取数据库会话from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from fastapi import Depends engine create_engine(settings.database_url, pool_pre_pingTrue) SessionLocal sessionmaker(bindengine, autoflushFalse) def get_db(): db SessionLocal() try: yield db finally: db.close()4.2 实现 LLM 抽取服务与 JSON 校验LLM 抽取服务要承担两件事拼装提示词、解析并校验模型输出。先定义 Pydantic 输出结构from pydantic import BaseModel, Field class ExtractedObligation(BaseModel): obligation_type: str Field(description义务类型例如 data_protection、consumer_rights) summary: str Field(description义务内容摘要) regulation_reference: str | None Field(defaultNone, description法规或条款编号) applicable_entity: str | None Field(defaultNone, description适用实体例如 EEA 用户数据处理者) due_date: str | None Field(defaultNone, description如文本中有明确时间给出 ISO 格式日期) risk_score: float Field(default0.5, ge0, le1) confidence_score: float Field(default0.6, ge0, le1)抽取服务使用httpx请求 OpenAI-compatible 接口。为了方便本地测试这里把 base_url 设计成可以指向 Ollama 的http://localhost:11434/v1import json import httpx from pydantic import ValidationError from app.config import settings from app.schemas import ExtractedObligation class LLMExtractor: def __init__(self, client: httpx.AsyncClient | None None): self.client client async def _complete(self, prompt: str, system_prompt: str) - str: headers { Authorization: fBearer {settings.llm_api_key}, Content-Type: application/json, } payload { model: settings.llm_model, messages: [ {role: system, content: system_prompt}, {role: user, content: prompt}, ], temperature: 0.2, response_format: {type: json_object}, } url f{settings.llm_base_url}/chat/completions async with self.client or httpx.AsyncClient(timeoutsettings.llm_timeout_seconds) as client: response await client.post(url, headersheaders, jsonpayload) response.raise_for_status() data response.json() return data[choices][0][message][content] async def extract_many(self, text: str) - list[ExtractedObligation]: system_prompt ( 你是一个合规文本分析助手。请从用户提供的文本中抽取合规义务。 所有输出必须是合法的 JSON 对象格式为 {\obligations\: [...]}。 不要添加任何解释文字。 ) prompt f请从以下文本中抽取合规义务\n{text[:8000]} raw await self._complete(prompt, system_prompt) try: payload json.loads(raw) obligations [ExtractedObligation(**item) for item in payload[obligations]] return obligations except (json.JSONDecodeError, KeyError, ValidationError) as exc: raise ValueError(fLLM 输出无法解析为合规义务列表: {exc}) from exc这里有几个工程点必须说明温度设置 0.2 是为了让输出更稳定合规场景不需要创造性表达。response_format是 OpenAI-compatible 接口的通用参数但并不是所有本地模型都支持。如果本地模型不支持就要在提示词里强制要求 JSON 输出并做多一次格式修复。文本截取 8000 字符只是为了示例。实际项目应该按章节切片而不是直接截断避免丢失关键义务。4.3 实现规则兜底和日期识别如果 LLM 服务不可用或者模型输出为空规则引擎仍然要能给出一个“待人工审核”的草稿。下面是一个最小规则函数import re from datetime import datetime, timedelta def apply_rules(text: str) - list[dict]: results [] keyword_pattern re.compile(r(shall|must|应当|必须|应在|required to), re.IGNORECASE) for match in keyword_pattern.finditer(text): start max(0, match.start() - 120) end min(len(text), match.end() 180) snippet text[start:end] due_date find_due_date(snippet) results.append({ obligation_type: general_obligation, summary: snippet.strip(), regulation_reference: None, applicable_entity: None, due_date: due_date.isoformat() if due_date else None, risk_score: 0.5, confidence_score: 0.3, }) return results def find_due_date(text: str) - datetime | None: patterns [ (r(\d)\s*days?, timedelta(days1)), (r(\d)\s*hours?, timedelta(hours1)), (r(\d)\s*weeks?, timedelta(weeks1)), ] for pattern, unit in patterns: match re.search(pattern, text, re.IGNORECASE) if match: return datetime.utcnow() int(match.group(1)) * unit return None规则结果置信度设置为 0.3意味着“仅供参考”。真正使用时要让前端把这个草稿标记为“待人工校验”不能直接生成任务。4.4 实现文档导入、义务审核和仪表盘 APIFastAPI 路由可以分成三个模块但为了便于阅读这里给出核心方法。文档导入接口保存原文from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceDocument from app.schemas import DocumentCreate router APIRouter(prefix/api/v1/documents, tags[documents]) router.post() def create_document(payload: DocumentCreate, db: Session Depends(get_db)): doc ComplianceDocument(**payload.model_dump()) db.add(doc) db.commit() db.refresh(doc) return {id: doc.id, title: doc.title, created_at: doc.created_at}触发 AI 抽取的接口需要异步处理。最简实现是直接await抽取但生产环境建议把任务提交到 Celery 或 Arq避免接口请求超时from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceDocument, ComplianceObligation from app.services.extractor import LLMExtractor from app.services.rules import apply_rules router APIRouter(prefix/api/v1/documents, tags[documents]) router.post(/{document_id}/extract) async def extract_obligations(document_id: int, db: Session Depends(get_db)): document db.get(ComplianceDocument, document_id) if not document: raise HTTPException(status_code404, detaildocument not found) extractor LLMExtractor() obligations [] try: obligations await extractor.extract_many(document.content_text) except Exception: obligations apply_rules(document.content_text) created [] for item in obligations: obligation ComplianceObligation( document_iddocument.id, obligation_typeitem.obligation_type, summaryitem.summary, regulation_referenceitem.regulation_reference, applicable_entityitem.applicable_entity, due_dateitem.due_date, risk_scoreitem.risk_score, confidence_scoreitem.confidence_score, statusdraft, ) db.add(obligation) created.append(obligation) db.commit() return {created: len(created), source: llm if created else rule}人工审核接口。这里的关键是在审核通过时同步生成任务from datetime import datetime from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceObligation, ComplianceTask router APIRouter(prefix/api/v1/obligations, tags[obligations]) router.post(/{obligation_id}/approve) def approve_obligation(obligation_id: int, owner: str, db: Session Depends(get_db)): obligation db.get(ComplianceObligation, obligation_id) if not obligation: raise HTTPException(status_code404, detailobligation not found) obligation.status active obligation.owner owner db.add( ComplianceTask( obligation_idobligation.id, assigneeowner, due_dateobligation.due_date, statustodo, ) ) db.commit() return {ok: True, obligation_id: obligation.id, status: obligation.status}仪表盘接口提供风险分布和过期统计from sqlalchemy import func, and_ from sqlalchemy.orm import Session from app.database import get_db from app.models import ComplianceObligation def get_dashboard_summary(db: Session): status_counts ( db.query(ComplianceObligation.status, func.count()) .group_by(ComplianceObligation.status) .all() ) risk_counts ( db.query(ComplianceObligation.risk_score, func.count()) .group_by(ComplianceObligation.risk_score) .all() ) now datetime.utcnow() overdue_obligations ( db.query(ComplianceObligation) .filter( ComplianceObligation.due_date.isnot(None), ComplianceObligation.due_date now, ComplianceObligation.status.notin_([closed, suppressed]), ) .count() ) return { status_counts: dict(status_counts), risk_counts: dict(risk_counts), overdue_obligation_count: overdue_obligations, }注意risk_counts直接用浮点数分组并不合适实际项目建议把风险分为高、中、低三档后再统计。这里只是为了演示概念。5. 验证方式与排查链路5.1 用本地模型和 Mock 数据跑通全流程在本地跑通流程时不一定要有云端模型。先在.env中配置VERITAS_DATABASE_URLpostgresqlpsycopg://veritas:veritaslocalhost:5432/veritas VERITAS_LLM_BASE_URLhttp://localhost:11434/v1 VERITAS_LLM_MODELqwen2.5:7b VERITAS_LLM_API_KEYollama如果已经启动 Ollama直接拉取模型ollama pull qwen2.5:7b然后启动 FastAPIuvicorn app.main:app --reload --port 8000准备一段示例法规文本通过接口导入curl -X POST http://localhost:8000/api/v1/documents \ -H Content-Type: application/json \ -d { title: Sample Data Protection Policy, content_text: The data controller must notify the relevant authority within 72 hours of a personal data breach. The controller shall also maintain a record of all processing activities., document_type: regulation }然后触发抽取curl -X POST http://localhost:8000/api/v1/documents/1/extract预期返回created大于 0并且生成的义务记录 status 为draft。如果要快速测试接口而模型尚未准备好最简单的方式是临时把LLMExtractor替换成返回固定结构的 Mock。不要在代码里写死 Mock而是通过环境变量控制if settings.llm_model mock: obligations [ExtractedObligation(...)] else: obligations await extractor.extract_many(...)这种设计能让前端开发和后端模型联调解耦。5.2 常见错误现象与排查路径合规追踪系统是一个典型的“数据流型”系统错误往往发生在输入、模型调用、校验和数据库写入四层。下面整理了一份高频排错表问题现象常见原因检查方式处理建议文档导入成功但抽取返回 0LLM 输出格式不符合预期查看模型返回原始内容和日志检查response_format是否被模型支持或改用规则兜底模型调用超时法规文本过长或模型响应慢检查接口耗时和超时配置增加切片逻辑分批请求提高超时时间但不超过 60 秒义务表出现 null 日期原文没有明确时间描述查看summary是否包含相对时间设置due_date为空并标记“人工补充截止日期”审核通过后任务未生成事务没有提交或接口报错查看接口返回和数据库任务表检查/approve接口的异常处理确认提交事务本地 Ollama 不支持 JSON mode模型配置与 OpenAI 参数不一致查看模型日志确认是否忽略response_format在提示词中强调只输出 JSON并在解析失败时做二次清洗数据库连接失败PostgreSQL 未启动或连接串错误执行docker compose ps检查端口和连接串启动容器确认.env中数据库 URL 正确排查顺序很重要。先确认输入内容有没有进库再确认模型有没有被调用之后看解析是否通过最后看数据库写入是否成功。不要一上来就去查业务代码。5.3 日志、监控和评估指标合规系统必须有审计日志。每次 AI 抽取、人工审核、任务状态变更都要记录操作人、时间和变更前后值。在最小系统中可以在关键函数里加统一的日志装饰器生产环境建议把日志发送到集中式平台。需要监控的指标包括抽取成功率LLM 返回可解析 JSON 的比例。校验通过率通过 Pydantic 校验的义务数量。人工修改率用户修改 AI 字段的比例。义务活跃时长从 draft 到 active 的平均时间。过期义务数每日过期未完成数量。人工修改率尤其重要。如果某个法域的文档经常被人工修改due_date说明 LLM 对该类文本的日期识别能力不足需要针对性调整提示词或增加规则。6. 生产环境落地与最佳实践6.1 学习环境与生产环境的差异本地跑通只是第一步。生产环境至少有五个方面要加固配置外置不要使用.env提交到仓库使用云环境变量服务或密钥管理工具。数据库迁移使用 Alembic 管理 schema 变更禁止手工改表结构。权限控制按公司、部门、法域划分数据权限避免一个员工看到其他地区的敏感合规记录。异步化AI 抽取耗时可能较长不能阻塞 HTTP 线程应该放入任务队列。审计记录每一次 AI 输出和人工修改的前后差异满足审计和监管要求。举例来说本地直接调用response_format很方便但在生产环境要评估这个参数是否在你选择的模型供应商中稳定。如果模型不支持 JSON mode你要在代码中多写一个“JSON 修复”函数把模型可能输出的 Markdown 代码块剥离出来。6.2 避免 AI 幻觉导致合规信息错误AI 在合规领域的最大风险不是“不会做”而是“一本正经地编造”。避免幻觉的核心不是换一个更大的模型而是设计流程约束。我的建议是所有 AI 抽取结果必须是draft状态不允许直接进入active。AI 输出必须携带confidence_score低于阈值的义务要标红提醒。法律条文原文必须保存在文档表中义务记录里的regulation_reference要能链接回原文片段。当法规文档更新时旧义务不能被静默删除而要进入suppressed并记录原因。重要法域可以要求双人复核避免单人误判。这五条比任何提示词工程都重要因为它们从流程上兜住了错误。6.3 可复用检查清单上线一个 AI 驱动的合规追踪系统前建议逐项核对以下清单检查项完成标准数据来源清晰每个文档都有来源 URL 或上传人文本切片策略长文档按章节或条款切片不直接截断模型提示词稳定同一文档重复抽取 3 次结构基本一致输出 schema 正确Pydantic 校验失败率低于 1%人工审核闭环没有人工审批的义务不会生成任务日期和时区统一所有 due_date 使用 UTC 保存展示时再转时区审计日志完整每个关键操作都有操作人、时间、前后值常规报告可用能按法域、风险等级、过期时间生成报告备份和恢复数据库每日备份且恢复演练通过这份清单可以直接放进项目文档也可以转化为 CI 检查项。6.4 扩展方向Veritas 的最小系统跑通后下一步可以沿着四个方向扩展。第一法规版本追踪。订阅监管机构更新源当法律文本变化时重新抽取义务并与旧义务做差异比对。这是合规系统最有价值的能力。第二多语言。通过翻译预处理或调用多语言模型让欧洲、东南亚、中东的法规都能进入同一套数据模型。第三自动提醒和证据收集。义务变为active后系统可以自动创建日历提醒并生成证据收集任务。例如“数据泄露通知义务”要求保存通知发送时间、发送对象和发送内容这些证据字段需要提前建模。第四AI Agent 化。当“抽取、审核、派单、催办、收集证据”都可以被可靠执行后可以设计成多个 Agent 协作的流程一个 Agent 负责法规监控一个负责任务分发一个负责证据链维护。但 Agent 化必须在人工审核闭环稳定之后再做否则错误会以更快的速度扩散。对刚开始学习这套系统的开发者最有效的练习不是追求复杂模型而是把一个真实法规文档跑通“导入、抽取、审核、生成任务、过期提醒”整个链路并尝试在没有 LLM 的情况下用规则引擎完成同样的事。只有同时理解规则和模型的边界才能真正把 AI 用进合规追踪这个严肃场景。