各位同仁，下午好！
今天，我们齐聚一堂，共同探讨一个充满挑战与机遇的议题：如何利用前沿的AI技术，特别是LangChain框架，来革新法律领域中最为繁琐、耗时且易出错的工作——百万字合同的条款冲突自动扫描与合规性评估。作为一名编程专家，我将带领大家深入剖析这个“法律 Agent”的构建过程，从理论到实践，从模块到系统，力求让大家对背后的技术原理和实现细节有清晰的理解。
法律领域的AI革命与LangChain的潜力
在数字时代，法律行业正面临前所未有的变革。传统的合同审查工作，往往需要资深律师投入大量时间进行逐字逐句的研读，尤其对于动辄数十页、上百页的复杂商业合同，其工作量是巨大的。这不仅带来了高昂的人力成本，更难以避免因疲劳或疏忽而导致的遗漏与错误，最终可能引发严重的法律风险。
人工智能，特别是大型语言模型（LLM）的崛起，为我们提供了一把解决这些痛点的利器。LLM以其强大的文本理解、生成和推理能力，展现出在法律文本分析方面的巨大潜力。然而，直接将原始合同喂给LLM并非万全之策，它面临上下文窗口限制、幻觉（hallucination）问题、以及缺乏领域特定知识等挑战。
这时，LangChain应运而生。LangChain是一个强大的LLM应用开发框架，它通过提供模块化组件和灵活的链式结构，使我们能够将LLM与其他数据源、计算逻辑和工具进行有机结合，从而构建出更复杂、更智能、更可靠的AI应用。它允许我们将大型、复杂的任务分解为一系列小任务，并以编排的方式调用LLM和外部工具来完成这些任务。对于法律合同分析这种需要多步骤推理、检索大量信息并进行结构化输出的场景，LangChain无疑是理想的选择。
今天，我们的目标就是利用LangChain，构建一个“法律 Agent”。这个Agent将具备以下核心能力：
- 海量合同文本的处理能力：突破LLM的上下文限制，有效处理百万字级别的合同。
- 关键条款的精准抽取与标准化：从非结构化的合同文本中识别并结构化存储重要条款。
- 条款冲突的自动扫描与识别：在多个合同或同一合同内部发现矛盾、不一致或潜在风险的条款。
- 合同条款的合规性评估：对照法律法规、行业标准或公司内部政策，评估条款的合规性并识别潜在违规。
整个过程将高度自动化，极大地提升效率，降低成本，并提高审查的准确性与一致性。
核心挑战与解决方案概述
在深入代码之前，我们先宏观地审视一下构建这个法律Agent所面临的核心挑战，并简要阐述我们的解决方案。
| 核心挑战 | 解决方案 War |
LLMs / ChatModels
这是驱动我们整个Agent的核心。LangChain支持多种大型语言模型（LLM）或聊天模型（Chat Model），例如OpenAI的GPT系列、Anthropic的Claude系列、以及各种开源模型如Llama、Mistral等。
对于法律文本分析，我们通常推荐使用能力更强、推理更准确的模型，例如GPT-4或Claude 3 Opus。虽然成本相对较高，但其在理解复杂语境、处理长文本依赖和生成高质量结构化输出方面的表现，能显著提升Agent的整体性能。
我们首先需要初始化一个LLM实例。
import os
from langchain_openai import ChatOpenAI
from langchain_community.llms import OpenAI # For older LLM-based chains if needed
# 确保你的OpenAI API Key已设置为环境变量
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
def initialize_llm(model_name: str = "gpt-4-turbo-preview", temperature: float = 0.1):
"""
初始化一个LangChain聊天模型实例。
:param model_name: 使用的模型名称，如"gpt-4-turbo-preview", "gpt-3.5-turbo"
:param temperature: 模型的创造性/随机性，0表示最确定性。法律场景通常设为低值。
:return: ChatOpenAI实例
"""
try:
llm = ChatOpenAI(model=model_name, temperature=temperature, openai_api_key=os.environ["OPENAI_API_KEY"])
print(f"成功初始化LLM: {model_name}")
return llm
except KeyError:
print("错误：请设置环境变量 'OPENAI_API_KEY'。")
raise
except Exception as e:
print(f"初始化LLM失败: {e}")
raise
# 示例使用
# llm = initialize_llm()Prompts (提示词)
提示词是与LLM沟通的“语言”。一个精心设计的提示词能够显著提升LLM的理解能力和任务完成质量。LangChain提供了丰富的提示词模板，帮助我们构建结构化、可复用的提示。
在法律场景中，提示词的设计至关重要。我们需要明确指定LLM的角色（例如，“你是一个专业的法律助手”）、任务目标、输出格式要求，并提供必要的上下文信息。
from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate, MessagesPlaceholder
def create_clause_extraction_prompt():
"""
创建用于条款抽取的提示词模板。
"""
system_message = SystemMessagePromptTemplate.from_template(
"你是一个专业的法律助手，擅长从合同文本中提取关键条款并进行结构化分析。请严格按照用户指定的JSON格式输出。"
)
human_message = HumanMessagePromptTemplate.from_template(
"请从以下合同文本片段中提取关键条款，包括但不限于付款条款、违约责任、管辖权、保密条款、合同期限等。请为每个条款提供类型、原始内容、简洁摘要和关键细节。最终结果请以Pydantic模型 `{format_instructions}` 的JSON格式返回。n"
"合同文本片段：n{contract_chunk}"
)
# MessagesPlaceholder 用于Agent中插入历史对话或Agent的思考过程
# agent_scratchpad 用于Agent记录其思考和工具使用过程
agent_scratchpad = MessagesPlaceholder(variable_name="agent_scratchpad")
# ChatPromptTemplate 组合多个消息
prompt_template = ChatPromptTemplate.from_messages([
system_message,
human_message,
agent_scratchpad # 在Agent中会用到
])
print("条款抽取提示词模板创建成功。")
return prompt_template
# 示例使用（Agent场景）
# extraction_prompt = create_clause_extraction_prompt()
# formatted_prompt = extraction_prompt.format_prompt(
# contract_chunk="...",
# format_instructions="...",
# agent_scratchpad=[] # 初始为空列表
# ).to_messages()Parsers (解析器)
LLM的原始输出通常是自由文本，这对于后续的自动化处理非常不便。LangChain的解析器可以将LLM的输出转换为结构化的数据格式，如JSON、Python对象等。PydanticOutputParser是一个非常强大的工具，它允许我们定义Pydantic模型来规范LLM的输出结构，并自动进行解析和校验。
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser
from typing import List
# 定义一个Pydantic模型来规范条款的结构
class Clause(BaseModel):
clause_type: str = Field(description="条款类型，如'付款条款', '违约责任', '保密条款'")
content: str = Field(description="条款的原始文本内容")
summary: str = Field(description="条款的简洁摘要，概括其核心要点")
key_details: List[str] = Field(description="条款中的关键细节或要素列表，例如付款金额、期限等")
clause_id: str = Field(description="条款的唯一标识符，可用于追踪")
source_contract: str = Field(description="条款来源的合同名称或标识")
class ContractTerms(BaseModel):
contract_name: str = Field(description="合同的名称或标识")
extracted_clauses: List[Clause] = Field(description="从合同中提取出的所有关键条款列表")
def get_clause_parser():
"""
获取用于解析合同条款的PydanticOutputParser。
"""
parser = PydanticOutputParser(pydantic_object=ContractTerms)
print("条款解析器创建成功。")
return parser
# 示例使用
# clause_parser = get_clause_parser()
# format_instructions = clause_parser.get_format_instructions()
# # 然后在prompt中引用 format_instructionsDocument Loaders (文档加载器)
LangChain提供了多种文档加载器，用于从不同来源加载文本数据，如本地文件、网页、PDF等。对于合同文本，我们通常会从文件系统加载。
from langchain_community.document_loaders import TextLoader, PyPDFLoader # 也可以用UnstructuredLoader处理更多格式
from langchain.docstore.document import Document
from typing import List
def load_documents_from_files(file_paths: List[str]) -> List[Document]:
"""
从指定的文件路径加载文档。
支持.txt和.pdf文件。
:param file_paths: 文件路径列表
:return: Document对象列表
"""
documents = []
for path in file_paths:
if path.endswith(".txt"):
loader = TextLoader(path, encoding="utf-8")
elif path.endswith(".pdf"):
loader = PyPDFLoader(path) # 需要安装 pypdf 库
# elif path.endswith(".docx"):
# from langchain_community.document_loaders import Docx2txtLoader
# loader = Docx2txtLoader(path) # 需要安装 python-docx 库
else:
print(f"警告：不支持的文件格式 {path}，跳过。")
continue
try:
docs = loader.load()
# 为每个文档添加元数据，比如原始文件路径
for doc in docs:
doc.metadata['source'] = path
if 'page' in doc.metadata:
doc.metadata['clause_id'] = f"{os.path.basename(path)}_p{doc.metadata['page']}_{hash(doc.page_content)[:8]}"
else:
doc.metadata['clause_id'] = f"{os.path.basename(path)}_{hash(doc.page_content)[:8]}"
documents.extend(docs)
print(f"成功加载文件: {path}，共 {len(docs)} 页/段。")
except Exception as e:
print(f"加载文件 {path} 失败: {e}")
return documents
# 示例：创建一些虚拟合同文件
def create_dummy_contract_files():
if not os.path.exists("contracts"):
os.makedirs("contracts")
with open("contracts/contract_A.txt", "w", encoding="utf-8") as f:
f.write("合同Ann条款1：本合同付款周期为自发票开具之日起30个自然日。n条款2：任何一方违约，应支付合同总金额的10%作为违约金。n条款3：本合同受中国法律管辖。n条款4：所有商业秘密均为保密信息，未经许可不得披露。n条款5：本合同有效期为1年。")
with open("contracts/contract_B.txt", "w", encoding="utf-8") as f:
f.write("合同Bnn条款A：支付条款为发货后60日内。n条款B：违约方需赔偿守约方因此遭受的全部损失。n条款C：本合同适用上海市地方法律。n条款D：保密义务持续至合同终止后3年。n条款E：合同期限为2年。")
with open("contracts/contract_C.txt", "w", encoding="utf-8") as f:
f.write("合同Cnn条款I：付款应在30天内完成。n条款II：违约金为每日千分之五。n条款III：争议提交北京仲裁委员会仲裁。n条款IV：保密条款与行业标准一致。n条款V：本合同可无限期自动续约。")
print("虚拟合同文件创建成功。")
# create_dummy_contract_files()
# contract_paths = [f"contracts/{f}" for f in os.listdir("contracts") if f.endswith(".txt")]
# raw_contracts = load_documents_from_files(contract_paths)Text Splitters (文本分割器)
由于LLM的上下文窗口限制，我们不能将整个百万字合同一次性输入。文本分割器的作用就是将长文档分割成更小的、可管理的块（chunks），同时尽量保持文本的语义完整性。RecursiveCharacterTextSplitter是一个常用的分割器，它会尝试使用多种分隔符（如段落、句子、单词）进行递归分割，直到满足块大小要求。
from langchain.text_splitter import RecursiveCharacterTextSplitter
def split_documents(documents: List[Document], chunk_size: int = 2000, chunk_overlap: int = 200) -> List[Document]:
"""
将加载的文档分割成更小的块。
:param documents: 待分割的Document对象列表
:param chunk_size: 每个块的最大字符数
:param chunk_overlap: 相邻块之间的重叠字符数，有助于保持上下文
:return: 分割后的Document对象列表
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len, # 使用字符长度作为度量
add_start_index=True # 添加块在原始文档中的起始位置
)
split_docs = text_splitter.split_documents(documents)
# 为每个分割后的文档块添加一个唯一的ID和来源信息
for i, doc in enumerate(split_docs):
doc.metadata['chunk_id'] = f"{doc.metadata.get('source', 'unknown_source')}_chunk_{i}"
doc.metadata['original_clause_id'] = doc.metadata.get('clause_id', 'N/A') # 记录原始文档的ID
print(f"文档分割完成，共生成 {len(split_docs)} 个块。")
return split_docs
# split_contract_chunks = split_documents(raw_contracts)Embeddings (嵌入)
嵌入是将文本转换为高维向量的过程。这些向量捕捉了文本的语义信息，使得语义相似的文本在向量空间中距离更近。这是实现RAG（Retrieval-Augmented Generation）和条款相似性检索的基础。OpenAI提供了强大的嵌入模型，HuggingFace也提供了许多开源嵌入模型。
from langchain_openai import OpenAIEmbeddings
def initialize_embeddings(api_key: str) -> OpenAIEmbeddings:
"""
初始化OpenAI嵌入模型。
:param api_key: OpenAI API Key
:return: OpenAIEmbeddings实例
"""
try:
embeddings = OpenAIEmbeddings(openai_api_key=api_key)
print("OpenAI嵌入模型初始化成功。")
return embeddings
except KeyError:
print("错误：请设置环境变量 'OPENAI_API_KEY'。")
raise
except Exception as e:
print(f"初始化嵌入模型失败: {e}")
raise
# embeddings_model = initialize_embeddings(os.environ["OPENAI_API_KEY"])Vector Stores (向量数据库)
向量数据库用于存储嵌入向量，并提供高效的相似性搜索功能。当我们需要检索与某个查询文本语义相似的合同条款或法律法规时，向量数据库能够快速返回最相关的结果。LangChain支持多种向量数据库，如FAISS（本地内存）、Chroma（本地或客户端-服务器）、Pinecone、Weaviate等。对于原型开发和中小型项目，FAISS和Chroma是很好的选择。
from langchain_community.vectorstores import FAISS
def create_vector_store(split_docs: List[Document], embeddings_model: OpenAIEmbeddings, store_path: str = "faiss_index") -> FAISS:
"""
基于分割后的文档和嵌入模型创建FAISS向量数据库。
:param split_docs: 分割后的Document对象列表
:param embeddings_model: 嵌入模型实例
:param store_path: 存储FAISS索引的路径
:return: FAISS向量数据库实例
"""
print("开始创建向量数据库...")
vector_store = FAISS.from_documents(split_docs, embeddings_model)
vector_store.save_local(store_path)
print(f"向量数据库创建并保存到 {store_path} 成功。")
return vector_store
def load_vector_store(embeddings_model: OpenAIEmbeddings, store_path: str = "faiss_index") -> FAISS:
"""
从本地路径加载FAISS向量数据库。
:param embeddings_model: 嵌入模型实例（加载时也需要）
:param store_path: 存储FAISS索引的路径
:return: FAISS向量数据库实例
"""
print(f"尝试从 {store_path} 加载向量数据库...")
try:
vector_store = FAISS.load_local(store_path, embeddings_model, allow_dangerous_deserialization=True)
print("向量数据库加载成功。")
return vector_store
except Exception as e:
print(f"加载向量数据库失败: {e}")
raise
# vector_store = create_vector_store(split_contract_chunks, embeddings_model)
# # 或者
# # vector_store = load_vector_store(embeddings_model)Chains (链)
链是LangChain的核心概念之一，它允许我们将多个LLM调用、提示词、解析器、检索器等组件组合成一个序列化的工作流。这使得我们可以构建复杂的、多步骤的推理过程。常用的链包括：
LLMChain: 最基础的链，将Prompt和LLM连接起来。RetrievalQAChain: 结合了检索和问答功能，常用于RAG。MapReduceDocumentsChain: 处理大量文档的常见模式，先对每个文档块进行独立处理（Map），再汇总结果（Reduce）。
在本案例中，我们主要通过Agent来编排工具，而工具内部可能包含简单的LLMChain或RetrievalQAChain。
Agents & Tools (智能体与工具)
Agent是LangChain中最强大的抽象之一。它允许LLM根据任务需求自主决定要执行的步骤，并选择合适的工具来完成这些步骤。Agent通常遵循ReAct（Reasoning and Acting）模式，即“思考（Thought）-行动（Action）”循环。
- Tools (工具)：Agent可以调用的外部功能，例如检索向量数据库、执行代码、调用API等。每个工具都有明确的描述，告诉Agent它能做什么以及何时使用。
- Agent Executor (Agent执行器)：负责驱动Agent的思考-行动循环，根据LLM的决策调用工具，并将工具的输出反馈给LLM进行下一步思考。
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 定义Agent可以使用的工具
# 注意：这些工具函数需要能够访问到前面创建的全局或通过参数传递的资源（如vector_store, llm）
# 辅助函数：将Document转换为Clause对象，假设Document的page_content是条款内容
# 并且metadata中包含 'clause_type', 'summary', 'source', 'original_clause_id'
def doc_to_clause(doc: Document) -> Clause:
return Clause(
clause_type=doc.metadata.get('clause_type', '未知类型'),
content=doc.page_content,
summary=doc.metadata.get('summary', doc.page_content[:100] + '...'),
key_details=[], # 检索时可能无法直接获取，后续可再提取
clause_id=doc.metadata.get('original_clause_id', doc.metadata.get('chunk_id', 'N/A')),
source_contract=doc.metadata.get('source', 'N/A')
)
@tool
def retrieve_similar_clauses(query_clause_summary: str, k: int = 5) -> List[dict]:
"""
根据给定的条款摘要，从合同知识库中检索最相似的K个合同条款。
返回一个包含条款内容、来源和元数据的字典列表。
"""
global_vector_store = globals().get('contract_vector_store') # 假设在全局环境中
if not global_vector_store:
raise ValueError("合同向量数据库未初始化。")
print(f"正在检索与 '{query_clause_summary}' 相似的条款...")
retriever = global_vector_store.as_retriever(search_kwargs={"k": k})
docs = retriever.get_relevant_documents(query_clause_summary)
results = []
for doc in docs:
# 将Document的元数据和内容打包成Agent易于理解的格式
results.append({
"content": doc.page_content,
"source_contract": doc.metadata.get('source', '未知合同'),
"clause_id": doc.metadata.get('original_clause_id', doc.metadata.get('chunk_id', 'N/A')),
"summary": doc.metadata.get('summary', doc.page_content[:100] + '...')
})
return results
# 定义冲突分析结果的Pydantic模型
class ConflictAnalysisResult(BaseModel):
is_conflict: bool = Field(description="两个条款之间是否存在冲突")
conflict_type: str = Field(description="冲突类型，如'直接矛盾', '隐含不一致', '遗漏', '无冲突'")
description: str = Field(description="冲突的具体描述、影响和涉及的关键点")
conflicting_clauses_summary: List[str] = Field(description="涉及冲突的条款的简洁摘要列表")
suggested_resolution: str = Field(description="可能的解决方案或建议，如'修改付款周期', '明确管辖权'")
conflict_parser = PydanticOutputParser(pydantic_object=ConflictAnalysisResult)
@tool
def analyze_clause_conflict(clause1_full_details: str, clause2_full_details: str) -> str:
"""
分析两个合同条款之间是否存在冲突，并提供详细的冲突分析。
输入是两个条款的详细内容（可以是原始文本或包含元数据的详细描述）。
返回一个JSON字符串，描述冲突分析结果。
"""
global_llm = globals().get('main_llm') # 假设在全局环境中
if not global_llm:
raise ValueError("LLM未初始化。")
analysis_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的法律分析师，擅长识别合同条款间的冲突。请严格按照JSON格式输出。"),
("human", "请分析以下两个条款之间是否存在冲突。如果存在，请说明冲突类型、详细描述、涉及的条款，并提出可能的解决方案。n"
"条款一：n{clause1}nn"
"条款二：n{clause2}nn"
"请以Pydantic模型 `{format_instructions}` 的JSON格式返回分析结果。"),
])
prompt = analysis_prompt.format(
clause1=clause1_full_details,
clause2=clause2_full_details,
format_instructions=conflict_parser.get_format_instructions()
)
response = global_llm.invoke(prompt)
return response.content # 返回原始JSON字符串，Agent会处理解析
# Agent的工具集
# tools_for_agent = [
# retrieve_similar_clauses,
# analyze_clause_conflict
# ]实战：构建百万字合同处理流程
现在，我们已经了解了LangChain的各个核心组件，是时候将它们组装起来，构建我们的“法律 Agent”了。整个流程将分为四个主要步骤：数据准备与预处理、条款抽取与标准化、条款冲突自动扫描、合规性评估。
第一步：数据准备与预处理
这是所有分析工作的基础。我们需要加载大量的合同文本，并将其转化为LLM可处理的格式。
import os
import uuid # 用于生成唯一ID
# 确保OPENAI_API_KEY已设置
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# 全局变量，方便工具函数访问
main_llm = None
embeddings_model = None
contract_vector_store = None
regulation_vector_store = None # 合规性评估用
def setup_global_resources():
global main_llm, embeddings_model
main_llm = initialize_llm()
embeddings_model = initialize_embeddings(os.environ["OPENAI_API_KEY"])
def prepare_contract_data(contract_dir: str = "contracts"):
"""
加载、分割合同，并创建向量数据库。
"""
global contract_vector_store
# 1. 创建虚拟合同文件（如果不存在）
if not os.path.exists(contract_dir) or not os.listdir(contract_dir):
print(f"'{contract_dir}' 目录不存在或为空，正在创建虚拟合同文件...")
create_dummy_contract_files()
# 2. 加载合同文档
contract_paths = [os.path.join(contract_dir, f) for f in os.listdir(contract_dir) if f.endswith(".txt")]
if not contract_paths:
raise FileNotFoundError(f"在目录 '{contract_dir}' 中未找到任何合同文件。")
raw_contracts = load_documents_from_files(contract_paths)
# 3. 文本分割
split_contract_chunks = split_documents(raw_contracts)
# 4. 创建并保存合同向量数据库
contract_vector_store = create_vector_store(split_contract_chunks, embeddings_model, store_path="faiss_contract_index")
print("合同数据准备完成。")
return split_contract_chunks
# 在主流程中调用
# setup_global_resources()
# contract_chunks = prepare_contract_data()第二步：条款抽取与标准化
这一步的目标是从每个合同块中识别并提取出关键的法律条款，并将其标准化为我们定义的 Clause Pydantic模型。这将为后续的冲突检测和合规性评估提供结构化的数据基础。
由于每个合同块都需要独立处理，并且LLM的调用是串行的，对于百万字合同的分割块数量可能非常庞大。我们需要一个高效的批量处理机制。
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser
from typing import List, Dict, Any
# 重新定义Clause和ContractTerms，增加ID和来源信息，确保一致性
class Clause(BaseModel):
clause_id: str = Field(description="条款的唯一标识符，可用于追踪")
source_contract: str = Field(description="条款来源的合同名称或标识")
clause_type: str = Field(description="条款类型，如'付款条款', '违约责任', '保密条款'")
content: str = Field(description="条款的原始文本内容")
summary: str = Field(description="条款的简洁摘要，概括其核心要点")
key_details: List[str] = Field(description="条款中的关键细节或要素列表，例如付款金额、期限等")
class ChunkExtractedClauses(BaseModel):
extracted_clauses: List[Clause] = Field(description="从合同片段中提取出的所有关键条款列表")
clause_chunk_parser = PydanticOutputParser(pydantic_object=ChunkExtractedClauses)
def extract_clauses_from_chunk_llm(llm: ChatOpenAI, chunk_doc: Document) -> List[Clause]:
"""
使用LLM从单个文档块中提取条款。
:param llm: ChatOpenAI实例
:param chunk_doc: 包含合同片段的Document对象
:return: 提取出的Clause对象列表
"""
extraction_prompt_template = ChatPromptTemplate.from_messages([
("system", "你是一个专业的法律助手，擅长从合同文本中提取关键条款并进行结构化分析。请严格按照JSON格式输出。"),
("human", "请从以下合同文本片段中提取关键条款，包括但不限于付款条款、违约责任、管辖权、保密条款、合同期限等。请为每个条款提供类型、原始内容、简洁摘要和关键细节。最终结果请以Pydantic模型 `{format_instructions}` 的JSON格式返回。n"
"合同文本片段：n{contract_chunk}n"
"请确保每个条款的 `clause_id` 是唯一的，其前缀为 '{base_id}'，并正确填写 `source_contract`。"),
])
source_contract_name = os.path.basename(chunk_doc.metadata.get('source', '未知合同'))
base_id = chunk_doc.metadata.get('chunk_id', str(uuid.uuid4())[:8])
prompt = extraction_prompt_template.format(
contract_chunk=chunk_doc.page_content,
format_instructions=clause_chunk_parser.get_format_instructions(),
base_id=base_id,
source_contract=source_contract_name
)
try:
response = llm.invoke(prompt)
parsed_output = clause_chunk_parser.parse(response.content)
# 补充或覆盖LLM可能未正确生成的ID和来源信息
for i, clause in enumerate(parsed_output.extracted_clauses):
clause.clause_id = f"{base_id}_{i}" # 确保每个条款ID唯一
clause.source_contract = source_contract_name
return parsed_output.extracted_clauses
except Exception as e:
print(f"从块 '{chunk_doc.metadata.get('chunk_id', 'N/A')}' 提取条款失败: {e}. LLM原始输出: {response.content if 'response' in locals() else 'N/A'}")
return []
def extract_all_clauses(llm: ChatOpenAI, contract_chunks: List[Document]) -> List[Clause]:
"""
从所有合同块中提取条款。
"""
all_extracted_clauses: List[Clause] = []
print(f"n开始从 {len(contract_chunks)} 个合同块中提取条款...")
for i, chunk in enumerate(contract_chunks):
print(f" 正在处理块 {i+1}/{len(contract_chunks)} (来源: {chunk.metadata.get('source', 'N/A')})...")
clauses = extract_clauses_from_chunk_llm(llm, chunk)
all_extracted_clauses.extend(clauses)
print(f"条款提取完成，共提取 {len(all_extracted_clauses)} 个条款。")
return all_extracted_clauses
# 在主流程中调用
# all_extracted_clauses = extract_all_clauses(main_llm, contract_chunks)重要提示：在实际应用中，由于LLM的API调用成本和速度，对数千上万个块进行逐一调用会非常昂贵和耗时。可以考虑以下优化：
- 并行处理：使用多线程或异步IO并行调用LLM。
- 批处理：如果LLM支持，尝试将多个小块合并成一个大的Prompt进行处理（但要小心上下文窗口限制）。
- 缓存：对已处理的块结果进行缓存。
- 优先级：优先处理关键合同或关键条款类型。
第三步：条款冲突自动扫描
这是Agent的核心功能之一。我们需要设计一个Agent，它能够：
- 遍历所有已提取的条款。
- 对于每个条款，从合同知识库中检索出语义相似的其他条款。
- 利用LLM判断这些相似条款之间是否存在冲突，并给出冲突报告和解决方案。
我们将使用前面定义的 `retrievesimilar