法律文档 Agent：长文本合同的条款提取与风险识别 RAG 方案

法律文档 Agent长文本合同的条款提取与风险识别 RAG 方案一、深度引言与场景痛点给一家律师事务所做合同审查Agent的时候遇到了一个长度问题。普通的RAG文档几千字到头了一份商业合同动辄30页、5万字起。传统的chunk切分策略500字一个chunkoverlap 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_factorylist) risk_flags: list[str] field(default_factorylist) 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_factorylist) # 条款解析器 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(rARTICLE\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_idf{contract.contract_id}_art_{article_num}, titletitle, article_numarticle_num, contentcontent, ) 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]条, rsubject\sto\sArticle\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(fRISK_{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(keylambda x: x[1], reverseTrue) 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_idCNTR-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场景也有类似的长文本强结构特征除了法律合同还有技术规范、医疗指南、政策文件这个双层检索的思路可以复用。核心是在切分文本之前先把文本的结构信息提取出来作为一级索引。