【大模型图解】「1 行 LangChain 代码」竟让 10 万份合同 3 分钟审完？

最新推荐文章于 2025-05-13 20:18:05 发布

AI程序猿人

最新推荐文章于 2025-05-13 20:18:05 发布

阅读量529

点赞数 9

文章标签： langchain 数据库人工智能 AI大模型大模型应用 AI LLM

本文链接：https://blog.csdn.net/python1222_/article/details/147924171

版权

基本功能

1. 链式调用

# 创建简单的链式调用
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
# 定义提示模板
prompt = PromptTemplate(
    input_variables=["contract_text"],
    template="请分析以下合同条款的风险点：{contract_text}"
)
# 初始化大语言模型
llm = OpenAI(temperature=0)
# 创建链条
chain = LLMChain(llm=llm, prompt=prompt)
# 运行链条
result = chain.run("甲方同意在任何情况下承担所有责任...")
print(result)

2. 文档处理

# 加载和处理文档
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 加载PDF文档
loader = PyPDFLoader("合同.pdf")
documents = loader.load()
# 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = text_splitter.split_documents(documents)

在这里插入图片描述

实用功能

1. 向量存储检索

# 使用向量存储进行语义搜索
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
# 创建嵌入模型
embeddings = OpenAIEmbeddings()
# 创建向量数据库
db = Chroma.from_documents(splits, embeddings)
# 相似度搜索
query = "违约责任条款"
similar_docs = db.similarity_search(query, k=3)
# 检索增强生成
from langchain.chains import RetrievalQA
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=db.as_retriever()
)
answer = qa_chain.run("请分析合同中的违约责任条款")

2. 代理和工具使用

# 创建能够使用工具的智能代理
from langchain.agents import initialize_agent, Tool
from langchain.tools import BaseTool
# 定义工具函数
def search_contracts(query):
    # 搜索合同库的逻辑
    return "找到3份相关合同..."
# 创建工具
tools = [
    Tool(
        name="ContractSearch",
        func=search_contracts,
        description="搜索合同数据库"
    )
]
# 初始化代理
agent = initialize_agent(
    tools,
    llm,
    agent="zero-shot-react-description",
    verbose=True
)
# 运行代理
agent.run("寻找所有提到不可抗力条款的合同")

3. 批量合同处理

# 批量处理多份合同
from langchain.callbacks import get_openai_callback
import os
from concurrent.futures import ThreadPoolExecutor
# 批处理函数
def process_contract(file_path):
    loader = PyPDFLoader(file_path)
    doc = loader.load()
    # 风险分析链
    risk_chain = LLMChain(
        llm=llm,
        prompt=PromptTemplate(
            template="分析合同风险点：{text}",
            input_variables=["text"]
        )
    )
    # 提取文本
    text = "\n".join([page.page_content for page in doc])
    # 分析风险
    return risk_chain.run(text)
# 并行处理多份合同
contract_dir = "contracts/"
contract_files = [os.path.join(contract_dir, f) for f in os.listdir(contract_dir)]
# 使用线程池加速处理
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(process_contract, contract_files))

Generated Image

常见任务示例

1. 智能合同审核系统

# 智能合同审核系统
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
from langchain.document_loaders import DirectoryLoader
import time
# 加载法律知识库
knowledge_loader = DirectoryLoader("法律知识库/", glob="**/*.pdf")
knowledge_docs = knowledge_loader.load()
knowledge_db = Chroma.from_documents(
    knowledge_docs,
    embeddings,
    collection_name="legal_knowledge"
)
# 创建审核记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)
# 智能审核链
def create_contract_review_chain():
    return ConversationalRetrievalChain.from_llm(
        llm=OpenAI(temperature=0),
        retriever=knowledge_db.as_retriever(),
        memory=memory
    )
# 创建审核函数
def audit_contract(contract_path, review_chain):
    # 加载合同
    loader = PyPDFLoader(contract_path)
    contract = loader.load()
    text = "\n".join([p.page_content for p in contract])
    # 定义审核点
    audit_points = [
        "是否存在对我方不利的违约条款",
        "责任界定是否明确",
        "付款条件是否合理",
        "是否有法律风险点"
    ]
    results = {}
    for point in audit_points:
        query = f"针对以下合同：{text[:1000]}...\n审核重点：{point}"
        response = review_chain({"question": query})
        results[point] = response["answer"]
    return results
# 批量审核
start_time = time.time()
contract_paths = ["合同1.pdf", "合同2.pdf", "合同3.pdf"]
review_chain = create_contract_review_chain()
audit_results = {}
for path in contract_paths:
    audit_results[path] = audit_contract(path, review_chain)
end_time = time.time()
print(f"审核完成，耗时：{end_time - start_time:.2f}秒")

Generated Image

2. 合同风险评分工具

# 合同风险评分工具
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional
# 定义风险评分模型
class RiskScore(BaseModel):
    overall_score: int = Field(description="总体风险评分，1-10")
    risk_areas: List[str] = Field(description="风险领域列表")
    recommendations: List[str] = Field(description="改进建议")
    confidence: float = Field(description="评分置信度，0-1")
# 创建输出解析器
parser = PydanticOutputParser(pydantic_object=RiskScore)
# 风险评分模板
risk_template = """
分析以下合同并给出风险评分:
{contract_text}
{format_instructions}
"""
prompt = PromptTemplate(
    template=risk_template,
    input_variables=["contract_text"],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)
# 风险评分链
risk_chain = LLMChain(
    llm=OpenAI(temperature=0),
    prompt=prompt,
    output_parser=parser
)
# 评分函数
def score_contract_risk(contract_path):
    loader = PyPDFLoader(contract_path)
    doc = loader.load()
    text = "\n".join([p.page_content for p in doc])
    # 限制文本长度
    max_length = 4000
    if len(text) > max_length:
        text = text[:max_length]
    # 获取风险评分
    try:
        result = risk_chain.run(text)
        return result
    except Exception as e:
        return f"评分出错: {str(e)}"
# 使用示例
sample_risk = score_contract_risk("样例合同.pdf")
print(f"风险评分: {sample_risk.overall_score}/10")
print(f"风险领域: {', '.join(sample_risk.risk_areas)}")
print(f"改进建议: {sample_risk.recommendations}")