手写AI合同审核系统：OCR + 混合RAG + LangGraph全栈实战

import cv2
import numpy as np

def enhance_contract_image(image_path: str) -> np.ndarray:
    img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)
    
    # 1. CLAHE 自适应直方图均衡
    clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8,8))
    enhanced = clahe.apply(img)
    
    # 2. 去噪（非局部均值去噪，保留边缘）
    denoised = cv2.fastNlMeansDenoising(enhanced, h=30)
    
    # 3. 二值化
    _, binary = cv2.threshold(denoised, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
    
    return binary

def deskew_image(img: np.ndarray) -> np.ndarray:
    coords = np.column_stack(np.where(img > 0))
    angle = cv2.minAreaRect(coords)[-1]
    
    if angle < -45:
        angle = 90 + angle
    
    (h, w) = img.shape[:2]
    center = (w // 2, h // 2)
    M = cv2.getRotationMatrix2D(center, angle, 1.0)
    return cv2.warpAffine(img, M, (w, h), 
                          flags=cv2.INTER_CUBIC,
                          borderMode=cv2.BORDER_REPLICATE)

from paddleocr import PaddleOCR

ocr = PaddleOCR(use_angle_cls=True, lang='en', use_gpu=False)

def paddle_ocr_page(image: np.ndarray) -> str:
    result = ocr.ocr(image, cls=True)
    lines = []
    for line in result[0]:
        text = line[1][0]
        confidence = line[1][1]
        lines.append(text)
    return '\n'.join(lines)

from openai import OpenAI
import base64

client = OpenAI()

def gpt4v_fallback(image: np.ndarray, paddle_conf: float) -> str:
    if paddle_conf > 0.85:
        return None  # 不需要兜底
    
    # 将图片转为 Base64
    _, buffer = cv2.imencode('.png', image)
    b64 = base64.b64encode(buffer).decode()
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Extract ALL text from this document page. Preserve the original layout order."},
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
            ]
        }],
        max_tokens=4096
    )
    return response.choices[0].message.content

from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever

def build_hybrid_index(chunks: list[str]):
    # 稠密检索：FAISS
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = FAISS.from_texts(chunks, embedding=embeddings)
    dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
    
    # 稀疏检索：BM25
    bm25_retriever = BM25Retriever.from_texts(chunks)
    bm25_retriever.k = 10
    
    # RRF 融合
    ensemble = EnsembleRetriever(
        retrievers=[dense_retriever, bm25_retriever],
        weights=[0.5, 0.5]
    )
    return ensemble

def reciprocal_rank_fusion(results: list[list], k: int = 60) -> list:
    fused_scores = {}
    for rank_list in results:
        for rank, doc in enumerate(rank_list, 1):
            doc_id = doc.metadata.get('id', doc.page_content[:50])
            fused_scores[doc_id] = fused_scores.get(doc_id, 0) + 1 / (k + rank)
    
    ranked = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return ranked

QUERY_VARIANTS = {
    "payment_terms": [
        "What are the payment terms and conditions?",
        "When are payments due? Net 30, Net 60?",
        "Payment schedule and invoicing requirements",
        "Late payment penalties and interest",
        "What currency are payments in?"
    ],
    "termination": [
        "What are the termination clauses?",
        "How can either party terminate this agreement?",
        "Notice period for termination",
        "Termination for convenience or cause",
        "Post-termination obligations"
    ],
    "liability": [
        "What are the liability caps?",
        "Limitation of liability clause",
        "Indemnification obligations",
        "Exclusions from liability limits",
        "Aggregate liability cap amount"
    ]
}

def multi_query_retrieve(retriever, field_name: str, k: int = 5):
    queries = QUERY_VARIANTS[field_name]
    all_docs = []
    for q in queries:
        docs = retriever.get_relevant_documents(q)
        all_docs.extend(docs[:k])
    
    # 去重
    seen = set()
    unique = []
    for d in all_docs:
        key = d.page_content[:100]
        if key not in seen:
            seen.add(key)
            unique.append(d)
    return unique[:k * 2]

CONTRACT_EXTRACTION_PROMPT = """
You are a contract analyst. Extract the following fields from the provided contract text.
For each field, provide:
- value: the extracted value
- confidence: a score from 0.0 to 1.0
- source: a brief snippet from the text that supports this value

Fields to extract:
1. contract_date: Date the contract was signed
2. effective_date: When the contract takes effect
3. expiry_date: Contract expiration or renewal date
4. party_names: Names of all contracting parties
5. payment_terms: Net 30, Net 60, etc.
6. liability_cap: Maximum liability amount
7. governing_law: Jurisdiction and governing law

Contract text:
{context}

Return JSON format.
"""

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional

class ContractState(TypedDict):
    file_path: str
    images: List
    ocr_text: str
    chunks: List[str]
    hybrid_index: object
    extracted_fields: dict
    confidence_scores: dict
    excel_path: Optional[str]
    error: Optional[str]

def build_contract_pipeline():
    workflow = StateGraph(ContractState)
    
    workflow.add_node("enhance_images", enhance_step)
    workflow.add_node("extract_text", ocr_step)
    workflow.add_node("build_index", index_step)
    workflow.add_node("multi_query_rag", rag_step)
    workflow.add_node("extract_fields", extract_step)
    workflow.add_node("generate_excel", excel_step)
    
    workflow.add_conditional_edges(
        "extract_text",
        check_ocr_quality,
        {True: "build_index", False: "extract_text"}  # 质量低则重试
    )
    
    workflow.set_entry_point("enhance_images")
    workflow.add_edge("enhance_images", "extract_text")
    workflow.add_edge("build_index", "multi_query_rag")
    workflow.add_edge("multi_query_rag", "extract_fields")
    workflow.add_edge("extract_fields", "generate_excel")
    workflow.add_edge("generate_excel", END)
    
    return workflow.compile()

from openpyxl import Workbook
from openpyxl.styles import PatternFill, Font

def generate_excel_report(fields: dict, rate_card: list, raw_text: str, output_path: str):
    wb = Workbook()
    
    # Sheet 1: 字段摘要
    ws1 = wb.active
    ws1.title = "Field Summary"
    
    green_fill = PatternFill(start_color="C6EFCE", end_color="C6EFCE", fill_type="solid")
    yellow_fill = PatternFill(start_color="FFEB9C", end_color="FFEB9C", fill_type="solid")
    red_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
    
    ws1.cell(row=1, column=1, value="Field")
    ws1.cell(row=1, column=2, value="Value")
    ws1.cell(row=1, column=3, value="Confidence")
    ws1.cell(row=1, column=4, value="Source Text")
    
    for i, (field, data) in enumerate(fields.items(), 2):
        ws1.cell(row=i, column=1, value=field)
        ws1.cell(row=i, column=2, value=data['value'])
        ws1.cell(row=i, column=3, value=data['confidence'])
        ws1.cell(row=i, column=4, value=data['source'])
        
        # 置信度着色
        conf = data['confidence']
        fill = green_fill if conf >= 0.9 else (yellow_fill if conf >= 0.7 else red_fill)
        ws1.cell(row=i, column=3).fill = fill
    
    # Sheet 2: 费率卡
    ws2 = wb.create_sheet("Rate Card")
    for row_data in rate_card:
        ws2.append(row_data)
    
    # Sheet 3: 原始 OCR 文本
    ws3 = wb.create_sheet("Raw OCR")
    ws3.cell(row=1, column=1, value=raw_text)
    
    wb.save(output_path)