Agent文档自动化处理实战：从杂乱数据到结构化知识库

1次阅读

共计 2408 个字符，预计需要花费 7 分钟才能阅读完成。

每次接手新的文档处理需求时，总会被这些场景困扰：合同里关键条款散落在不同段落，财报 PDF 中的表格数据需要手动复制粘贴，海量文档的分类归档消耗团队大量时间。更让人头疼的是：

格式混乱 ：同一份文档可能包含扫描图片、手写注释、复杂表格等混合内容
信息冗余 ：有效信息往往只占文档 20% 内容，其余是模板化文字
标注成本 ：训练监督模型需要数千份标注样本，业务部门根本等不起

正则表达式 ：
只能处理固定模板文档
维护成本随规则数量指数级增长
对版式变化零容错
纯大模型方案 ：
GPT- 3 处理单文档消耗 4GB 内存
API 调用延迟高达 2 - 3 秒 / 页
无法保证金融数据处理的确定性

class DocumentAgent:
    """
    三层处理架构示例
    preprocessor: 格式标准化 / 分页 / 去噪
    nlp_engine: 实体识别 / 关系抽取
    kg_builder: 知识图谱存储
    """
    def __init__(self):
        self.preprocessor = PDFExtractor()
        self.nlp_engine = SpacyNLP()
        self.kg_builder = Neo4jLoader()

from PyPDF2 import PdfReader
from bs4 import BeautifulSoup

def parse_file(file_path: str) -> str:
    """支持 PDF/HTML 的容错解析"""
    try:
        if file_path.endswith('.pdf'):
            with open(file_path, 'rb') as f:
                reader = PdfReader(f)
                return ''.join([page.extract_text() for page in reader.pages])
        else:
            with open(file_path, 'r', encoding='utf-8') as f:
                soup = BeautifulSoup(f.read(), 'html.parser')
                return soup.get_text()
    except Exception as e:
        logging.error(f"解析失败 {file_path}: {str(e)}")
        raise DocumentParseError from e

import spacy
from spacy.matcher import PhraseMatcher

nlp = spacy.load('zh_core_web_lg')

# 添加金融领域专有词典
financial_terms = ["LIBOR", "SWAP", "交叉违约"]
matcher = PhraseMatcher(nlp.vocab)
patterns = [nlp(text) for text in financial_terms]
matcher.add("FIN_TERMS", patterns)

def extract_entities(text: str) -> dict:
    doc = nlp(text)
    return {"orgs": [ent.text for ent in doc.ents if ent.label_ == "ORG"],
        "financial_terms": [doc[start:end].text 
                           for _, start, end in matcher(doc)]
    }

import tracemalloc

def check_memory():
    tracemalloc.start()
    # ... 执行文档处理操作
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')
    for stat in top_stats[:5]:  # 打印内存占用前 5
        print(stat)

import asyncio
from redis import asyncio as aioredis

async def process_queue():
    redis = await aioredis.from_url("redis://localhost")
    while True:
        doc_data = await redis.lpop("doc_queue")
        if not doc_data:
            await asyncio.sleep(1)
            continue
        await process_document(doc_data)

敏感信息过滤 ：

# 匹配身份证 / 银行卡号等
SENSITIVE_PATTERN = r'\b(\d{17}[0-9X]|\d{16})\b'

def sanitize_text(text: str) -> str:
    return re.sub(SENSITIVE_PATTERN, '[REDACTED]', text)

重试机制 ：

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def call_ml_service(data):
    response = requests.post(API_ENDPOINT, json=data)
    response.raise_for_status()
    return response.json()