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

发布时间:2026/11/4 11:43:31

法律文档 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场景也有类似的长文本强结构特征除了法律合同还有技术规范、医疗指南、政策文件这个双层检索的思路可以复用。核心是在切分文本之前先把文本的结构信息提取出来作为一级索引。
延伸阅读

更多相关文章

2026/11/4 11:41:28

TradingAgents-CN实战突破:7个智能场景的极简修复方案

TradingAgents-CN实战突破:7个智能场景的极简修复方案 【免费下载链接】TradingAgents-CN 基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版 项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN 面向中文用户的TradingAgents-…

2026/10/27 11:39:01

两个开源项目,带你系统学习 AI Agent

两个开源项目,带你系统学习 AI Agent 学 Agent 这条路上,我踩过一个坑:碎片文章看了很多,概念记了一堆,但脑子里始终没有一个完整的知识框架。直到我找到两个开源项目,才真正把 Agent 从「听说过」变成了「…

2026/11/4 11:40:47

如何搞定课程设计?别人走过的捷径,你也可以走

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

2026/11/4 11:40:47

【WebSocket】Spring WebSocket 应用出现“Connection closed before receiving a handshake response”错误,可能原因有哪些?

WebSocket 握手失败深度排查:IoT 设备指令下发系统的连接诊断体系 用户问题原文:Spring WebSocket 应用出现“Connection closed before receiving a handshake response”错误,可能原因有哪些? 在构建 IoT 设备指令下发系统时,我们曾因无法准确诊断握手失败原因导致设备大…

2026/11/4 11:40:47

【WebSocket】如何记录和追踪 WebSocket 消息的完整链路(用于日志审计或问题回溯)?

WebSocket 消息链路追踪实战:在线客服系统的全链路审计体系 用户问题原文:如何记录和追踪 WebSocket 消息的完整链路(用于日志审计或问题回溯)? 在构建在线客服系统时,我们曾因无法追踪消息链路导致严重客诉:用户投诉客服消息丢失,但技术团队无法确定是客户端发送失败、…

2026/11/4 11:40:47

分享Claude进行AI编程的相关技巧及程序员未来发展方向

文章目录背景一、核心前提:指令要“精准落地”,别让AI猜需求二、场景适配:选对场景,避开AI的“短板”✅ 高准确率建议使用AI场景的案例❌ 低准确率场景(尽量不用,避免内耗)三、AI Code关键技巧及…

2026/11/4 11:35:47

Windows 操作系统基础命令详解

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

2026/11/3 7:47:06

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

在智慧养宠时代,基于视觉的宠物身份识别技术正逐步从实验室走向家庭应用。然而,与配合度高的人类不同,宠物是典型的“非合作生物识别对象”——它们不会主动配合摄像头调整姿态,更无法控制光照环境。实际应用中,我们常…

2026/11/3 14:24:53

SSM电子书店源码全解析:结构、原理、部署与排错实战

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

2026/11/4 0:03:20

AI辅助技术栈评估:提升研发决策效率60%

1. 项目背景与核心价值技术决策一直是研发团队面临的重大挑战。随着技术生态的快速演进,技术栈的选择、维护和迭代变得愈发复杂。我们团队在过去两年里,先后经历了三次重大技术栈迁移,每次都需要投入大量人力进行技术评估和风险分析。直到去年…

2026/11/4 0:03:20

从生成Token到直接决策:Jev如何重塑AI Agent工程实践

1. 从"生成文字"到"拍板决策":Jev 到底在解决什么问题大多数人第一次接触大模型,脑子里建立的模型是"输入问题、输出文字"。你问它今天天气怎么样,它给你一段描述;你让它写个周报,它给你…

2026/11/4 0:03:20

32G统一内存跑大模型:内存账、TPS实测与端云分流

想在一台 32G 统一内存的小主机上把大模型跑成"日常可用",这件事我前后折腾了小半年。标题里说的这台 Mac mini M6,是我们这次讨论的目标平台,具体到手规格以实物为准;但下面这套推算方法和实测思路,对任何统…

2026/11/3 7:47:06

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

在智慧养宠时代,基于视觉的宠物身份识别技术正逐步从实验室走向家庭应用。然而,与配合度高的人类不同,宠物是典型的“非合作生物识别对象”——它们不会主动配合摄像头调整姿态,更无法控制光照环境。实际应用中,我们常…

2026/11/3 14:24:53

SSM电子书店源码全解析:结构、原理、部署与排错实战

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

还想了解更多?直接咨询顾问

免费诊断 + 免费方案 + 透明报价。

全国咨询热线400-8866-253
免费获取方案
☎咨询二维码 ☎ ↑