基于AI Agent与RAG的文档合规智能评估系统设计与实现

1. 项目概述：一个面向文档合规与安全评估的智能代理

最近在折腾一个挺有意思的项目，叫 DocSentinel。简单来说，它是一个基于 AI Agent 架构的文档合规与安全评估工具。想象一下，你手头有一堆合同、报告或者政策文件，需要快速判断里面有没有敏感信息、是否符合特定法规要求，或者是否存在潜在的安全风险。传统做法要么是人工逐字逐句审阅，效率低下且容易遗漏；要么是依赖一些简单的关键词匹配工具，准确率堪忧，误报率高。DocSentinel 就是为了解决这个痛点而生的。

它的核心思路，是把当下流行的几项技术——大语言模型、检索增强生成、智能体框架以及文档解析——给“攒”到了一起，形成一个能自动理解文档内容、调用专业工具进行分析、并给出结构化评估报告的智能工作流。这玩意儿特别适合法务、风控、安全审计以及需要处理大量外部文档的运营团队。我自己在搭建和测试的过程中，发现它不仅能显著提升文档审查的效率，更重要的是，通过可配置的规则和模型，能让评估过程更加标准化和可追溯，避免了人为判断的主观性偏差。

2. 核心架构与技术选型解析

DocSentinel 不是一个单一的工具，而是一个由多个组件协同工作的系统。理解它的架构，是后续进行部署、定制和优化的基础。

2.1 总体架构与数据流

整个系统的运行遵循一个清晰的管道式工作流。首先，用户通过一个 Web 界面或 API 上传文档。系统后端接收到文档后，启动处理流水线。第一步是文档解析，将 PDF、Word、Excel 等不同格式的文档，转换成纯文本或结构化的数据。这是所有后续分析的基础，解析的准确性直接决定了最终结果的质量。

解析后的文本会进入检索增强生成环节。系统会从一个预设的、不断更新的知识库中，检索与当前文档内容相关的合规条款、安全标准或风险案例。这个知识库可能包含了 GDPR、HIPAA 等法规的核心要求，或者公司内部的保密政策。RAG 的作用是为 LLM 提供精准的上下文，让它不是凭空想象，而是基于事实和规则进行判断。

接下来，大语言模型作为“大脑”登场。它综合原始文档内容和 RAG 提供的相关知识，执行具体的分析任务。例如，判断某个条款是否违反了数据最小化原则，或者某段描述是否包含了不应公开的技术细节。这里的关键是，LLM 并不直接输出“是”或“否”，而是遵循一套预先定义好的结构化输出格式，比如 JSON，里面包含了风险点、置信度、原文引用和修改建议。

最后，一个智能体框架负责协调整个流程。它决定在哪个环节调用哪个工具（解析器、检索器、LLM），如何处理异常（比如解析失败），以及如何将 LLM 的输出整理成最终的人类可读报告。整个系统的状态、任务队列和结果都会通过 API 暴露出来，方便集成到其他系统或由前端界面展示。

2.2 关键技术组件选型理由

为什么选择这些技术栈？每个选择背后都有具体的考量。

后端框架：FastAPI选用 FastAPI 而非 Django 或 Flask，主要基于性能和开发效率的权衡。DocSentinel 的核心是异步处理文档分析任务，这些任务往往是 I/O 密集型的（等待模型推理、等待数据库查询）。FastAPI 原生支持async/await，能够轻松构建高性能的异步 API，完美匹配这种场景。此外，它自动生成的交互式 API 文档对前后端协作非常友好，强类型提示也让代码更健壮，减少运行时错误。

大语言模型服务：Ollama 与 OpenAI API 并存模型层采用了混合策略。本地部署使用Ollama，主要原因在于数据安全和可控性。处理合规与安全文档，内容可能高度敏感，不适合传出内网。Ollama 允许我们在自己的服务器上运行如 Llama 3、Mistral 等开源模型，虽然推理速度可能稍慢，但保证了数据的物理隔离。对于对延迟敏感或需要最强分析能力的场景，则集成OpenAI API（如 GPT-4）。这种设计提供了灵活性：在开发测试阶段或用小规模数据验证时用 Ollama；在生产环境处理复杂文档时，可按需切换到云端大模型。关键在于，系统通过统一的接口抽象了模型调用，切换模型提供商几乎不需要改动业务代码。

