基于LLM的智能票据分类系统:从原理到部署实践

基于LLM的智能票据分类系统:从原理到部署实践
如果你还在手动整理报销票据或者为团队报销流程的低效而头疼那么今天介绍的这个开源项目可能会改变你的工作方式。Sorted Receipts 解决了一个看似简单但实际很麻烦的问题如何让多人高效地上传和分类各种票据。传统的做法是建立共享文件夹、使用在线表格或者更原始的——邮件来回发送图片。这些方法要么无法自动分类要么需要大量人工干预。而 Sorted Receipts 的核心思路是提供一个统一的上传链接用户通过这个链接提交票据图片或PDF后台的 LLM大语言模型自动识别票据内容并分类。这不仅仅是又一个AI识别票据的工具。关键差异在于它的工作流设计轻量级的前端交互配合可配置的LLM后端。用户无需安装任何应用上传后系统自动处理支持自定义分类规则并能导出结构化数据。对于中小企业、自由职业团队或需要频繁处理报销的部门来说这种链接即服务的模式大幅降低了使用门槛。本文将带你从零部署 Sorted Receipts重点解析三个核心环节环境配置与依赖安装、LLM集成与票据识别逻辑、自定义分类与数据导出。我们不仅会给出详细的代码示例还会分享实际部署中容易遇到的坑——比如如何选择性价比高的LLM服务、如何处理模糊图片、以及如何设置分类规则避免误判。1. 这篇文章真正要解决的问题在企业报销或费用管理流程中票据收集与分类一直是效率瓶颈。典型场景包括销售团队外出产生的交通、餐饮票据需要统一提交项目组采购办公用品或服务的发票需要归档远程团队成员分散在各地票据格式五花八门传统解决方案的痛点很明显邮件收集票据散落在不同邮件中整理耗时易错共享文件夹缺乏自动分类后期需要人工筛选专业报销软件成本高、流程复杂不适合小型团队Sorted Receipts 的突破点在于它用最简化的前端一个上传链接承接用户输入用可配置的LLM后端实现智能分类。这种设计的好处是对上传者零门槛不需要解释请按日期命名文件或发票和收据分开放对管理员可定制可以根据公司财务要求设置分类规则成本可控基于开源框架LLM服务可按需选择从本地部署到云端API更重要的是这个项目演示了如何将LLM能力产品化到一个具体业务场景中——不是炫技而是解决真实痛点。下面我们就从基础概念开始逐步拆解它的实现原理。2. 基础概念与核心原理2.1 Sorted Receipts 的架构组成Sorted Receipts 本质上是一个微服务架构的应用主要包含三个模块前端上传接口提供一个稳定的HTTP链接接收多文件上传文件预处理管道对上传的图片或PDF进行格式转换、文字提取预处理LLM分类引擎调用大语言模型分析票据内容按规则分类2.2 LLM在票据识别中的特殊价值你可能会问OCR光学字符识别技术已经很成熟为什么还需要LLM关键在于语义理解与上下文判断。举例说明一张餐饮小票上既有食物项也有服务费OCR只能提取文字LLM能判断这属于业务招待还是团队聚餐一张模糊的出租车票OCR可能识别失败LLM可以根据残留的出租字样和金额范围推断类别跨语言票据如英文发票中的Tax和中文发票的税LLM能统一归类到税费科目这种理解能力来自于LLM在大量文本数据上的预训练让它能够处理OCR输出中的噪声、不完整信息和多义性。2.3 关键术语解释术语解释在项目中的作用票据分类根据内容将票据归到预设类别核心功能如交通费、办公用品统一上传链接一个固定的URL用于接收所有上传简化用户操作避免配置困扰LLM提示工程设计给LLM的指令引导其输出结构化结果决定分类准确性的关键数据导出将分类结果转换为CSV或JSON与现有财务系统对接3. 环境准备与前置条件在开始部署前请确保你的环境满足以下要求3.1 系统与环境要求操作系统Linux (Ubuntu 20.04 推荐) 或 macOSWindows可通过WSL运行Python版本3.8-3.113.9推荐避免使用已停止支持的版本内存至少4GB可用内存如果本地运行LLM需要更多网络能正常访问PyPI和可能的LLM API服务3.2 核心依赖说明Sorted Receipts 依赖几个关键库各自的作用如下# 核心依赖功能说明 - fastapi提供上传接口和Web界面 - pydantic数据验证和设置管理 - python-multipart处理文件上传 - pillow图像预处理尺寸调整、格式转换 - pytesseract或easyocrOCR文字提取 - openai或llama-cpp-pythonLLM集成重要提醒如果你计划使用云端LLM服务如OpenAI GPT需要提前准备相应的API密钥。如果希望本地运行可以考虑Llama.cpp等开源方案但需要足够的计算资源。4. 完整部署与配置步骤4.1 第一步获取项目代码Sorted Receipts是开源项目可以直接从GitHub克隆# 克隆项目代码 git clone https://github.com/username/sorted-receipts.git cd sorted-receipts # 创建虚拟环境推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装依赖 pip install -r requirements.txt4.2 第二步配置文件设置项目使用环境变量管理配置创建.env文件# 创建配置文件 cp .env.example .env编辑.env文件根据你的需求调整以下关键配置# LLM服务配置以OpenAI为例 LLM_PROVIDERopenai OPENAI_API_KEYyour_api_key_here LLM_MODELgpt-3.5-turbo # 应用基础配置 UPLOAD_FOLDER./uploads MAX_FILE_SIZE10485760 # 10MB ALLOWED_EXTENSIONSpdf,png,jpg,jpeg # 分类规则配置JSON格式 CATEGORY_RULES{transport: [出租车, 地铁, 机票], meal: [餐厅, 外卖, 咖啡]}4.3 第三步启动应用服务Sorted Receipts使用FastAPI框架启动命令如下# 开发环境启动 uvicorn main:app --reload --host 0.0.0.0 --port 8000 # 生产环境建议使用更稳定的方式 # 使用gunicorn需要先安装pip install gunicorn gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000启动成功后访问 http://localhost:8000 可以看到上传界面http://localhost:8000/docs 查看API文档。5. 核心代码解析与自定义5.1 文件上传接口实现上传接口的设计要点是简单、稳定、支持批量。以下是核心代码# 文件routers/upload.py from fastapi import APIRouter, File, UploadFile, HTTPException from fastapi.responses import JSONResponse import os from datetime import datetime router APIRouter() router.post(/upload) async def upload_receipts(files: list[UploadFile] File(...)): 处理多文件上传保存文件并返回处理ID if not files: raise HTTPException(status_code400, detail没有上传文件) # 生成批处理ID batch_id datetime.now().strftime(%Y%m%d_%H%M%S) saved_files [] for file in files: # 验证文件类型 if not allowed_file(file.filename): continue # 生成安全文件名 filename f{batch_id}_{secure_filename(file.filename)} file_path os.path.join(UPLOAD_FOLDER, filename) # 保存文件 with open(file_path, wb) as buffer: content await file.read() buffer.write(content) saved_files.append({ original_name: file.filename, saved_path: file_path, file_size: len(content) }) # 异步触发处理流程 process_receipts.delay(batch_id, saved_files) return JSONResponse({ batch_id: batch_id, message: f成功上传 {len(saved_files)} 个文件, status_url: f/status/{batch_id} }) def allowed_file(filename: str) - bool: 检查文件扩展名是否允许 allowed_extensions {pdf, png, jpg, jpeg} return . in filename and \ filename.rsplit(., 1)[1].lower() in allowed_extensions5.2 LLM分类提示工程分类准确性的关键在于给LLM的提示词设计。以下是优化后的提示词模板# 文件services/llm_classifier.py def build_classification_prompt(text_content: str, categories: dict) - str: 构建LLM分类提示词 prompt f 你是一个专业的财务助理需要根据票据内容将其分类。 可用的分类类别和关键词 {format_categories(categories)} 票据内容OCR提取文本 {text_content} 请按以下JSON格式回复 {{ category: 最匹配的类别名称, confidence: 置信度0-1, amount: 识别出的金额, date: 识别出的日期, reason: 分类理由 }} 要求 1. 如果无法确定类别category设为unknown 2. 金额格式化为数字如123.45 3. 日期格式化为YYYY-MM-DD 4. 置信度基于匹配程度评估 return prompt def format_categories(categories: dict) - str: 格式化分类规则用于提示词 formatted [] for category, keywords in categories.items(): formatted.append(f- {category}: {, .join(keywords)}) return \n.join(formatted)5.3 分类结果处理与导出处理完的票据数据需要结构化存储和导出能力# 文件services/result_handler.py import json import csv from typing import List, Dict class ResultHandler: def __init__(self, output_dir: str ./results): self.output_dir output_dir os.makedirs(output_dir, exist_okTrue) def save_json(self, batch_id: str, results: List[Dict]): 保存JSON格式结果 filename f{batch_id}_results.json filepath os.path.join(self.output_dir, filename) with open(filepath, w, encodingutf-8) as f: json.dump({ batch_id: batch_id, processed_at: datetime.now().isoformat(), results: results }, f, ensure_asciiFalse, indent2) return filepath def save_csv(self, batch_id: str, results: List[Dict]): 保存CSV格式结果便于导入Excel filename f{batch_id}_results.csv filepath os.path.join(self.output_dir, filename) with open(filepath, w, newline, encodingutf-8) as f: if results: fieldnames results[0].keys() writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(results) return filepath def get_summary(self, results: List[Dict]) - Dict: 生成处理摘要 category_count {} total_amount 0.0 for result in results: category result.get(category, unknown) category_count[category] category_count.get(category, 0) 1 amount result.get(amount, 0) if isinstance(amount, (int, float)): total_amount amount return { total_receipts: len(results), category_distribution: category_count, total_amount: round(total_amount, 2) }6. 实际测试与效果验证6.1 测试数据准备为了验证系统效果建议准备不同类型的测试票据# 测试用例示例 test_cases [ { description: 出租车票测试, expected_category: transport, sample_text: 北京市出租车发票 金额: 45.00 日期: 2024-03-15 }, { description: 餐饮发票测试, expected_category: meal, sample_text: 星巴克咖啡 金额: 38.50 日期: 2024-03-14 }, { description: 模糊票据测试, expected_category: unknown, sample_text: 发票 金额: 120 日期: 无法识别 } ]6.2 运行测试流程启动服务后可以通过API或界面进行测试# 使用curl测试上传接口 curl -X POST http://localhost:8000/upload \ -F filestaxi_receipt.jpg \ -F filesrestaurant_bill.pdf \ -H Content-Type: multipart/form-data预期返回结果{ batch_id: 20240315_143022, message: 成功上传 2 个文件, status_url: /status/20240315_143022 }6.3 检查处理状态和结果通过status_url检查处理进度# 查询处理状态 curl http://localhost:8000/status/20240315_143022处理完成后典型的返回结果如下{ batch_id: 20240315_143022, status: completed, results: [ { filename: taxi_receipt.jpg, category: transport, confidence: 0.95, amount: 45.00, date: 2024-03-15, reason: 匹配到出租车关键词 }, { filename: restaurant_bill.pdf, category: meal, confidence: 0.88, amount: 256.00, date: 2024-03-14, reason: 识别到餐厅消费特征 } ], summary: { total_receipts: 2, category_distribution: {transport: 1, meal: 1}, total_amount: 301.00 } }7. 常见问题与排查思路在实际部署中你可能会遇到以下典型问题问题现象可能原因排查方式解决方案上传失败提示文件类型不支持文件扩展名不在允许列表中检查ALLOWED_EXTENSIONS配置添加缺失的扩展名或转换文件格式LLM分类结果不准确提示词设计不合理或分类规则模糊查看LLM的完整响应日志优化提示词增加示例明确分类边界处理速度慢图片尺寸过大或LLM API响应延迟监控每个处理环节耗时添加图片压缩考虑LLM缓存使用异步处理内存使用过高同时处理文件过多或内存泄漏使用内存监控工具限制并发处理数添加处理队列中文识别效果差OCR模型对中文支持不佳测试不同OCR引擎切换至支持中文更好的OCR如PaddleOCR7.1 性能优化建议对于生产环境使用建议进行以下优化# 文件处理并发控制 from concurrent.futures import ThreadPoolExecutor import asyncio class ProcessingPool: def __init__(self, max_workers: int 3): self.executor ThreadPoolExecutor(max_workersmax_workers) async def process_batch(self, batch_files: list): 控制并发处理数量避免资源耗尽 loop asyncio.get_event_loop() tasks [] for file_info in batch_files: task loop.run_in_executor( self.executor, self.process_single_file, file_info ) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) return results8. 最佳实践与工程建议8.1 分类规则设计原则基于实际项目经验分类规则的设计应该遵循互斥性优先类别之间界限清晰避免重叠容错性设计为模糊票据设置待确认类别可扩展结构预留自定义字段适应不同财务需求{ categories: { transport: { keywords: [出租车, 地铁, 机票, 火车票], amount_range: [5, 5000], required_fields: [date, amount] }, meal: { keywords: [餐厅, 外卖, 咖啡, 餐饮], amount_range: [10, 1000], required_fields: [date, amount] }, office_supplies: { keywords: [文具, 打印, 办公用品], amount_range: [1, 10000], required_fields: [date, amount, vendor] } }, fallback_category: unknown, confidence_threshold: 0.7 }8.2 安全与隐私考虑处理财务票据时安全性和隐私保护至关重要文件存储加密敏感票据文件应该加密存储访问日志审计记录所有文件访问和处理操作数据自动清理设置定期清理机制删除过期文件API访问限制实施速率限制和身份验证# 简单的自动清理机制示例 import schedule import time def cleanup_old_files(): 定期清理超过30天的文件 now time.time() for filename in os.listdir(UPLOAD_FOLDER): filepath os.path.join(UPLOAD_FOLDER, filename) if os.path.isfile(filepath): file_age now - os.path.getmtime(filepath) if file_age 30 * 24 * 60 * 60: # 30天 os.remove(filepath) # 每天执行一次清理 schedule.every().day.at(02:00).do(cleanup_old_files)8.3 生产环境部署建议对于企业级使用建议采用以下架构使用Docker容器化部署保证环境一致性配置反向代理Nginx处理静态文件和SSL使用Redis作为任务队列和缓存设置监控告警监控服务健康状态定期备份分类规则和处理结果# Dockerfile示例 FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 8000 CMD [gunicorn, -w, 4, -k, uvicorn.workers.UvicornWorker, main:app, --bind, 0.0.0.0:8000]9. 扩展应用与二次开发Sorted Receipts的基础架构可以扩展到更多场景9.1 多语言支持扩展通过动态提示词切换支持多语言票据def get_localized_prompt(language: str) - str: 根据语言获取本地化提示词 prompts { zh: 你是一个中文财务助理..., en: You are an English-speaking financial assistant..., ja: あなたは日本語の財務アシスタントです... } return prompts.get(language, prompts[zh])9.2 与现有系统集成提供Webhook接口将处理结果推送到现有财务系统app.post(/webhook/{batch_id}) async def trigger_webhook(batch_id: str, webhook_url: str): 处理完成后触发Webhook通知 results get_processing_results(batch_id) async with httpx.AsyncClient() as client: response await client.post( webhook_url, jsonresults, headers{Content-Type: application/json} ) return {status: webhook_triggered, response_status: response.status_code}9.3 自定义处理管道高级用户可以通过插件机制扩展处理流程class ProcessingPipeline: def __init__(self): self.plugins [] def add_plugin(self, plugin): 添加处理插件 self.plugins.append(plugin) def process(self, file_path: str) - Dict: 执行处理管道 result {original_file: file_path} for plugin in self.plugins: result.update(plugin.execute(result)) return result # 示例插件增值税发票识别 class VATInvoicePlugin: def execute(self, context: Dict) - Dict: # 专用增值税发票识别逻辑 return {vat_info: extract_vat_info(context[text_content])}Sorted Receipts的价值不仅在于它提供的现成功能更在于它展示了一种思路如何用现代AI技术解决具体的业务流程痛点。通过本文的详细拆解你应该能够根据实际需求部署、定制甚至扩展这个系统。关键是要理解技术的选择LLM vs 传统OCR服务于业务目标——在这里是降低报销流程的摩擦成本。在实际应用中建议先从一个小团队开始试点收集反馈迭代优化分类规则再逐步推广到更大范围。