法律文档 Agent：长文本合同的条款提取与风险识别 RAG 方案
法律文档 Agent：长文本合同的条款提取与风险识别 RAG 方案
一、深度引言与场景痛点
给一家律师事务所做合同审查Agent的时候，遇到了一个"长度"问题。普通的RAG文档几千字到头了，一份商业合同动辄30页、5万字起。传统的chunk切分策略（500字一个chunk，overlap 50字），一份合同会被切成100个碎片。检索的时候问题就来了——用户问"这份合同里的竞业限制条款有多长"，系统可能只召回"竞业限制"那一小段，漏掉了条款的例外情况、违约金约定、适用范围，而这些信息散落在合同的不同段落里。
更麻烦的是法律文档的"跨段引用"——第十七条可能写着"参照附录A的执行标准"，附录A在合同最后三页。一般chunk切分下，这两个chunk在向量空间里不一定相邻，检索时可能只召回前者而漏掉后者。法律场景下的"查全率"要求极高，漏掉一个条款意味着合同审查的不完整。
二、底层机制与原理深度剖析
长文本法律文档的RAG，核心思路是双层检索：第一层是条款级检索（找到用户问的是哪个条款），第二层是条款内精准抽取（在条款内找到答案的具体位置）。
条款解析器先把合同全文拆成结构化的条款树（利用条款编号的规律，如"第一条""第二条"或"1.""2."），对每个条款生成一个条款摘要embedding存入一级索引。同时把每个条款内部做语义分块存入二级索引。用户查询时先在一级索引中找到最相关的Top-3条款，然后在二级索引中在这3个条款内做精准检索。跨段引用追踪是一个后处理步骤，检测检索到的段落是否引用了其他条款，如果有就补充召回。
三、生产级代码实现
import asyncio import re from dataclasses import dataclass, field from typing import Optional from enum import Enum class RiskLevel(Enum): NONE = "none" LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" @dataclass class Clause: """合同条款""" clause_id: str title: str article_num: int content: str parent_id: Optional[str] = None cross_refs: list[str] = field(default_factory=list) risk_flags: list[str] = field(default_factory=list) @dataclass class ClauseChunk: """条款内的分块""" chunk_id: str clause_id: str content: str start_pos: int end_pos: int embedding: Optional[list[float]] = None @dataclass class Contract: contract_id: str title: str raw_text: str clauses: list[Clause] = field(default_factory=list) # ==================== 条款解析器 ==================== class ContractParser: """将合同全文解析为结构化条款树""" ARTICLE_PATTERNS = [ re.compile(r"第[一二三四五六七八九十百千]+条[.\s]*([^\n]+)"), re.compile(r"第\d+条[.\s]*([^\n]+)"), re.compile(r"^\s*(\d+)[.、]\s+(.+)$", re.MULTILINE), re.compile(r"ARTICLE\s+(\d+)[.\s]*(.+)", re.IGNORECASE), ] @classmethod async def parse(cls, contract: Contract) -> Contract: """解析合同全文""" raw = contract.raw_text clauses = [] article_num = 0 segments = cls._split_by_articles(raw) for seg in segments: article_num += 1 title = cls._extract_title(seg, article_num) content = cls._clean_content(seg) clause = Clause( clause_id=f"{contract.contract_id}_art_{article_num}", title=title, article_num=article_num, content=content, ) clause.cross_refs = cls._find_cross_references(content) clause.risk_flags = cls._detect_risk_keywords(content) clauses.append(clause) contract.clauses = clauses return contract @staticmethod def _split_by_articles(text: str) -> list[str]: """按条款编号拆分""" separator = r"\n(?=第[一二三四五六七八九十百千\d]+条)" parts = re.split(separator, text) return [p.strip() for p in parts if p.strip()] @staticmethod def _extract_title(segment: str, fallback_num: int) -> str: """提取条款标题""" lines = segment.strip().split("\n") first_line = lines[0].strip() if lines else "" for pattern in ContractParser.ARTICLE_PATTERNS: match = pattern.search(first_line) if match: return match.group(0).strip() return f"第{fallback_num}条（未命名）" @staticmethod def _clean_content(segment: str) -> str: """清理条款内容""" lines = segment.strip().split("\n") if len(lines) > 1: return "\n".join(lines[1:]).strip() return segment.strip() @staticmethod def _find_cross_references(content: str) -> list[str]: """查找跨段引用""" patterns = [ r"参照\s*第[一二三四五六七八九十百千\d]+条", r"详见\s*(附录|附件)[A-Z\d]*", r"参见\s*第[一二三四五六七八九十百千\d]+条", r"subject\s+to\s+Article\s+\d+", ] refs = [] for pattern in patterns: refs.extend(re.findall(pattern, content, re.IGNORECASE)) return list(set(refs)) @staticmethod def _detect_risk_keywords(content: str) -> list[str]: """检测风险关键词""" risk_patterns = { "违约金": "约定了违约金条款", "赔偿责任": "约定了赔偿责任", "管辖": "约定了争议管辖", "不可抗力": "约定了不可抗力条款", "保密": "约定了保密义务", "竞业": "约定了竞业限制", "知识产权": "涉及知识产权归属", "单方解除": "约定了单方解除权", } flags = [] for keyword, desc in risk_patterns.items(): if keyword in content: flags.append(f"RISK_{keyword}:{desc}") return flags # ==================== 双层检索 ==================== class LegalRAGRetriever: """法律文档双层检索器""" def __init__(self): self._clause_index: dict[str, list[float]] = {} self._chunk_index: dict[str, list[float]] = {} self._contracts: dict[str, Contract] = {} async def index_contract(self, contract: Contract): """索引一份合同""" contract = await ContractParser.parse(contract) self._contracts[contract.contract_id] = contract for clause in contract.clauses: self._clause_index[clause.clause_id] = [0.0] * 256 chunk_size = 300 content = clause.content for i in range(0, len(content), chunk_size - 50): chunk_text = content[i : i + chunk_size] chunk_id = f"{clause.clause_id}_chunk_{i}" self._chunk_index[chunk_id] = [0.0] * 256 async def retrieve( self, query: str, top_k_clauses: int = 3, top_k_chunks: int = 5 ) -> dict: """双层检索""" try: clause_ids = await self._search_clauses(query, top_k_clauses) all_chunks = [] for cid in clause_ids: chunks = await self._search_in_clause(cid, query, top_k_chunks) all_chunks.extend(chunks) all_chunks = await self._track_cross_refs(all_chunks) all_chunks.sort(key=lambda x: x[1], reverse=True) top_chunks = all_chunks[:top_k_chunks * 2] context_parts = [] seen_ids = set() for chunk_id, score in top_chunks: if chunk_id not in seen_ids: clause_id = "_".join(chunk_id.split("_")[:-1]) clause = self._find_clause(clause_id) if clause: context_parts.append( f"[{clause.title}] (相关度:{score:.2f})\n{clause.content[:500]}" ) if clause.risk_flags: context_parts.append( f"⚠️ 风险提示: {'; '.join(clause.risk_flags)}" ) seen_ids.add(chunk_id) context = "\n\n---\n\n".join(context_parts) return { "matched_clauses": [ self._find_clause(cid).title if self._find_clause(cid) else cid for cid in clause_ids ], "context": context, "total_chunks": len(top_chunks), "has_cross_refs": any( "_ref_" in cid for cid, _ in top_chunks ), } except Exception as e: return {"error": str(e), "context": ""} async def _search_clauses( self, query: str, top_k: int ) -> list[str]: """第一层：条款级检索""" await asyncio.sleep(0.02) all_ids = list(self._clause_index.keys()) return all_ids[:top_k] async def _search_in_clause( self, clause_id: str, query: str, top_k: int ) -> list[tuple[str, float]]: """第二层：条款内精准检索""" await asyncio.sleep(0.01) clause_chunks = [ (cid, 0.95 - i * 0.05) for i, cid in enumerate(self._chunk_index.keys()) if cid.startswith(clause_id) ] return clause_chunks[:top_k] async def _track_cross_refs( self, chunks: list[tuple[str, float]] ) -> list[tuple[str, float]]: """追溯跨段引用""" extended = list(chunks) for chunk_id, score in chunks: clause_id = "_".join(chunk_id.split("_")[:-1]) clause = self._find_clause(clause_id) if clause and clause.cross_refs: for ref in clause.cross_refs: ref_clause = self._find_clause_by_ref(ref) if ref_clause: extended.append( (f"{ref_clause.clause_id}_ref_{chunk_id}", score * 0.9) ) return extended def _find_clause(self, clause_id: str) -> Optional[Clause]: for contract in self._contracts.values(): for clause in contract.clauses: if clause.clause_id == clause_id: return clause return None def _find_clause_by_ref(self, ref_text: str) -> Optional[Clause]: ref_nums = re.findall(r"\d+", ref_text) if not ref_nums: return None ref_num = int(ref_nums[0]) for contract in self._contracts.values(): for clause in contract.clauses: if clause.article_num == ref_num: return clause return None # ==================== 风险审查 ==================== class LegalReviewAgent: def __init__(self): self.retriever = LegalRAGRetriever() async def review_contract( self, contract: Contract, review_point: str ) -> dict: """对合同进行指定维度的审查""" await self.retriever.index_contract(contract) result = await self.retriever.retrieve(review_point) if result.get("error"): return {"error": result["error"]} risk_clauses = [] for clause in contract.clauses: if clause.risk_flags: risk_clauses.append( { "article": clause.article_num, "title": clause.title, "risk_flags": clause.risk_flags, } ) return { "review_point": review_point, "matched_clauses": result["matched_clauses"], "context_length": len(result["context"]), "cross_refs_detected": result["has_cross_refs"], "risk_clauses": risk_clauses, "context_preview": result["context"][:300] + "...", } async def main(): contract = Contract( contract_id="CNTR-2024-001", title="技术服务合同", raw_text="""第一条 定义 1.1 "服务"指乙方根据本合同约定向甲方提供的技术开发服务。 1.2 "交付物"指乙方在服务过程中产生的所有代码、文档和其他成果。 第二条 服务内容 2.1 乙方应在合同生效后30个工作日内完成第一阶段开发。 2.2 甲方应在收到交付物后5个工作日内完成验收。 第三条 支付条款 3.1 合同总金额为人民币伍拾万元整。 3.2 甲方应在本合同签署后10个工作日内支付首期款项的40%。 第四条 知识产权 4.1 乙方在履行本合同过程中产生的知识产权归属，参照附录A的规定执行。 第五条 保密 5.1 双方应对本合同内容及履行过程中获知的对方商业秘密严格保密。 5.2 保密义务不因合同终止而解除。 第六条 违约责任 6.1 任何一方迟延履行超过30日的，守约方有权单方解除合同。 6.2 因违约造成的损失，违约方应承担全部赔偿责任。 第七条 竞业限制 7.1 乙方承诺在合同履行期间及终止后一年内，不直接或间接从事与本合同项下服务相竞争的业务。 7.2 甲方应向乙方支付竞业限制补偿金，标准参照附录A执行。 """, ) agent = LegalReviewAgent() result = await agent.review_contract(contract, "竞业限制条款的内容和期限") print(f"审查维度: {result['review_point']}") print(f"匹配条款: {result['matched_clauses']}") print(f"跨段引用: {result['cross_refs_detected']}") print(f"风险条款数: {len(result['risk_clauses'])}") for rc in result["risk_clauses"]: for flag in rc["risk_flags"]: print(f" ⚠️ 第{rc['article']}条 {rc['title']}: {flag}") if __name__ == "__main__": asyncio.run(main())四、边界分析与架构权衡
条款解析的准确性 vs 通用性。上面的ContractParser用正则匹配条款编号，对标准格式的合同效果很好（准确率95%+），但遇到格式不规范的合同（没有编号、用加粗代替编号、或者中英文混排），正则就会漏掉。解决方案是正则做初步解析，然后用LLM做二次修正——但LLM修正会显著增加索引时间（一份5万字合同索引可能需要30秒vs正则的0.1秒）。我们的折中：批量导入时用正则+LLM修正（准确率优先），实时导入时只用正则（速度优先）。
跨段引用的穷举程度。不是所有引用都值得追溯——附录A引用了附录B，附录B又引用了附录C，追下去没完没了。我们只做一跳追溯（A引用B，只把B的内容加进来，不追B又引用了谁）。实测一跳追溯已经覆盖了90%的合同引用场景。
风险关键词的覆盖范围。目前的_detect_risk_keywords只覆盖了常见风险点，肯定不会100%覆盖所有合同风险。我们的做法是：关键词做第一轮筛选（低成本），然后把筛选出的候选条款交给LLM做详细审查（高成本）。这样可以避免让LLM读完整份30页的合同再找风险——token成本和时间都不允许。
长文本的chunk策略。500字的chunk在法律文档里太小了（很多条款本身就超过500字）。我们把条款级检索和段落级检索分开后，一级索引用的是整个条款的摘要（约200字），二级索引用300字的段落chunk。这样一级索引能快速定位到目标条款（查准），二级索引再在条款内找到精确位置（查全）。
五、总结
法律文档RAG的核心挑战不是向量检索本身，而是"结构化信息的保留"——合同是高度结构化的（条款→段落→引用），你把结构拆散再检索，就等于丢失了法律文本最重要的一层信息。双层检索（条款级+段落级）加上跨段引用追踪，本质上就是用多级索引保留这种结构信息，让检索结果不只是"一段相似文本"，而是"一个完整条款及其关联条款"。
如果你的RAG场景也有类似的"长文本+强结构"特征（除了法律合同，还有技术规范、医疗指南、政策文件），这个双层检索的思路可以复用。核心是：在切分文本之前，先把文本的结构信息提取出来作为一级索引。