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 抽取结果、分派任务。
• 团队成员：查看分配给自己的义务，提交完成证明，更新状态。

核心业务流如下：

1. 合规负责人导入一份法规文本，可以是 PDF、Word 或纯文本。
2. 后端调用 AI 服务，把文本拆成结构化条款。
3. AI 对每条条款进行归类，识别出“义务动作”“适用对象”“时间期限”“风险等级”。
4. 合规负责人逐条确认或修改 AI 结果。
5. 确认后的数据生成一条义务记录，并自动派生一个或多个任务。
6. 任务分配负责人后，系统在截止日期前发送提醒。
7. 负责人完成任务后，系统记录完成人、完成时间、证明材料。
8. 所有关键操作写入审计日志。

2.2 模块拆分

按照上面的流程，Veritas 可以拆成以下模块：

模块	职责
规则管理	法规文本上传、版本管理、条款结构维护
AI 抽取服务	调用大模型提取结构化信息
义务管理	管理合规义务，维护风险等级、责任方、到期时间
任务管理	任务的创建、分配、完成、延期
提醒通知	按截止时间发送邮件或站内通知
审计日志	记录用户操作和 AI 行为，支持追溯
用户与权限	角色管理，控制不同用户可见的数据范围

为了控制 MVP 的复杂度，暂时不做复杂的权限矩阵，只实现管理员和普通用户两个角色。

2.3 技术栈选型

对于 MVP，我推荐以下技术栈：

• 后端框架：Python + FastAPI，开发效率高，天然支持异步，自带 OpenAPI 文档。
• 数据库：PostgreSQL，事务支持好，适合业务系统；也可以先用 SQLite 快速验证。
• ORM：SQLAlchemy 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_uuid=True), primary_key=True, default=uuid.uuid4 ) created_at: Mapped[datetime] = mapped_column( DateTime, default=datetime.utcnow ) class User(Base): __tablename__ = "users" username: Mapped[str] = mapped_column(String(50), unique=True, index=True) email: Mapped[str] = mapped_column(String(120), unique=True, index=True) role: Mapped[str] = mapped_column(String(20), default="user") is_active: Mapped[bool] = mapped_column(Boolean, default=True) class ComplianceRule(Base): __tablename__ = "compliance_rules" title: Mapped[str] = mapped_column(String(200)) rule_code: Mapped[str] = mapped_column(String(100), unique=True, index=True) jurisdiction: Mapped[str] = mapped_column(String(50), index=True) published_date: Mapped[datetime] = mapped_column(DateTime, nullable=True) source_url: Mapped[str] = mapped_column(String(500), default="") version: Mapped[str] = mapped_column(String(20), default="1.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, default=dict) human_confirmed: Mapped[bool] = mapped_column(Boolean, default=False) 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), default="medium") due_type: Mapped[str] = mapped_column(String(20), default="one_time") due_days: Mapped[int] = mapped_column(Integer, default=30) owner_id: Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=True) 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"), nullable=True) due_date: Mapped[datetime] = mapped_column(DateTime) status: Mapped[str] = mapped_column(String(20), default="pending") completed_at: Mapped[datetime] = mapped_column(DateTime, nullable=True) 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, default=datetime.utcnow) class AuditLog(Base): __tablename__ = "audit_logs" user_id: Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=True) 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, default=dict) 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": f"Bearer {self.api_key}", "Content-Type": "application/json" } resp = requests.post( f"{self.api_base}/chat/completions", headers=headers, json=payload, timeout=60 ) 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.md

5.2 requirements.txt

fastapi==0.111.0 uvicorn[standard]==0.30.1 sqlalchemy==2.0.30 psycopg2-binary==2.9.9 python-dotenv==1.0.1 requests==2.32.3 pydantic==2.7.4

如果你本地还没有 PostgreSQL，可以直接使用 Docker Compose 启动一个。

5.3 docker-compose.yml

version: "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", "postgresql+psycopg2://veritas:veritas@localhost: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, echo=False) SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) class Base(DeclarativeBase): pass

5.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 = True

5.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(bind=engine) app = FastAPI(title="Veritas Compliance Tracker", version="0.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( title=rule_title, rule_code=rule_code, jurisdiction=jurisdiction, ) db.add(rule) db.flush() client = AIClient( api_base=settings.AI_API_BASE, api_key=settings.AI_API_KEY, model=settings.AI_MODEL ) try: clauses = client.analyze_text(raw_text) except Exception as e: raise HTTPException(status_code=500, detail=f"AI 抽取失败: {str(e)}") created_clauses = [] for item in clauses: clause = ComplianceClause( rule_id=rule.id, clause_number=item.get("clause_number", "0"), original_text=raw_text[:2000], ai_extracted=item, human_confirmed=False, ) db.add(clause) db.flush() obligation = ComplianceObligation( clause_id=clause.id, title=item.get("summary", "")[:200], description=item.get("required_action", ""), required_action=item.get("required_action", ""), applicable_object=item.get("applicable_object", ""), risk_level=item.get("risk_level", "medium"), due_type=item.get("due_type", "one_time"), due_days=item.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_model=list[ClauseOut]) def list_clauses(db: Session = Depends(get_db)): return db.query(ComplianceClause).all() @app.get("/obligations", response_model=list[ObligationOut]) def list_obligations(db: Session = Depends(get_db)): return db.query(ComplianceObligation).all() @app.post("/tasks", response_model=TaskOut) 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_model=list[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. 常见问题与排查思路

问题现象	常见原因	解决思路
导入规则时提示 500	AI 服务地址不可达或 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 跑起来，再逐步迭代。现在就可以打开终端，把代码下载下来试一遍，亲手体验一次法规文本到任务清单的转化过程。