智能体协调：MCP 与 OpenClaw 的考量智能体框架的选择比较有趣。MCP是一个新兴的智能体协作协议，它强调标准化智能体之间的通信和工具调用。如果 DocSentinel 的目标是成为一个能与其他 AI 智能体（比如一个自动合同谈判智能体）协同工作的平台组件，那么采用 MCP 会很有前瞻性，有利于生态集成。而OpenClaw更像一个一体化的智能体应用开发框架，提供了从任务规划、工具调用到记忆管理的完整工具箱。如果我们的重点是快速构建一个功能完整、独立运行的文档分析智能体，OpenClaw 可能上手更快。在项目初期，我建议从 OpenClaw 开始，快速验证核心流程；当需要更复杂的多智能体交互时，再考虑引入 MCP 协议进行重构。

文档解析与向量检索文档解析没有采用单一的库，而是根据格式组合使用。PDF 用pdfplumber或PyMuPDF，它们对表格和复杂版式的处理更精准；Office 文档用python-docx和openpyxl。解析后的文本经过分块和清洗，使用如sentence-transformers生成向量，存入ChromaDB或Qdrant这类轻量级向量数据库供 RAG 检索。选择它们是因为其易于集成、性能不错，且与 Python 生态结合紧密。

注意：文档解析是整个流程中最容易出错的环节。特别是扫描版 PDF，依赖 OCR 质量。在实际项目中，需要为解析模块设计完善的异常处理和回退机制，比如当自动解析失败时，触发人工审核流程，而不是让整个任务静默失败。

3. 核心功能模块深度实现

了解了架构，我们深入到各个核心模块，看看它们具体是如何实现的，以及有哪些实操中的“坑”。

3.1 文档解析与预处理流水线

文档解析不是简单调用一个库就完事了，它需要一个健壮的流水线。

