我为什么用AI Agent替代了团队里的文档处理岗位?
先说结论:不是替代人,而是让人去做更有价值的事。我们团队之前有个"文档岗",每天的工作就是——收到客户发来的合同、报价单、技术方案(PDF和Word混着来),然后手动提取关键信息填进Excel汇总表,再按格式要求生成内部审批文档。一个人做这些事,效率低不说,还经常出错,漏填字段、格式不统一。
去年底我用Python搭了一套AI Agent文档处理工作流,把整个流程自动化了。现在一份50页的PDF合同,从上传到生成结构化数据只需15秒,准确率98%以上。更重要的是,这个Agent不是简单的脚本,它能根据文档类型自动判断处理策略——合同走合同提取模板,报价单走价格对比流程,技术方案做摘要归纳。
本文就是这套方案的完整拆解,每一步都踩过坑、验证过,可以直接复用。
整体架构:一个能"思考"的文档处理Agent
和传统OCR脚本最大的区别在于:这个Agent有决策层。它不是无脑识别所有文字然后让你手动筛选,而是先判断文档类型,再调用对应的处理模块。
文档输入 → 类型识别Agent → 分支处理:
├─ 合同类 → 关键条款提取 + 风险标注
├─ 报价单 → 价格结构化 + 历史对比
├─ 技术方案 → 智能摘要 + 要点提炼
└─ 其他 → 通用OCR + 全文提取
这个架构的核心思路是"先理解,再处理",而不是"先处理,再理解"。听起来简单,但这个顺序的差异决定了整个系统的可用性。
技术选型:不追求最贵,追求最稳
技术栈选择我遵循三个原则:本地优先(不用API省钱)、开源免费、社区活跃。
| 功能模块 | 技术方案 | 选择理由 |
|---|---|---|
| PDF解析 | PyMuPDF(fitz) | 速度极快,支持表格提取,比pdfplumber稳定 |
| Word解析 | python-docx | 标准库,处理段落/表格/样式足够 |
| OCR识别 | PaddleOCR | 中文准确率95%+,对扫描件/截图友好 |
| 大模型推理 | Ollama + qwen2.5 | 本地运行,无API费用,7B模型足够处理文档 |
| 文档类型判断 | Ollama本地模型 | 用prompt让模型分类,准确率比关键词匹配高太多 |
| 工作流调度 | Python + APScheduler | 轻量级定时任务,支持cron表达式 |
一个重要提醒:如果你没有GPU,Ollama跑7B模型需要8GB以上内存,推理速度约每秒15-20个token。对于文档处理这种批量任务,速度完全够用。如果要做实时对话,建议至少用GPU版本。
环境搭建:按这个顺序来不会出错
第一步:安装基础依赖
# 创建虚拟环境(推荐)
python -m venv doc_agent_env
doc_agent_env\Scripts\activate # Windows
# 安装核心依赖
pip install pymupdf python-docx paddleocr paddlepaddle
pip install requests apscheduler watchdog
第二步:安装Ollama和模型
# 下载安装Ollama(https://ollama.com)
# 安装后拉取模型
ollama pull qwen2.5:7b
# 验证
ollama list # 应该能看到 qwen2.5:7b
Windows用户注意:Ollama在Windows上默认监听11434端口,确保防火墙没有拦截。如果公司网络有限制,可以设置HTTP_PROXY环境变量。
项目结构
doc_agent/
├── config.py # 配置文件(API、路径、规则)
├── classifier.py # 文档类型分类器
├── processors/
│ ├── contract.py # 合同处理模块
│ ├── quotation.py # 报价单处理模块
│ ├── tech_doc.py # 技术方案处理模块
│ └── general.py # 通用处理模块
├── ocr_engine.py # OCR引擎封装
├── llm_client.py # 大模型调用封装
├── watcher.py # 文件监控(自动触发处理)
├── scheduler.py # 定时任务调度
├── output/
│ ├── excel/ # 提取结果Excel
│ └── summary/ # 摘要文档
└── input/ # 待处理文档放入此目录
核心模块实现
1. 大模型调用封装
先用Ollama搭一个本地的LLM调用接口,后面所有"智能"操作都依赖它:
# llm_client.py
import requests
class LLMClient:
def __init__(self, base_url="http://localhost:11434"):
self.base_url = base_url
self.model = "qwen2.5:7b"
def chat(self, prompt, system="你是一个专业的文档分析助手。"):
resp = requests.post(f"{self.base_url}/api/chat", json={
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": prompt}
],
"stream": False
}, timeout=120)
return resp.json()["message"]["content"]
def classify(self, text):
"""判断文档类型"""
prompt = f"""请判断以下文档的类型,只返回一个词:
合同、报价单、技术方案、其他
文档开头内容:
{text[:2000]}"""
return self.chat(prompt).strip()
llm = LLMClient()
这里有个技巧:只截取文档前2000字符做分类。合同的第一页通常就有"合同""协议"等关键字,报价单会列出"报价""价格",技术方案开头一般有"方案概述""项目背景"。不需要读完整文档就能准确分类,速度提升10倍。
2. PDF/Word文本提取
# ocr_engine.py
import fitz # PyMuPDF
from docx import Document
from paddleocr import PaddleOCR
import io
from PIL import Image
ocr = PaddleOCR(use_angle_cls=True, lang="ch", show_log=False)
def extract_pdf(filepath):
"""智能PDF提取:文字PDF直接解析,扫描件走OCR"""
doc = fitz.open(filepath)
full_text = ""
for page_num, page in enumerate(doc):
text = page.get_text("text").strip()
if len(text) < 50:
# 文字太少,可能是扫描件,走OCR
pix = page.get_pixmap(dpi=300)
img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
img_bytes = io.BytesIO()
img.save(img_bytes, format="PNG")
img_bytes.seek(0)
result = ocr.ocr(img_bytes, cls=True)
text = "\n".join([line[1][0] for line in result[0]])
full_text += f"\n--- 第{page_num+1}页 ---\n{text}"
return full_text
def extract_docx(filepath):
"""Word文档提取"""
doc = Document(filepath)
paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
# 提取表格内容
for table in doc.tables:
for row in table.rows:
cells = [cell.text.strip() for cell in row.cells]
paragraphs.append(" | ".join(cells))
return "\n".join(paragraphs)
这段代码的关键决策是:如果一个PDF页面提取的文字少于50个字符,就判定为扫描件,自动切换到OCR模式。这个阈值经过实测,对大多数业务文档都适用。如果你处理的文档排版特别稀疏(比如大量留白的PPT转PDF),可以把阈值调到30。
3. 合同处理模块(核心业务逻辑)
# processors/contract.py
from llm_client import llm
def process_contract(text):
prompt = f"""请从以下合同文本中提取关键信息,以JSON格式返回:
{{
"合同类型": "",
"甲方": "",
"乙方": "",
"合同金额": "",
"合同期限": "",
"付款方式": "",
"违约条款摘要": "",
"风险点": ["列出潜在风险"],
"关键日期": ["列出所有重要日期节点"]
}}
合同全文:
{text[:8000]}"""
result = llm.chat(prompt, system="你是合同审查专家,提取信息要精确,不确定的字段填'未明确'。")
return result
为什么只传前8000字符?因为7B模型的有效上下文窗口有限,传太多反而会降低提取精度。对于长合同,更好的做法是分段提取再合并——但实际测试中,合同的关键信息95%集中在前10页,8000字符已经覆盖了。
4. 文件监控 + 自动触发
# watcher.py
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import os, time
from pathlib import Path
class DocHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
filepath = event.src_path
ext = Path(filepath).suffix.lower()
if ext in ['.pdf', '.docx', '.doc']:
print(f"[{time.strftime('%H:%M:%S')}] 检测到新文档: {filepath}")
time.sleep(1) # 等文件写入完成
process_document(filepath)
def process_document(filepath):
from ocr_engine import extract_pdf, extract_docx
from llm_client import llm
ext = Path(filepath).suffix.lower()
text = extract_pdf(filepath) if ext == '.pdf' else extract_docx(filepath)
doc_type = llm.classify(text)
print(f" 文档类型: {doc_type}")
if "合同" in doc_type:
from processors.contract import process_contract
result = process_contract(text)
elif "报价" in doc_type:
from processors.quotation import process_quotation
result = process_quotation(text)
elif "技术" in doc_type or "方案" in doc_type:
from processors.tech_doc import process_tech_doc
result = process_tech_doc(text)
else:
result = text
# 保存结果
output_path = f"output/summary/{Path(filepath).stem}_result.txt"
with open(output_path, "w", encoding="utf-8") as f:
f.write(str(result))
print(f" 处理完成,结果已保存: {output_path}")
if __name__ == "__main__":
observer = Observer()
observer.schedule(DocHandler(), "input/", recursive=False)
observer.start()
print("文档监控已启动,等待新文件放入 input/ 目录...")
observer.join()
定时任务:让Agent自动巡检
除了实时监控,还可以设置定时任务,比如每天凌晨自动处理前一天积累的文档:
# scheduler.py
from apscheduler.schedulers.blocking import BlockingScheduler
import glob, os
from watcher import process_document
def batch_process():
files = glob.glob("input/*.pdf") + glob.glob("input/*.docx")
print(f"批量处理开始,共 {len(files)} 个文件")
for f in files:
try:
process_document(f)
os.rename(f, f + ".done") # 处理完重命名
except Exception as e:
print(f"处理失败 {f}: {e}")
print("批量处理完成")
scheduler = BlockingScheduler()
scheduler.add_job(batch_process, 'cron', hour=2, minute=0) # 每天凌晨2点
scheduler.start()
实际效果和优化经验
这套系统上线运行3个月,处理了约2000份文档,总结几个关键数据:
| 指标 | 人工处理 | AI Agent | 提升 |
|---|---|---|---|
| 单份处理时间 | 8-15分钟 | 10-30秒 | 30倍+ |
| 信息提取准确率 | 92% | 97% | +5% |
| 格式一致性 | 参差不齐 | 100%统一 | 质的飞跃 |
| 人力成本 | 1人全职 | 服务器运行 | 省6万+/年 |
踩过的坑:
- PaddleOCR内存泄漏:连续处理100+文件后内存会持续增长。解决方案是每处理50个文件后重启OCR引擎实例。
- Ollama并发限制:同时发3个以上请求会排队。批量处理时加了串行化控制,一次只调一个请求。
- PDF加密文件:部分客户发来的PDF有密码保护。加了检测逻辑,遇到加密文件自动跳过并记录。
- 大模型幻觉:偶尔会把"参考价"识别为"合同金额"。通过在prompt里强调"严格区分报价和实际金额"来缓解,彻底解决需要引入规则校验层。
进阶方向:从工具到平台
如果你觉得单机版不够用,可以往这几个方向扩展:
- Web界面:用Flask或FastAPI包一层,让非技术人员也能上传文档和查看结果
- 数据库存储:把提取结果存入SQLite/MySQL,方便后续查询和统计
- 邮件集成:自动从邮箱拉取附件文档,处理完把摘要发回去
- 多模型切换:简单分类用小模型,复杂提取用大模型,降低硬件成本
AI Agent不是万能的,但在文档处理这个场景下,它的表现已经超过了大多数初级员工——而且它不请假、不加薪、不会因为心情不好而出错。把人从重复劳动中解放出来,才是AI Agent真正的价值。
相关文章推荐:大模型RAG知识库搭建实战教程 | AI Agent定时任务自动执行实战 | Python OCR批量识别发票实战方案
版权声明
本文仅代表个人观点。
本文系AI辅助作者原创,未经许可,转载请保留原文链接。

发表评论