当一家初创公司从本地市场走向全球时最先遇到的可能不是业务增长问题而是合规问题。GDPR、CCPA、PIPL、LGPD、SOC 2……不同国家和地区的法规不断更新人工跟踪很容易出现遗漏。Veritas 就是为解决这个场景设计的一个基于 AI 的全球初创企业合规跟踪器用 AI 辅助法规分类、任务生成和风险预警同时保留人工审核入口避免“全自动踩雷”。下面完整拆解这个系统的设计思路、核心代码和落地注意事项。1. 项目背景与核心概念1.1 什么是合规跟踪器合规跟踪器是一类帮助企业识别、记录、跟踪和完成监管义务的软件系统。传统的合规管理通常依赖 Excel 表、邮件提醒和线下检查一旦企业进入多个司法管辖区法规数量会指数级增长。一家 SaaS 公司如果同时服务欧盟、美国和东南亚用户可能同时受到 GDPR、CCPA、PDPA 等多部法律约束而每部法律又包含数十条具体义务。合规跟踪器的作用就是把这些分散的法规要求整理成结构化任务分配给负责人并持续跟踪完成状态。Veritas 在这个基础上增加了 AI 环节。它不只是“法规库 任务列表”而是尝试用大模型自动解读新法规、判断适用范围、生成合规义务并给出风险等级建议。这样做的好处是降低人工整理成本尤其在法规文本很长、更新频繁的场景下AI 可以快速产出“初稿”再由合规人员审核确认。1.2 为什么选择 AI 做合规合规本质上是一个“信息抽取 规则匹配 任务管理”的过程。传统实现方式有两种一种是纯人工成本高、速度慢另一种是写死规则只能覆盖已知法规难以应对新法条。AI 的强项在于理解自然语言可以从长篇法律文本中提炼出“谁、在什么情况下、需要做什么、截止日期是什么”这类关键信息。不过 AI 也并非万能。大模型存在“幻觉”问题可能生成不存在的义务或错误的适用条件。因此 Veritas 在设计上把 AI 定位为“辅助引擎”而不是最终决策者。AI 生成的每一项义务都必须进入pending_review状态经过人工确认后才能变成正式任务。这个设计非常重要能避免企业把 AI 幻觉当成真实合规要求导致资源浪费或误导。1.3 Veritas 的功能边界Veritas 的核心功能包括四块公司信息管理、法规库管理、AI 合规分析、任务与风险跟踪。公司信息用于提供 AI 分析所需的上下文比如业务地区、所属行业、员工规模法规库用于存储原始法律文本或摘要AI 合规分析负责把法规与公司画像结合生成适用义务任务与风险跟踪用于把义务拆成具体待办事项并给出风险评分。在边界上Veritas 不会尝试“自动代扣罚款”或“自动提交申报文件”那是政府对接层的事。它更接近一个“合规工作台”帮助团队知道需要做什么、什么时候做、做到什么程度。下面我们从技术选型开始一步步实现这个系统。2. 环境准备与技术选型2.1 技术栈说明Veritas 后端采用 Python 生态核心依赖包括 FastAPI、SQLAlchemy、Pydantic 和一个可切换的 LLM 客户端。FastAPI 用于提供 REST API自带 Swagger 文档方便调试和对接前端SQLAlchemy 负责数据库读写支持 SQLite、PostgreSQL、MySQL 等常见数据库Pydantic 负责接口入参校验和响应序列化。LLM 客户端抽象成接口是项目的一个关键设计。开发阶段用 Mock 实现不需要真实模型也可以跑通整个流程生产环境可以切换到 OpenAI、国内大模型或其他私有化部署模型。这样既能快速演示又能避免把某个供应商写死在业务代码里。版本方面建议使用 Python 3.10 及以上、FastAPI 0.100 以上、SQLAlchemy 2.x。实际版本需要根据你的项目情况调整本文示例以常见环境为例重点演示配置思路而不是绑定某个固定版本。2.2 项目目录结构建议把项目拆成多个模块避免把所有代码塞进一个文件。目录结构如下veritas/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models.py │ ├── llm_client.py │ └── compliance_service.py ├── requirements.txt ├── .env └── README.mdmain.py是 FastAPI 入口负责创建应用和路由models.py定义数据库表结构llm_client.py封装不同的大模型客户端compliance_service.py是核心业务逻辑层负责把法规和公司信息变成合规义务与任务。2.3 环境变量与配置项目使用环境变量控制数据库地址和 LLM 提供方避免把密钥硬编码到代码里。.env文件示例如下DATABASE_URLsqlite:///./veritas.db LLM_PROVIDERmock OPENAI_API_KEYsk-xxx OPENAI_MODELgpt-4o-mini在本地开发阶段LLM_PROVIDER设置为mock不需要真实 API Key。接入生产模型时再改为openai并填入对应密钥。这里特别提醒任何真实密钥都只能放在服务端环境变量或密钥管理服务中绝不能提交到 Git 仓库否则一旦泄露就可能被恶意使用。3. 核心模块设计3.1 数据模型设计Veritas 的数据库模型包括五张表公司信息表company_profiles、法规表regulations、合规义务表compliance_obligations、合规任务表compliance_tasks和审计日志表audit_logs。公司表存储企业名称、目标市场、行业和员工数法规表存储法规标题、适用地区、生效日期、来源链接和全文义务表是 AI 分析结果包含义务摘要、适用范围、风险等级、AI 置信度和审核状态任务表把义务拆成可执行的待办审计日志表记录谁在什么时候做了什么操作满足企业内部审计需求。这里需要重点说明的是义务表上的唯一约束。AI 可能会被重复调用如果不加控制同一法规对公司可能生成大量重复义务。唯一约束(company_id, regulation_id, obligation_summary)可以从数据库层面兜底但更稳妥的做法是在业务层先查询已存在的记录避免重复分析。3.2 LLM 客户端抽象LLM 客户端抽象的核心是定义一个classify_regulation方法输入法规文本和公司信息输出结构化的合规分析结果。输出结果至少包含四类信息义务列表、适用范围、风险等级、建议截止日期和置信度。{ obligations: [用户提交数据前必须获得明确同意], scope: 所有面向欧盟用户的线上服务, risk_level: high, due_suggestion: 2025-12-31, confidence: 0.85 }risk_level是low、medium、high三选一confidence是 0 到 1 之间的小数表示模型对结果的把握程度。Mock 实现返回固定结果方便测试OpenAI 实现通过 prompt 让大模型输出 JSON再用json.loads解析。3.3 AI 合规分析引擎分析引擎是整个系统的业务核心它把公司画像和法规文本交给 LLM拿到结果后保存义务并创建任务。在analyze_regulation方法中第一步是从数据库读取公司信息和法规第二步构造公司信息字典第三步调用 LLM第四步解析日期和风险等级第五步创建ComplianceObligation记录第六步根据义务列表创建ComplianceTask最后写入审计日志。日期解析不能直接信任模型输出。模型可能返回2025/12/31、2025-12-31或者纯文本“三个月内”所以_parse_date方法需要做格式兼容解析失败就返回None由人工后来补充。3.4 任务生成与风险评分AI 返回的obligations是一个列表每一项都是一条独立义务。任务生成时把每条义务作为一条待办事项写入compliance_tasks标题可以用“落实合规义务 1”“落实合规义务 2”这类自动编号便于快速认领。风险评分采用简单加权模型。每个义务的risk_level映射为权重low 是 1medium 是 2high 是 3。企业所有已审核通过义务的权重相加得到一个最高 100 的风险分。分数越高说明当前需要处理的合规负担越大。这个评分不是法律意义上的风险评级而是帮助团队排定优先级的一个参考信号。4. 完整实战案例4.1 初始化项目与所需依赖先创建项目目录和虚拟环境然后准备依赖文件。在requirements.txt中写入以下内容fastapi uvicorn[standard] sqlalchemy pydantic pydantic-settings python-dotenv openai安装依赖pip install -r requirements.txt如果只需要跑 Mock 模式openai也可以先不装但为了方便后续切换建议一次性装好。4.2 配置环境变量在项目根目录创建.env文件DATABASE_URLsqlite:///./veritas.db LLM_PROVIDERmock OPENAI_API_KEY OPENAI_MODELgpt-4o-mini本地先用 SQLite不需要额外启动数据库服务。后续接入 PostgreSQL 时只需要修改DATABASE_URL例如DATABASE_URLpostgresqlpsycopg://user:passwordlocalhost:5432/veritas4.3 编写数据库模型 models.py文件路径app/models.pyfrom datetime import datetime from sqlalchemy import ( Column, Integer, String, DateTime, Date, Text, ForeignKey, Boolean, Float, UniqueConstraint, func ) from sqlalchemy.orm import declarative_base, relationship Base declarative_base() class CompanyProfile(Base): __tablename__ company_profiles id Column(Integer, primary_keyTrue, indexTrue) name Column(String(200), nullableFalse) region Column(String(20), nullableFalse) industry Column(String(100), nullableFalse) employee_count Column(Integer, default0) created_at Column(DateTime, server_defaultfunc.now()) obligations relationship(ComplianceObligation, back_populatescompany) class Regulation(Base): __tablename__ regulations id Column(Integer, primary_keyTrue, indexTrue) title Column(String(300), nullableFalse) region Column(String(20), nullableFalse) category Column(String(50), nullableTrue) effective_date Column(String(20), nullableTrue) source_url Column(String(500), nullableTrue) full_text Column(Text, nullableFalse) created_at Column(DateTime, server_defaultfunc.now()) obligations relationship(ComplianceObligation, back_populatesregulation) class ComplianceObligation(Base): __tablename__ compliance_obligations id Column(Integer, primary_keyTrue, indexTrue) company_id Column(Integer, ForeignKey(company_profiles.id), nullableFalse) regulation_id Column(Integer, ForeignKey(regulations.id), nullableFalse) obligation_summary Column(Text, nullableFalse) applicable_scope Column(String(500), nullableTrue) risk_level Column(String(20), defaultlow) status Column(String(20), defaultpending_review) ai_confidence Column(Float, default0.0) due_date Column(Date, nullableTrue) created_at Column(DateTime, server_defaultfunc.now()) updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) company relationship(CompanyProfile, back_populatesobligations) regulation relationship(Regulation, back_populatesobligations) tasks relationship(ComplianceTask, back_populatesobligation) __table_args__ ( UniqueConstraint( company_id, regulation_id, obligation_summary, nameuniq_obligation ), ) class ComplianceTask(Base): __tablename__ compliance_tasks id Column(Integer, primary_keyTrue, indexTrue) obligation_id Column(Integer, ForeignKey(compliance_obligations.id), nullableFalse) title Column(String(300), nullableFalse) description Column(Text, nullableTrue) status Column(String(20), defaulttodo) assignee Column(String(100), nullableTrue) due_date Column(Date, nullableTrue) completed_at Column(DateTime, nullableTrue) created_at Column(DateTime, server_defaultfunc.now()) obligation relationship(ComplianceObligation, back_populatestasks) class AuditLog(Base): __tablename__ audit_logs id Column(Integer, primary_keyTrue, indexTrue) entity_type Column(String(50), nullableFalse) entity_id Column(Integer, nullableFalse) action Column(String(100), nullableFalse) detail Column(Text, nullableTrue) created_at Column(DateTime, server_defaultfunc.now())在这个模型中AI_confidence字段用来记录模型输出置信度status字段先设为pending_review等待人工审核。如果后续要增加“被驳回”状态可以继续扩展状态枚举。每张表都带created_at方便追溯数据产生时间。4.4 编写 LLM 客户端 llm_client.py文件路径app/llm_client.pyimport json from abc import ABC, abstractmethod from openai import OpenAI class LLMClient(ABC): abstractmethod def classify_regulation(self, regulation_text: str, company_info: dict) - dict: 返回示例 { obligations: [..., ...], scope: ..., risk_level: low|medium|high, due_suggestion: 2025-12-31, confidence: 0.85 } ... class MockLLMClient(LLMClient): def classify_regulation(self, regulation_text: str, company_info: dict) - dict: return { obligations: [ 根据用户所在地区提供隐私政策, 用户行使删除权时需在30天内响应, ], scope: 面向目标市场的线上服务, risk_level: medium, due_suggestion: 2025-12-31, confidence: 0.75, } class OpenAILLMClient(LLMClient): def __init__(self, api_key: str, model: str gpt-4o-mini): self.client OpenAI(api_keyapi_key) self.model model def classify_regulation(self, regulation_text: str, company_info: dict) - dict: prompt self._build_prompt(regulation_text, company_info) response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是合规分析助手。请只输出JSON不要输出多余内容。}, {role: user, content: prompt}, ], temperature0, ) content response.choices[0].message.content return json.loads(self._clean_json(content)) def _build_prompt(self, regulation_text: str, company_info: dict) - str: company_desc ( f公司名称{company_info[name]}目标市场{company_info[region]} f所属行业{company_info[industry]}员工数{company_info[employee_count]} ) return ( 请根据公司信息分析以下法规输出适用义务、范围、风险等级、建议截止日期、置信度。\n f公司信息{company_desc}\n f法规正文\n{regulation_text[:4000]} ) def _clean_json(self, text: str) - str: text text.strip() if text.startswith(json): text text[7:] elif text.startswith(): text text[3:] if text.endswith(): text text[:-3] return text.strip()这里有两个关键点。第一LLMClient是抽象基类业务代码只依赖这个抽象不依赖具体供应商。第二_clean_json用来去除模型输出中可能存在的 Markdown 代码块标记否则json.loads会直接报错。实际使用时不同模型可能输出不同的包裹格式需要针对模型调优。4.5 编写合规分析服务 compliance_service.py文件路径app/compliance_service.pyfrom datetime import date from sqlalchemy.orm import Session from .models import ( CompanyProfile, Regulation, ComplianceObligation, ComplianceTask, AuditLog, ) from .llm_client import LLMClient RISK_WEIGHT {low: 1, medium: 2, high: 3} class ComplianceService: def __init__(self, db: Session, llm: LLMClient): self.db db self.llm llm def analyze_regulation(self, company_id: int, regulation_id: int) - ComplianceObligation: company self.db.get(CompanyProfile, company_id) regulation self.db.get(Regulation, regulation_id) if not company or not regulation: raise ValueError(company or regulation not found) company_info { name: company.name, region: company.region, industry: company.industry, employee_count: company.employee_count, } result self.llm.classify_regulation(regulation.full_text, company_info) due_date self._parse_date(result.get(due_suggestion)) risk_level self._normalize_risk(result.get(risk_level, medium)) obligation ComplianceObligation( company_idcompany.id, regulation_idregulation.id, obligation_summary\n.join(result.get(obligations, [])), applicable_scoperesult.get(scope), risk_levelrisk_level, statuspending_review, ai_confidencefloat(result.get(confidence, 0.0)), due_datedue_date, ) self.db.add(obligation) self.db.flush() self._create_tasks(obligation, result.get(obligations, []), due_date) self._add_audit_log( compliance_obligation, obligation.id, create, fAI生成义务待人工审核置信度{obligation.ai_confidence}, ) self.db.commit() self.db.refresh(obligation) return obligation def _create_tasks(self, obligation, obligations, due_date): for idx, obligation_text in enumerate(obligations, start1): task ComplianceTask( obligation_idobligation.id, titlef落实合规义务 {idx}, descriptionobligation_text, statustodo, due_datedue_date, ) self.db.add(task) def approve_obligation(self, obligation_id: int) - ComplianceObligation: obligation self.db.get(ComplianceObligation, obligation_id) if not obligation: raise ValueError(obligation not found) obligation.status approved self._add_audit_log( compliance_obligation, obligation_id, approve, 人工审核通过, ) self.db.commit() self.db.refresh(obligation) return obligation def calculate_risk_score(self, company_id: int) - int: obligations self.db.query(ComplianceObligation).filter( ComplianceObligation.company_id company_id, ComplianceObligation.status approved, ).all() if not obligations: return 0 total 0 for obligation in obligations: total RISK_WEIGHT.get(obligation.risk_level, 0) return min(total, 100) def _normalize_risk(self, value) - str: value (value or ).lower() if value in (high, 高风险): return high if value in (low, 低风险): return low return medium def _parse_date(self, value): if not value: return None try: return date.fromisoformat(value) except ValueError: return None def _add_audit_log(self, entity_type, entity_id, action, detail): log AuditLog( entity_typeentity_type, entity_identity_id, actionaction, detaildetail, ) self.db.add(log)这段代码的核心是analyze_regulation。它首先从数据库取公司信息和法规然后把它们传给 LLM拿到 JSON 结果后写入数据库。这里没有直接信任模型返回的所有字段而是做了标准化处理风险等级统一转成三档日期解析失败则留空置信度强制转成浮点数。这样即使不同模型返回格式有细微差异也不会导致系统崩溃。4.6 编写 FastAPI 入口 main.py文件路径app/main.pyimport os from datetime import date from typing import List from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from .models import ( Base, CompanyProfile, Regulation, ComplianceObligation, ComplianceTask, ) from .llm_client import MockLLMClient, OpenAILLMClient from .compliance_service import ComplianceService DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./veritas.db) engine create_engine(DATABASE_URL, connect_args{check_same_thread: False}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base.metadata.create_all(bindengine) app FastAPI(titleVeritas Compliance Tracker, version0.1.0) def get_db(): db SessionLocal() try: yield db finally: db.close() def get_llm(): provider os.getenv(LLM_PROVIDER, mock).lower() if provider openai: return OpenAILLMClient( api_keyos.getenv(OPENAI_API_KEY, ), modelos.getenv(OPENAI_MODEL, gpt-4o-mini), ) return MockLLMClient() class CompanyCreate(BaseModel): name: str region: str industry: str employee_count: int 0 class CompanyOut(BaseModel): id: int name: str region: str industry: str employee_count: int class Config: from_attributes True class RegulationCreate(BaseModel): title: str Field(..., description法规标题) region: str Field(..., description适用地区例如 EU/US) category: str | None None effective_date: str | None None source_url: str | None None full_text: str Field(..., min_length50, description法规全文或关键摘要) class RegulationOut(BaseModel): id: int title: str region: str category: str | None effective_date: str | None source_url: str | None full_text: str class Config: from_attributes True class TaskOut(BaseModel): id: int obligation_id: int title: str description: str status: str assignee: str | None due_date: str | None completed_at: str | None class Config: from_attributes True app.post(/companies, response_modelCompanyOut, status_code201) def create_company(payload: CompanyCreate, db: Session Depends(get_db)): company CompanyProfile( namepayload.name, regionpayload.region, industrypayload.industry, employee_countpayload.employee_count, ) db.add(company) db.commit() db.refresh(company) return company app.post(/regulations, response_modelRegulationOut, status_code201) def create_regulation(payload: RegulationCreate, db: Session Depends(get_db)): regulation Regulation( titlepayload.title, regionpayload.region, categorypayload.category, effective_datepayload.effective_date, source_urlpayload.source_url, full_textpayload.full_text, ) db.add(regulation) db.commit() db.refresh(regulation) return regulation app.post(/analyze/{regulation_id}) def analyze(regulation_id: int, company_id: int, db: Session Depends(get_db)): llm get_llm() service ComplianceService(db, llm) try: obligation service.analyze_regulation(company_id, regulation_id) except ValueError as e: raise HTTPException(status_code404, detailstr(e)) return { obligation_id: obligation.id, obligation_summary: obligation.obligation_summary, risk_level: obligation.risk_level, status: obligation.status, ai_confidence: obligation.ai_confidence, due_date: obligation.due_date.isoformat() if obligation.due_date else None, } app.post(/obligations/{obligation_id}/approve) def approve(obligation_id: int, db: Session Depends(get_db)): service ComplianceService(db, get_llm()) try: obligation service.approve_obligation(obligation_id) except ValueError as e: raise HTTPException(status_code404, detailstr(e)) return {id: obligation.id, status: obligation.status} app.get(/companies/{company_id}/tasks, response_modelList[TaskOut]) def list_tasks(company_id: int, db: Session Depends(get_db)): obligations db.query(ComplianceObligation).filter( ComplianceObligation.company_id company_id ).all() obligation_ids [o.id for o in obligations] if not obligation_ids: return [] tasks db.query(ComplianceTask).filter( ComplianceTask.obligation_id.in_(obligation_ids) ).all() return tasksmain.py中需要注意两点。第一Base.metadata.create_all(bindengine)在开发阶段会自动建表生产环境更推荐使用 Alembic 做迁移避免表结构变更导致的数据丢失。第二get_llm()根据环境变量切换模型客户端这个模式可以继续扩展成工厂函数比如根据请求头或租户信息选择不同模型。4.7 运行与验证在项目根目录执行启动命令uvicorn app.main:app --reload启动成功后访问http://127.0.0.1:8000/docs能看到 Swagger 接口文档。接下来按顺序创建公司、创建法规、执行分析。先创建一家公司curl -X POST http://127.0.0.1:8000/companies \ -H Content-Type: application/json \ -d { name: Example AI Startup, region: EU, industry: SaaS, employee_count: 120 }预期返回包含公司 id 的 JSON。再创建一条法规curl -X POST http://127.0.0.1:8000/regulations \ -H Content-Type: application/json \ -d { title: GDPR Privacy Policy Requirements, region: EU, category: privacy, effective_date: 2025-01-01, source_url: https://example.com/gdpr-summary, full_text: 企业收集欧盟用户个人数据前应明确告知数据处理目的、保存期限、用户权利等信息。用户有权请求删除个人数据企业应在合理时间内响应。 }然后调用分析接口把法规 id 和公司 id 传入curl -X POST http://127.0.0.1:8000/analyze/1?company_id1Mock 模式返回的结果中义务状态会是pending_review置信度为0.75。接着可以查看该公司所有任务curl http://127.0.0.1:8000/companies/1/tasks通过 Swagger 页面可以直接点击执行普通团队成员也能快速学会使用。5. 常见问题与排查思路5.1 高频问题速查表问题现象常见原因解决思路AI 分类结果不稳定Prompt 设计不够明确模型随机性高使用结构化输出要求设置temperature0增加 few-shot 示例法规文本过长超出模型上下文输入整篇法规全文预分割文本或引入 RAG 只检索关键条款同一法规反复生成重复义务缺少幂等控制查询已有记录增加唯一约束json.loads解析失败模型输出包含 Markdown 代码块用_clean_json剥离代码块标记并做异常兜底风险等级无法排序字符串格式不统一在业务层标准化为low/medium/high生产环境 API Key 泄露密钥写入代码或日志使用环境变量日志脱敏配置最小权限5.2 排查流程建议如果 AI 分析结果不理想不要先改代码先看原始输入输出。可以写一个简单的 Python 脚本直接调用llm.classify_regulation打印完整 prompt 和模型返回值确认是不是模型本身的问题。确认模型输出没问题后再检查解析层和数据库层。如果任务创建成功但查询不到优先检查数据库连接配置是否指向正确的库。SQLite 默认在当前目录如果启动命令的工作目录不一致可能会出现“表存在但数据看不到”的情况。建议在接口里加一个日志输出记录当前数据库路径和创建记录数。6. 最佳实践与工程建议6.1 AI 输出质量控制AI 在法律场景下的输出必须保持克制。建议在系统提示词中明确要求“只输出 JSON”并用 few-shot 示例约束格式。同时在业务层增加置信度阈值当ai_confidence低于 0.6 时强制进入pending_review并标记为“低置信度建议优先人工检查”。这样即使用户没有主动看状态团队也会注意到高风险记录。大模型采用temperature0可以降低随机性但无法完全消除幻觉。更可靠的方式是引入检索增强生成把法规文本分词后存入向量数据库分析时只检索与公司相关的条款片段减少过长输入带来的注意力分散。这是 Veritas 下一步可以优化的关键方向。6.2 数据安全与隐私合规系统本身处理的就是敏感数据尤其是法规原文和公司内部义务信息。所有接口必须加认证和授权不能直接暴露在公网。生产环境建议使用 OAuth2、API Key 或企业内部 SSO。涉及用户个人数据的场景还要注意数据最小化原则只采集服务所需字段避免把无关个人信息发送给大模型。调用外部大模型时要评估数据是否会被供应商留存。如果公司信息或法规文本属于敏感内容建议采用私有化部署模型或在发送前做字段脱敏。日志记录方面不能把完整 prompt 和模型原始输出原样打到日志里必要时应截断或替换敏感字段。6.3 合规任务生命周期Veritas 目前只实现了最简单的任务状态todo、done通过completed_at判断。真实生产项目建议引入状态机包含todo、in_progress、in_review、done、cancelled等状态。每次状态变更都要写入审计日志并记录操作人、操作时间和变更原因。截止日期管理也非常重要。AI 建议的due_date只是预估需要由业务负责人确认。可以增加定时任务在截止日期前 7 天、3 天、1 天发送提醒避免遗漏。生产环境需要额外考虑日期展示时区建议统一存储为 UTC展示时再转换到用户本地时区。6.4 可扩展性考虑如果未来要支持更多国家法规不建议把所有法规全文都塞进数据库后丢给模型。更合理的做法是引入两层结构第一层是法规元数据包括标题、适用地区、分类第二层是义务详情每一条义务单独存储便于检索和对比。法规语义检索可以用向量数据库也可以用传统全文索引根据团队能力和数据量决定。对多租户支持敏感的 SaaS 场景数据模型需要增加tenant_id字段所有查询都按租户隔离。当前示例没有这个字段开发者可以在CompanyProfile上增加租户概念并在路由层强制传入。7. 总结与下一步学习方向这篇文章围绕 Veritas 这个 AI 驱动的全球初创企业合规跟踪器从需求背景、技术选型、数据库设计、LLM 抽象、合规分析引擎到 FastAPI 接口完成了一个可运行的 MVP。可以看到AI 在这类系统里的核心价值是“快速起草初稿”而不是替代专业人士做判断。真正的可靠来自两件事一是把模型输出标准化二是让所有结果都经过人工复核。下一步建议从三个方向继续深入。第一学习使用向量数据库和 Embedding 模型做法规语义检索解决长文本分析不准确的问题第二尝试接入定时任务和消息通知让任务闭环更完整第三引入更细粒度的权限模型把系统做成多租户 SaaS。每个方向都可以继续写出一系列实战文章。如果你正在做类似的企业级 AI 应用希望这份代码和设计思路能给你一个可用的起点。也欢迎在评论区聊聊你在合规项目里遇到过的坑。 AI合规跟踪器：从法规分析到任务闭环的实战指南
当一家初创公司从本地市场走向全球时最先遇到的可能不是业务增长问题而是合规问题。GDPR、CCPA、PIPL、LGPD、SOC 2……不同国家和地区的法规不断更新人工跟踪很容易出现遗漏。Veritas 就是为解决这个场景设计的一个基于 AI 的全球初创企业合规跟踪器用 AI 辅助法规分类、任务生成和风险预警同时保留人工审核入口避免“全自动踩雷”。下面完整拆解这个系统的设计思路、核心代码和落地注意事项。1. 项目背景与核心概念1.1 什么是合规跟踪器合规跟踪器是一类帮助企业识别、记录、跟踪和完成监管义务的软件系统。传统的合规管理通常依赖 Excel 表、邮件提醒和线下检查一旦企业进入多个司法管辖区法规数量会指数级增长。一家 SaaS 公司如果同时服务欧盟、美国和东南亚用户可能同时受到 GDPR、CCPA、PDPA 等多部法律约束而每部法律又包含数十条具体义务。合规跟踪器的作用就是把这些分散的法规要求整理成结构化任务分配给负责人并持续跟踪完成状态。Veritas 在这个基础上增加了 AI 环节。它不只是“法规库 任务列表”而是尝试用大模型自动解读新法规、判断适用范围、生成合规义务并给出风险等级建议。这样做的好处是降低人工整理成本尤其在法规文本很长、更新频繁的场景下AI 可以快速产出“初稿”再由合规人员审核确认。1.2 为什么选择 AI 做合规合规本质上是一个“信息抽取 规则匹配 任务管理”的过程。传统实现方式有两种一种是纯人工成本高、速度慢另一种是写死规则只能覆盖已知法规难以应对新法条。AI 的强项在于理解自然语言可以从长篇法律文本中提炼出“谁、在什么情况下、需要做什么、截止日期是什么”这类关键信息。不过 AI 也并非万能。大模型存在“幻觉”问题可能生成不存在的义务或错误的适用条件。因此 Veritas 在设计上把 AI 定位为“辅助引擎”而不是最终决策者。AI 生成的每一项义务都必须进入pending_review状态经过人工确认后才能变成正式任务。这个设计非常重要能避免企业把 AI 幻觉当成真实合规要求导致资源浪费或误导。1.3 Veritas 的功能边界Veritas 的核心功能包括四块公司信息管理、法规库管理、AI 合规分析、任务与风险跟踪。公司信息用于提供 AI 分析所需的上下文比如业务地区、所属行业、员工规模法规库用于存储原始法律文本或摘要AI 合规分析负责把法规与公司画像结合生成适用义务任务与风险跟踪用于把义务拆成具体待办事项并给出风险评分。在边界上Veritas 不会尝试“自动代扣罚款”或“自动提交申报文件”那是政府对接层的事。它更接近一个“合规工作台”帮助团队知道需要做什么、什么时候做、做到什么程度。下面我们从技术选型开始一步步实现这个系统。2. 环境准备与技术选型2.1 技术栈说明Veritas 后端采用 Python 生态核心依赖包括 FastAPI、SQLAlchemy、Pydantic 和一个可切换的 LLM 客户端。FastAPI 用于提供 REST API自带 Swagger 文档方便调试和对接前端SQLAlchemy 负责数据库读写支持 SQLite、PostgreSQL、MySQL 等常见数据库Pydantic 负责接口入参校验和响应序列化。LLM 客户端抽象成接口是项目的一个关键设计。开发阶段用 Mock 实现不需要真实模型也可以跑通整个流程生产环境可以切换到 OpenAI、国内大模型或其他私有化部署模型。这样既能快速演示又能避免把某个供应商写死在业务代码里。版本方面建议使用 Python 3.10 及以上、FastAPI 0.100 以上、SQLAlchemy 2.x。实际版本需要根据你的项目情况调整本文示例以常见环境为例重点演示配置思路而不是绑定某个固定版本。2.2 项目目录结构建议把项目拆成多个模块避免把所有代码塞进一个文件。目录结构如下veritas/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models.py │ ├── llm_client.py │ └── compliance_service.py ├── requirements.txt ├── .env └── README.mdmain.py是 FastAPI 入口负责创建应用和路由models.py定义数据库表结构llm_client.py封装不同的大模型客户端compliance_service.py是核心业务逻辑层负责把法规和公司信息变成合规义务与任务。2.3 环境变量与配置项目使用环境变量控制数据库地址和 LLM 提供方避免把密钥硬编码到代码里。.env文件示例如下DATABASE_URLsqlite:///./veritas.db LLM_PROVIDERmock OPENAI_API_KEYsk-xxx OPENAI_MODELgpt-4o-mini在本地开发阶段LLM_PROVIDER设置为mock不需要真实 API Key。接入生产模型时再改为openai并填入对应密钥。这里特别提醒任何真实密钥都只能放在服务端环境变量或密钥管理服务中绝不能提交到 Git 仓库否则一旦泄露就可能被恶意使用。3. 核心模块设计3.1 数据模型设计Veritas 的数据库模型包括五张表公司信息表company_profiles、法规表regulations、合规义务表compliance_obligations、合规任务表compliance_tasks和审计日志表audit_logs。公司表存储企业名称、目标市场、行业和员工数法规表存储法规标题、适用地区、生效日期、来源链接和全文义务表是 AI 分析结果包含义务摘要、适用范围、风险等级、AI 置信度和审核状态任务表把义务拆成可执行的待办审计日志表记录谁在什么时候做了什么操作满足企业内部审计需求。这里需要重点说明的是义务表上的唯一约束。AI 可能会被重复调用如果不加控制同一法规对公司可能生成大量重复义务。唯一约束(company_id, regulation_id, obligation_summary)可以从数据库层面兜底但更稳妥的做法是在业务层先查询已存在的记录避免重复分析。3.2 LLM 客户端抽象LLM 客户端抽象的核心是定义一个classify_regulation方法输入法规文本和公司信息输出结构化的合规分析结果。输出结果至少包含四类信息义务列表、适用范围、风险等级、建议截止日期和置信度。{ obligations: [用户提交数据前必须获得明确同意], scope: 所有面向欧盟用户的线上服务, risk_level: high, due_suggestion: 2025-12-31, confidence: 0.85 }risk_level是low、medium、high三选一confidence是 0 到 1 之间的小数表示模型对结果的把握程度。Mock 实现返回固定结果方便测试OpenAI 实现通过 prompt 让大模型输出 JSON再用json.loads解析。3.3 AI 合规分析引擎分析引擎是整个系统的业务核心它把公司画像和法规文本交给 LLM拿到结果后保存义务并创建任务。在analyze_regulation方法中第一步是从数据库读取公司信息和法规第二步构造公司信息字典第三步调用 LLM第四步解析日期和风险等级第五步创建ComplianceObligation记录第六步根据义务列表创建ComplianceTask最后写入审计日志。日期解析不能直接信任模型输出。模型可能返回2025/12/31、2025-12-31或者纯文本“三个月内”所以_parse_date方法需要做格式兼容解析失败就返回None由人工后来补充。3.4 任务生成与风险评分AI 返回的obligations是一个列表每一项都是一条独立义务。任务生成时把每条义务作为一条待办事项写入compliance_tasks标题可以用“落实合规义务 1”“落实合规义务 2”这类自动编号便于快速认领。风险评分采用简单加权模型。每个义务的risk_level映射为权重low 是 1medium 是 2high 是 3。企业所有已审核通过义务的权重相加得到一个最高 100 的风险分。分数越高说明当前需要处理的合规负担越大。这个评分不是法律意义上的风险评级而是帮助团队排定优先级的一个参考信号。4. 完整实战案例4.1 初始化项目与所需依赖先创建项目目录和虚拟环境然后准备依赖文件。在requirements.txt中写入以下内容fastapi uvicorn[standard] sqlalchemy pydantic pydantic-settings python-dotenv openai安装依赖pip install -r requirements.txt如果只需要跑 Mock 模式openai也可以先不装但为了方便后续切换建议一次性装好。4.2 配置环境变量在项目根目录创建.env文件DATABASE_URLsqlite:///./veritas.db LLM_PROVIDERmock OPENAI_API_KEY OPENAI_MODELgpt-4o-mini本地先用 SQLite不需要额外启动数据库服务。后续接入 PostgreSQL 时只需要修改DATABASE_URL例如DATABASE_URLpostgresqlpsycopg://user:passwordlocalhost:5432/veritas4.3 编写数据库模型 models.py文件路径app/models.pyfrom datetime import datetime from sqlalchemy import ( Column, Integer, String, DateTime, Date, Text, ForeignKey, Boolean, Float, UniqueConstraint, func ) from sqlalchemy.orm import declarative_base, relationship Base declarative_base() class CompanyProfile(Base): __tablename__ company_profiles id Column(Integer, primary_keyTrue, indexTrue) name Column(String(200), nullableFalse) region Column(String(20), nullableFalse) industry Column(String(100), nullableFalse) employee_count Column(Integer, default0) created_at Column(DateTime, server_defaultfunc.now()) obligations relationship(ComplianceObligation, back_populatescompany) class Regulation(Base): __tablename__ regulations id Column(Integer, primary_keyTrue, indexTrue) title Column(String(300), nullableFalse) region Column(String(20), nullableFalse) category Column(String(50), nullableTrue) effective_date Column(String(20), nullableTrue) source_url Column(String(500), nullableTrue) full_text Column(Text, nullableFalse) created_at Column(DateTime, server_defaultfunc.now()) obligations relationship(ComplianceObligation, back_populatesregulation) class ComplianceObligation(Base): __tablename__ compliance_obligations id Column(Integer, primary_keyTrue, indexTrue) company_id Column(Integer, ForeignKey(company_profiles.id), nullableFalse) regulation_id Column(Integer, ForeignKey(regulations.id), nullableFalse) obligation_summary Column(Text, nullableFalse) applicable_scope Column(String(500), nullableTrue) risk_level Column(String(20), defaultlow) status Column(String(20), defaultpending_review) ai_confidence Column(Float, default0.0) due_date Column(Date, nullableTrue) created_at Column(DateTime, server_defaultfunc.now()) updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) company relationship(CompanyProfile, back_populatesobligations) regulation relationship(Regulation, back_populatesobligations) tasks relationship(ComplianceTask, back_populatesobligation) __table_args__ ( UniqueConstraint( company_id, regulation_id, obligation_summary, nameuniq_obligation ), ) class ComplianceTask(Base): __tablename__ compliance_tasks id Column(Integer, primary_keyTrue, indexTrue) obligation_id Column(Integer, ForeignKey(compliance_obligations.id), nullableFalse) title Column(String(300), nullableFalse) description Column(Text, nullableTrue) status Column(String(20), defaulttodo) assignee Column(String(100), nullableTrue) due_date Column(Date, nullableTrue) completed_at Column(DateTime, nullableTrue) created_at Column(DateTime, server_defaultfunc.now()) obligation relationship(ComplianceObligation, back_populatestasks) class AuditLog(Base): __tablename__ audit_logs id Column(Integer, primary_keyTrue, indexTrue) entity_type Column(String(50), nullableFalse) entity_id Column(Integer, nullableFalse) action Column(String(100), nullableFalse) detail Column(Text, nullableTrue) created_at Column(DateTime, server_defaultfunc.now())在这个模型中AI_confidence字段用来记录模型输出置信度status字段先设为pending_review等待人工审核。如果后续要增加“被驳回”状态可以继续扩展状态枚举。每张表都带created_at方便追溯数据产生时间。4.4 编写 LLM 客户端 llm_client.py文件路径app/llm_client.pyimport json from abc import ABC, abstractmethod from openai import OpenAI class LLMClient(ABC): abstractmethod def classify_regulation(self, regulation_text: str, company_info: dict) - dict: 返回示例 { obligations: [..., ...], scope: ..., risk_level: low|medium|high, due_suggestion: 2025-12-31, confidence: 0.85 } ... class MockLLMClient(LLMClient): def classify_regulation(self, regulation_text: str, company_info: dict) - dict: return { obligations: [ 根据用户所在地区提供隐私政策, 用户行使删除权时需在30天内响应, ], scope: 面向目标市场的线上服务, risk_level: medium, due_suggestion: 2025-12-31, confidence: 0.75, } class OpenAILLMClient(LLMClient): def __init__(self, api_key: str, model: str gpt-4o-mini): self.client OpenAI(api_keyapi_key) self.model model def classify_regulation(self, regulation_text: str, company_info: dict) - dict: prompt self._build_prompt(regulation_text, company_info) response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是合规分析助手。请只输出JSON不要输出多余内容。}, {role: user, content: prompt}, ], temperature0, ) content response.choices[0].message.content return json.loads(self._clean_json(content)) def _build_prompt(self, regulation_text: str, company_info: dict) - str: company_desc ( f公司名称{company_info[name]}目标市场{company_info[region]} f所属行业{company_info[industry]}员工数{company_info[employee_count]} ) return ( 请根据公司信息分析以下法规输出适用义务、范围、风险等级、建议截止日期、置信度。\n f公司信息{company_desc}\n f法规正文\n{regulation_text[:4000]} ) def _clean_json(self, text: str) - str: text text.strip() if text.startswith(json): text text[7:] elif text.startswith(): text text[3:] if text.endswith(): text text[:-3] return text.strip()这里有两个关键点。第一LLMClient是抽象基类业务代码只依赖这个抽象不依赖具体供应商。第二_clean_json用来去除模型输出中可能存在的 Markdown 代码块标记否则json.loads会直接报错。实际使用时不同模型可能输出不同的包裹格式需要针对模型调优。4.5 编写合规分析服务 compliance_service.py文件路径app/compliance_service.pyfrom datetime import date from sqlalchemy.orm import Session from .models import ( CompanyProfile, Regulation, ComplianceObligation, ComplianceTask, AuditLog, ) from .llm_client import LLMClient RISK_WEIGHT {low: 1, medium: 2, high: 3} class ComplianceService: def __init__(self, db: Session, llm: LLMClient): self.db db self.llm llm def analyze_regulation(self, company_id: int, regulation_id: int) - ComplianceObligation: company self.db.get(CompanyProfile, company_id) regulation self.db.get(Regulation, regulation_id) if not company or not regulation: raise ValueError(company or regulation not found) company_info { name: company.name, region: company.region, industry: company.industry, employee_count: company.employee_count, } result self.llm.classify_regulation(regulation.full_text, company_info) due_date self._parse_date(result.get(due_suggestion)) risk_level self._normalize_risk(result.get(risk_level, medium)) obligation ComplianceObligation( company_idcompany.id, regulation_idregulation.id, obligation_summary\n.join(result.get(obligations, [])), applicable_scoperesult.get(scope), risk_levelrisk_level, statuspending_review, ai_confidencefloat(result.get(confidence, 0.0)), due_datedue_date, ) self.db.add(obligation) self.db.flush() self._create_tasks(obligation, result.get(obligations, []), due_date) self._add_audit_log( compliance_obligation, obligation.id, create, fAI生成义务待人工审核置信度{obligation.ai_confidence}, ) self.db.commit() self.db.refresh(obligation) return obligation def _create_tasks(self, obligation, obligations, due_date): for idx, obligation_text in enumerate(obligations, start1): task ComplianceTask( obligation_idobligation.id, titlef落实合规义务 {idx}, descriptionobligation_text, statustodo, due_datedue_date, ) self.db.add(task) def approve_obligation(self, obligation_id: int) - ComplianceObligation: obligation self.db.get(ComplianceObligation, obligation_id) if not obligation: raise ValueError(obligation not found) obligation.status approved self._add_audit_log( compliance_obligation, obligation_id, approve, 人工审核通过, ) self.db.commit() self.db.refresh(obligation) return obligation def calculate_risk_score(self, company_id: int) - int: obligations self.db.query(ComplianceObligation).filter( ComplianceObligation.company_id company_id, ComplianceObligation.status approved, ).all() if not obligations: return 0 total 0 for obligation in obligations: total RISK_WEIGHT.get(obligation.risk_level, 0) return min(total, 100) def _normalize_risk(self, value) - str: value (value or ).lower() if value in (high, 高风险): return high if value in (low, 低风险): return low return medium def _parse_date(self, value): if not value: return None try: return date.fromisoformat(value) except ValueError: return None def _add_audit_log(self, entity_type, entity_id, action, detail): log AuditLog( entity_typeentity_type, entity_identity_id, actionaction, detaildetail, ) self.db.add(log)这段代码的核心是analyze_regulation。它首先从数据库取公司信息和法规然后把它们传给 LLM拿到 JSON 结果后写入数据库。这里没有直接信任模型返回的所有字段而是做了标准化处理风险等级统一转成三档日期解析失败则留空置信度强制转成浮点数。这样即使不同模型返回格式有细微差异也不会导致系统崩溃。4.6 编写 FastAPI 入口 main.py文件路径app/main.pyimport os from datetime import date from typing import List from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from .models import ( Base, CompanyProfile, Regulation, ComplianceObligation, ComplianceTask, ) from .llm_client import MockLLMClient, OpenAILLMClient from .compliance_service import ComplianceService DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./veritas.db) engine create_engine(DATABASE_URL, connect_args{check_same_thread: False}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base.metadata.create_all(bindengine) app FastAPI(titleVeritas Compliance Tracker, version0.1.0) def get_db(): db SessionLocal() try: yield db finally: db.close() def get_llm(): provider os.getenv(LLM_PROVIDER, mock).lower() if provider openai: return OpenAILLMClient( api_keyos.getenv(OPENAI_API_KEY, ), modelos.getenv(OPENAI_MODEL, gpt-4o-mini), ) return MockLLMClient() class CompanyCreate(BaseModel): name: str region: str industry: str employee_count: int 0 class CompanyOut(BaseModel): id: int name: str region: str industry: str employee_count: int class Config: from_attributes True class RegulationCreate(BaseModel): title: str Field(..., description法规标题) region: str Field(..., description适用地区例如 EU/US) category: str | None None effective_date: str | None None source_url: str | None None full_text: str Field(..., min_length50, description法规全文或关键摘要) class RegulationOut(BaseModel): id: int title: str region: str category: str | None effective_date: str | None source_url: str | None full_text: str class Config: from_attributes True class TaskOut(BaseModel): id: int obligation_id: int title: str description: str status: str assignee: str | None due_date: str | None completed_at: str | None class Config: from_attributes True app.post(/companies, response_modelCompanyOut, status_code201) def create_company(payload: CompanyCreate, db: Session Depends(get_db)): company CompanyProfile( namepayload.name, regionpayload.region, industrypayload.industry, employee_countpayload.employee_count, ) db.add(company) db.commit() db.refresh(company) return company app.post(/regulations, response_modelRegulationOut, status_code201) def create_regulation(payload: RegulationCreate, db: Session Depends(get_db)): regulation Regulation( titlepayload.title, regionpayload.region, categorypayload.category, effective_datepayload.effective_date, source_urlpayload.source_url, full_textpayload.full_text, ) db.add(regulation) db.commit() db.refresh(regulation) return regulation app.post(/analyze/{regulation_id}) def analyze(regulation_id: int, company_id: int, db: Session Depends(get_db)): llm get_llm() service ComplianceService(db, llm) try: obligation service.analyze_regulation(company_id, regulation_id) except ValueError as e: raise HTTPException(status_code404, detailstr(e)) return { obligation_id: obligation.id, obligation_summary: obligation.obligation_summary, risk_level: obligation.risk_level, status: obligation.status, ai_confidence: obligation.ai_confidence, due_date: obligation.due_date.isoformat() if obligation.due_date else None, } app.post(/obligations/{obligation_id}/approve) def approve(obligation_id: int, db: Session Depends(get_db)): service ComplianceService(db, get_llm()) try: obligation service.approve_obligation(obligation_id) except ValueError as e: raise HTTPException(status_code404, detailstr(e)) return {id: obligation.id, status: obligation.status} app.get(/companies/{company_id}/tasks, response_modelList[TaskOut]) def list_tasks(company_id: int, db: Session Depends(get_db)): obligations db.query(ComplianceObligation).filter( ComplianceObligation.company_id company_id ).all() obligation_ids [o.id for o in obligations] if not obligation_ids: return [] tasks db.query(ComplianceTask).filter( ComplianceTask.obligation_id.in_(obligation_ids) ).all() return tasksmain.py中需要注意两点。第一Base.metadata.create_all(bindengine)在开发阶段会自动建表生产环境更推荐使用 Alembic 做迁移避免表结构变更导致的数据丢失。第二get_llm()根据环境变量切换模型客户端这个模式可以继续扩展成工厂函数比如根据请求头或租户信息选择不同模型。4.7 运行与验证在项目根目录执行启动命令uvicorn app.main:app --reload启动成功后访问http://127.0.0.1:8000/docs能看到 Swagger 接口文档。接下来按顺序创建公司、创建法规、执行分析。先创建一家公司curl -X POST http://127.0.0.1:8000/companies \ -H Content-Type: application/json \ -d { name: Example AI Startup, region: EU, industry: SaaS, employee_count: 120 }预期返回包含公司 id 的 JSON。再创建一条法规curl -X POST http://127.0.0.1:8000/regulations \ -H Content-Type: application/json \ -d { title: GDPR Privacy Policy Requirements, region: EU, category: privacy, effective_date: 2025-01-01, source_url: https://example.com/gdpr-summary, full_text: 企业收集欧盟用户个人数据前应明确告知数据处理目的、保存期限、用户权利等信息。用户有权请求删除个人数据企业应在合理时间内响应。 }然后调用分析接口把法规 id 和公司 id 传入curl -X POST http://127.0.0.1:8000/analyze/1?company_id1Mock 模式返回的结果中义务状态会是pending_review置信度为0.75。接着可以查看该公司所有任务curl http://127.0.0.1:8000/companies/1/tasks通过 Swagger 页面可以直接点击执行普通团队成员也能快速学会使用。5. 常见问题与排查思路5.1 高频问题速查表问题现象常见原因解决思路AI 分类结果不稳定Prompt 设计不够明确模型随机性高使用结构化输出要求设置temperature0增加 few-shot 示例法规文本过长超出模型上下文输入整篇法规全文预分割文本或引入 RAG 只检索关键条款同一法规反复生成重复义务缺少幂等控制查询已有记录增加唯一约束json.loads解析失败模型输出包含 Markdown 代码块用_clean_json剥离代码块标记并做异常兜底风险等级无法排序字符串格式不统一在业务层标准化为low/medium/high生产环境 API Key 泄露密钥写入代码或日志使用环境变量日志脱敏配置最小权限5.2 排查流程建议如果 AI 分析结果不理想不要先改代码先看原始输入输出。可以写一个简单的 Python 脚本直接调用llm.classify_regulation打印完整 prompt 和模型返回值确认是不是模型本身的问题。确认模型输出没问题后再检查解析层和数据库层。如果任务创建成功但查询不到优先检查数据库连接配置是否指向正确的库。SQLite 默认在当前目录如果启动命令的工作目录不一致可能会出现“表存在但数据看不到”的情况。建议在接口里加一个日志输出记录当前数据库路径和创建记录数。6. 最佳实践与工程建议6.1 AI 输出质量控制AI 在法律场景下的输出必须保持克制。建议在系统提示词中明确要求“只输出 JSON”并用 few-shot 示例约束格式。同时在业务层增加置信度阈值当ai_confidence低于 0.6 时强制进入pending_review并标记为“低置信度建议优先人工检查”。这样即使用户没有主动看状态团队也会注意到高风险记录。大模型采用temperature0可以降低随机性但无法完全消除幻觉。更可靠的方式是引入检索增强生成把法规文本分词后存入向量数据库分析时只检索与公司相关的条款片段减少过长输入带来的注意力分散。这是 Veritas 下一步可以优化的关键方向。6.2 数据安全与隐私合规系统本身处理的就是敏感数据尤其是法规原文和公司内部义务信息。所有接口必须加认证和授权不能直接暴露在公网。生产环境建议使用 OAuth2、API Key 或企业内部 SSO。涉及用户个人数据的场景还要注意数据最小化原则只采集服务所需字段避免把无关个人信息发送给大模型。调用外部大模型时要评估数据是否会被供应商留存。如果公司信息或法规文本属于敏感内容建议采用私有化部署模型或在发送前做字段脱敏。日志记录方面不能把完整 prompt 和模型原始输出原样打到日志里必要时应截断或替换敏感字段。6.3 合规任务生命周期Veritas 目前只实现了最简单的任务状态todo、done通过completed_at判断。真实生产项目建议引入状态机包含todo、in_progress、in_review、done、cancelled等状态。每次状态变更都要写入审计日志并记录操作人、操作时间和变更原因。截止日期管理也非常重要。AI 建议的due_date只是预估需要由业务负责人确认。可以增加定时任务在截止日期前 7 天、3 天、1 天发送提醒避免遗漏。生产环境需要额外考虑日期展示时区建议统一存储为 UTC展示时再转换到用户本地时区。6.4 可扩展性考虑如果未来要支持更多国家法规不建议把所有法规全文都塞进数据库后丢给模型。更合理的做法是引入两层结构第一层是法规元数据包括标题、适用地区、分类第二层是义务详情每一条义务单独存储便于检索和对比。法规语义检索可以用向量数据库也可以用传统全文索引根据团队能力和数据量决定。对多租户支持敏感的 SaaS 场景数据模型需要增加tenant_id字段所有查询都按租户隔离。当前示例没有这个字段开发者可以在CompanyProfile上增加租户概念并在路由层强制传入。7. 总结与下一步学习方向这篇文章围绕 Veritas 这个 AI 驱动的全球初创企业合规跟踪器从需求背景、技术选型、数据库设计、LLM 抽象、合规分析引擎到 FastAPI 接口完成了一个可运行的 MVP。可以看到AI 在这类系统里的核心价值是“快速起草初稿”而不是替代专业人士做判断。真正的可靠来自两件事一是把模型输出标准化二是让所有结果都经过人工复核。下一步建议从三个方向继续深入。第一学习使用向量数据库和 Embedding 模型做法规语义检索解决长文本分析不准确的问题第二尝试接入定时任务和消息通知让任务闭环更完整第三引入更细粒度的权限模型把系统做成多租户 SaaS。每个方向都可以继续写出一系列实战文章。如果你正在做类似的企业级 AI 应用希望这份代码和设计思路能给你一个可用的起点。也欢迎在评论区聊聊你在合规项目里遇到过的坑。