import pdfplumber from docx import Document import logging from typing import Optional, Dict, Any class DocumentParser: def __init__(self): self.logger = logging.getLogger(__name__) def parse(self, file_path: str, file_type: str) -> Optional[Dict[str, Any]]: """解析文档，返回结构化的文本和元数据。""" try: if file_type == 'application/pdf': return self._parse_pdf(file_path) elif file_type in ['application/vnd.openxmlformats-officedocument.wordprocessingml.document', 'application/msword']: return self._parse_docx(file_path) elif file_type in ['application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'application/vnd.ms-excel']: return self._parse_excel(file_path) else: self.logger.warning(f"Unsupported file type: {file_type}") return None except Exception as e: self.logger.error(f"Failed to parse {file_path}: {e}", exc_info=True) # 这里可以接入告警系统，通知管理员处理解析失败的文件 return None def _parse_pdf(self, file_path: str) -> Dict[str, Any]: """解析PDF，优先提取文本，处理表格。""" full_text = [] tables_data = [] with pdfplumber.open(file_path) as pdf: for page_num, page in enumerate(pdf.pages): # 提取文本 text = page.extract_text() if text: full_text.append(f"--- Page {page_num+1} ---\n{text}") # 提取表格 tables = page.extract_tables() for table in tables: if table: # 过滤空表 # 将表格转换为Markdown格式字符串，便于后续处理 md_table = self._table_to_markdown(table) tables_data.append(md_table) return { "raw_text": "\n".join(full_text), "tables": tables_data, "metadata": {"page_count": len(pdf.pages)} } def _table_to_markdown(self, table): """将二维列表表格转换为Markdown格式字符串。""" if not table: return "" header = table[0] rows = table[1:] # 简单实现，实际中需处理单元格内换行等复杂情况 md_lines = ["| " + " | ".join(str(cell) for cell in header) + " |"] md_lines.append("|" + "---|" * len(header)) for row in rows: md_lines.append("| " + " | ".join(str(cell) for cell in row) + " |") return "\n".join(md_lines)

实操要点与避坑指南：

1. 编码与清洗：解析出的文本经常包含多余的空格、换行符和乱码。必须进行标准化清洗，比如使用unicodedata.normalize('NFKC', text)处理 Unicode，并用正则表达式移除无意义的空白字符。
2. 分块策略：清洗后的长文本不能直接丢给 LLM 或向量化。需要根据语义进行分块。简单的按固定字符数分块会切断句子或段落。更好的方法是使用递归字符文本分割器，优先在段落、句子边界处切割，并设置合理的重叠度（例如 100-200 字符），以保证上下文的连贯性。
3. 表格处理：表格是文档的信息富集区，但也是最难解析的部分。pdfplumber的表格提取并非百分百准确，特别是对于合并单元格、虚线边框的表格。一个关键技巧是：在解析后，增加一个“表格验证”步骤，比如检查提取出的表格行列数是否合理，或者用一个小型规则模型判断提取出的内容是否像表格数据。对于极其重要的表格，可以考虑在 UI 上提供“表格预览与修正”功能，允许用户微调。
4. 元数据保留：解析时务必保留页码、章节标题（如果可识别）、字体大小等元数据。这些信息在后续的风险定位和报告生成中至关重要，能让你快速告诉用户“风险出现在第5页的‘责任条款’部分”。

3.2 检索增强生成与知识库构建

RAG 模块的质量，直接决定了 AI 分析的依据是否可靠。

知识库构建流程：

1. 数据源：收集所有相关的合规文档、安全标准、历史审计报告、风险案例库。这些可以是 PDF、网页（需爬取并清理）、内部 Wiki 页面。
2. 加载与解析：使用上述文档解析器处理这些源文件。
3. 文本分块：对解析出的文本进行分块。对于法规条文，可以按“条”、“款”进行分块，保持法律条文的完整性。
4. 向量化：使用嵌入模型将文本块转换为向量。这里的选择很重要。对于中文合规文档，text2vec、BGE系列模型是很好的选择。如果涉及多语言，multilingual-e5模型可能更合适。关键是要在你自己的领域数据上评估嵌入模型的效果，而不是盲目选择榜单第一名。
5. 存储：将向量和对应的文本块、元数据（如出自哪部法规第几条）存入向量数据库。

检索与生成过程：当分析一份新文档时：

1. 将文档分块，并为每个块生成向量。
2. 对于每个文档块，在知识库中进行相似度搜索，召回 Top-K 个最相关的知识片段。
3. 将文档块和召回的知识片段一起构建提示词，发送给 LLM 进行分析。

# 简化的RAG查询示例 from sentence_transformers import SentenceTransformer import chromadb class RAGRetriever: def __init__(self, model_name='BAAI/bge-large-zh', collection_name="compliance_knowledge"): self.embedder = SentenceTransformer(model_name) self.client = chromadb.PersistentClient(path="./chroma_db") self.collection = self.client.get_or_create_collection(collection_name) def query(self, document_chunk: str, top_k: int = 3) -> list: # 生成查询向量 query_embedding = self.embedder.encode(document_chunk).tolist() # 检索 results = self.collection.query( query_embeddings=[query_embedding], n_results=top_k ) # 组装上下文 contexts = [] if results['documents']: for doc, meta in zip(results['documents'][0], results['metadatas'][0]): contexts.append(f"[来源: {meta.get('source', 'Unknown')}] {doc}") return contexts

核心经验：

• 检索不是越多越好：top_k通常设置在 3-5 之间。过多的无关上下文会干扰 LLM 判断，增加成本并可能降低准确性。
• 重排序：简单的向量相似度检索可能不够精准。可以引入一个轻量级的“交叉编码器”模型对召回的片段进行重排序，选出与问题最相关的 1-2 个片段，这能显著提升效果。
• 知识库更新：合规规则是动态变化的。必须建立知识库的定期更新机制，可以设置为每周或每月自动从权威源抓取最新文档，解析后增量更新向量库。

3.3 基于LLM的智能分析与报告生成

这是 DocSentinel 的“大脑”。我们设计了一套提示词工程和输出规范，来引导 LLM 进行标准化分析。

分析提示词设计：提示词必须清晰定义角色、任务、输入和输出格式。

你是一个专业的合规与安全审计专家。你的任务是分析给定的文档片段，判断其是否存在合规或安全风险。 请严格遵循以下步骤进行分析： 1. 理解内容：仔细阅读“待分析文档”和提供的“相关合规知识”。 2. 风险识别：对照知识，判断文档内容是否存在潜在风险。风险类型可能包括：数据泄露风险、条款违规风险、表述模糊风险、权限过度风险等。 3. 评估与引用：如果存在风险，请明确指出： - 风险类型： - 风险描述： - 置信度（高/中/低）： - 在文档片段中的具体引用（原文）： - 依据的合规条款（来自提供的知识）： - 修改建议： 4. 如果不存在明显风险，则输出：{"risks": []} 请以纯JSON格式输出，格式如下： { "risks": [ { "risk_type": "...", "description": "...", "confidence": "high|medium|low", "quote_from_doc": "...", "reference_from_knowledge": "...", "suggestion": "..." } ] } 待分析文档： {document_chunk} 相关合规知识： {retrieved_context}

模型调用与后处理：

import openai from ollama import chat import json class RiskAnalyzer: def __init__(self, model_provider="ollama", model_name="llama3:latest"): self.provider = model_provider self.model_name = model_name def analyze_chunk(self, document_chunk: str, context: list) -> dict: prompt = self._build_prompt(document_chunk, context) try: if self.provider == "openai": response = openai.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.1, # 低温度保证输出稳定性 response_format={"type": "json_object"} # 强制JSON输出 ) result_text = response.choices[0].message.content else: # ollama response = chat(model=self.model_name, messages=[ {"role": "user", "content": prompt} ]) result_text = response['message']['content'] # 尝试解析JSON result = json.loads(result_text) # 验证结果结构 if "risks" not in result or not isinstance(result["risks"], list): raise ValueError("Invalid response structure from LLM") return result except json.JSONDecodeError as e: self.logger.error(f"Failed to parse LLM JSON response: {e}. Raw text: {result_text[:500]}") # 可以尝试用正则表达式从错误响应中提取JSON，或者返回空结果 return {"risks": []} except Exception as e: self.logger.error(f"LLM analysis failed: {e}") return {"risks": []}

关键注意事项：

1. 输出格式控制：LLM 的“幻觉”和格式不稳定性是最大挑战。除了在提示词中明确要求 JSON，使用 OpenAI 的response_format参数或 Ollama 的相应功能（如果有）来强制 JSON 输出。即使这样，解析失败的情况仍需处理，代码中必须有健壮的try-catch和降级策略。
2. 置信度校准：LLM 给出的“高/中/低”置信度是主观的。在真实系统中，需要结合其他信号来综合判断。例如，可以记录模型生成该结果时的“对数概率”作为内部置信度分数，或者设计一个“验证器”模型对高风险结果进行二次校验。
3. 成本与延迟优化：分析长文档时，串行处理每个分块速度慢。可以将文档分块后，并发调用模型 API（注意 API 速率限制）。同时，可以设计一个“过滤器”：先用一个简单规则或小模型快速扫描，只将疑似有问题的分块发送给大模型进行深度分析，这能大幅降低成本。

3.4 智能体协调与任务编排

智能体框架负责将以上所有模块串联成一个自动化工作流。以 OpenClaw 的思路为例，我们可以定义一个主智能体。

# 概念性代码，展示智能体决策逻辑 class DocSentinelAgent: def __init__(self, parser, retriever, analyzer, report_gen): self.parser = parser self.retriever = retriever self.analyzer = analyzer self.report_generator = report_gen def process_document(self, file_path: str, file_type: str) -> dict: """处理单个文档的完整流程。""" # 1. 解析文档 parsed_data = self.parser.parse(file_path, file_type) if not parsed_data: return {"status": "failed", "error": "Document parsing failed"} # 2. 分块 document_chunks = self._chunk_text(parsed_data["raw_text"]) all_risks = [] # 3. 对每个分块进行RAG+分析 for idx, chunk in enumerate(document_chunks): # 3.1 检索相关知识 context = self.retriever.query(chunk) # 3.2 LLM分析风险 risk_result = self.analyzer.analyze_chunk(chunk, context) # 为每个风险添加位置信息 for risk in risk_result.get("risks", []): risk["chunk_index"] = idx risk["page"] = self._map_chunk_to_page(idx, parsed_data.get("metadata")) # 映射回页码 all_risks.extend(risk_result.get("risks", [])) # 4. 汇总生成报告 final_report = self.report_generator.generate(all_risks, parsed_data) return { "status": "success", "document_id": "generated_id", "summary": final_report.get("summary"), "detailed_risks": final_report.get("detailed_risks"), "recommendations": final_report.get("recommendations") } def _chunk_text(self, text: str): # 实现文本分块逻辑 pass def _map_chunk_to_page(self, chunk_idx, metadata): # 实现分块到原始页码的映射 pass

智能体设计的经验：

• 状态管理与持久化：智能体需要记录任务状态（排队、解析中、分析中、完成、失败）。这通常借助数据库或消息队列（如 Celery + Redis）来实现，确保系统重启后任务能恢复。
• 错误处理与重试：网络波动、模型服务暂时不可用是常态。智能体需要有重试机制（特别是对 LLM 的调用），并设置最大重试次数。对于持续失败的任务，应标记为失败并通知管理员，而不是无限期阻塞队列。
• 可观测性：在关键步骤（解析开始/结束、检索、模型调用）记录详细的日志和指标（如耗时）。这有助于性能监控和问题排查。可以使用像 Prometheus 这样的工具来收集指标。

4. 系统部署、评估与优化实践

让一个原型系统变得稳定、可用，需要经过部署、评估和持续的优化。

4.1 系统部署与配置

推荐使用 Docker Compose 进行部署，这能很好地管理各个服务的依赖。

# docker-compose.yml 示例 version: '3.8' services: api: build: ./backend ports: - "8000:8000" environment: - OPENAI_API_KEY=${OPENAI_API_KEY} - OLLAMA_HOST=http://ollama:11434 - DATABASE_URL=postgresql://user:pass@db:5432/docsentinel depends_on: - db - ollama - chroma volumes: - ./uploads:/app/uploads # 挂载上传目录 ollama: image: ollama/ollama:latest ports: - "11434:11434" volumes: - ollama_data:/root/.ollama chroma: image: chromadb/chroma:latest environment: - IS_PERSISTENT=TRUE - PERSIST_DIRECTORY=/chroma_data volumes: - chroma_data:/chroma_data db: image: postgres:15 environment: - POSTGRES_USER=user - POSTGRES_PASSWORD=pass - POSTGRES_DB=docsentinel volumes: - postgres_data:/var/lib/postgresql/data volumes: ollama_data: chroma_data: postgres_data:

部署要点：

• 环境变量管理：所有敏感信息（API Keys、数据库密码）必须通过环境变量注入，切勿写在代码或配置文件中。
• 文件存储：上传的文档建议存储在网络存储或对象存储中，并设置定期清理策略，避免磁盘被占满。
• 健康检查：为每个服务配置健康检查端点，方便使用 Kubernetes 或 Docker Swarm 时的运维管理。
• 初始化脚本：在首次启动时，需要运行脚本来初始化数据库表结构，以及向向量数据库灌入初始的知识库数据。

4.2 效果评估与迭代优化

系统上线后，不能“一放了之”，必须建立评估体系。

评估指标：

1. 准确率与召回率：这是核心。需要构建一个标注好的测试集（一批文档，以及人工标注的风险点）。用系统跑一遍，计算：
   • 精确率：系统找出的风险中，有多少是真正的风险？（减少误报）
   • 召回率：真正的风险中，系统找出了多少？（减少漏报）
   • F1分数：两者的调和平均数，综合指标。
2. 处理速度：平均每页文档的处理时间。这直接影响用户体验。
3. 成本：如果使用付费 API，平均每份文档的分析成本是多少。
4. 稳定性：系统无故障运行时间，任务失败率。

优化方向：

• 提示词工程：这是提升效果性价比最高的方法。根据测试集上的错误案例，反复迭代提示词。例如，如果发现模型经常漏掉某种特定风险，就在提示词中增加针对该风险的强调和示例。
• 知识库优化：检查 RAG 检索出的上下文是否真的相关。如果不相关，可能是嵌入模型不匹配，或者知识库分块策略有问题。尝试不同的分块大小、重叠度，甚至更换嵌入模型。
• 模型微调：如果领域非常垂直（比如专做医疗数据合规），且有足够多的高质量标注数据，可以考虑对开源 LLM 进行 LoRA 等方式的微调，让它更擅长你的特定任务。
• 流程优化：分析性能瓶颈。如果大部分时间花在文档解析上，可以考虑引入更快的解析库或异步解析。如果 LLM 调用是瓶颈，可以调研更快的推理引擎（如 vLLM）或量化模型来加速本地推理。

4.3 常见问题与排查实录

在实际运行中，你肯定会遇到各种问题。以下是一些典型问题及排查思路：

问题1：LLM 输出格式不稳定，经常返回非 JSON 内容。

• 排查：首先检查提示词是否清晰要求了 JSON，并提供了严格的示例。然后，检查模型调用时的参数，特别是temperature是否设置过高（建议低于 0.3）。对于 OpenAI，使用response_format参数。对于 Ollama，查看其 API 是否支持类似功能。
• 解决：在代码中加强后处理。如果返回的不是合法 JSON，可以尝试用正则表达式提取可能存在的 JSON 部分，或者让模型进行“二次修正”。最坏情况下，将此结果标记为分析失败，记录日志，并可能将对应分块送入“待人工审核”队列。

问题2：处理长文档时，内存占用过高，甚至导致服务崩溃。

• 排查：监控服务的内存使用情况。问题可能出在：1）一次性将整个大文档读入内存进行解析；2）同时并发处理太多分块，所有分块的上下文都保存在内存中等待模型响应。
• 解决：采用流式或分页的方式读取和解析文档，避免全量加载。对于并发，使用有界队列和线程池/进程池，控制同时处理的任务数。考虑将耗时长的分析任务放入消息队列（如 RabbitMQ, Redis Queue），由后台工作进程异步处理，减轻 Web 服务器的压力。

问题3：知识库检索出的内容与文档不相关，导致分析结果驴唇不对马嘴。

• 排查：手动检查几个查询案例。打印出文档分块的文本和检索到的 Top-K 片段的文本，直观判断相关性。计算查询向量与召回片段向量的余弦相似度，看分数是否普遍偏低。
• 解决：如果相似度低，可能是嵌入模型不适用。尝试在你自己领域的数据上测试不同嵌入模型。如果相似度不低但内容还是不相关，可能是知识库本身的质量问题，或者分块方式不合理（比如把不同主题的内容切到了一个块里）。优化知识库的清洗和分块策略。

问题4：系统误报率太高，用户抱怨“狼来了”。

• 排查：收集用户反馈的误报案例，分析其共同特征。是某一类风险容易误报？还是某个特定的文档类型？
• 解决：引入“置信度过滤”和“白名单规则”。对于置信度为“低”的风险，可以选择不显示给最终用户，仅记录在日志中供管理员复查。同时，可以建立一些简单的正则表达式或关键词白名单规则，对于明显符合安全规范的语句（如“本文件遵守 ISO27001 标准”），直接在 RAG 或分析前就过滤掉，不提交给 LLM，减少干扰。

构建这样一个系统是一个持续迭代的过程。从最简单的流程跑通，到逐步优化每个模块的准确率和效率，再到完善系统的监控、告警和运维体系，每一步都需要结合实际的业务反馈和数据来进行。最重要的是，始终保持系统的可解释性和可干预性，毕竟 AI 是辅助工具，最终的责任和决策权仍然在人类专家手中。