在日常财务管理和报销流程中处理纸质或电子收据往往令人头疼。无论是个人记账还是企业报销收集、分类、汇总各类消费凭证都需要耗费大量时间。特别是当多个客户或团队成员需要提交收据时传统的邮件附件、微信群文件等方式容易造成遗漏和混乱。最近在技术社区看到一个很有意思的项目——Sorted Receipts它通过一个共享链接让客户批量上传收据然后利用大语言模型LLM自动识别和分类。这种思路既解决了收据收集的标准化问题又通过AI技术大幅提升了处理效率。本文将完整解析这类收据智能处理系统的实现原理并手把手教你搭建一个可运行的Demo版本。无论你是想了解LLM在实际业务中的应用还是需要为团队开发一个类似的收据管理工具这篇文章都会提供从技术选型到代码实现的完整指南。我们将使用Python FastAPI作为后端框架结合OCR技术和开源LLM模型实现收据的自动识别和分类。1. 收据处理的技术背景与业务价值1.1 传统收据处理的痛点分析在实际业务场景中收据处理通常面临以下几个核心问题收集环节的标准化缺失不同客户提交的收据格式各异——可能是照片、扫描件、PDF文件甚至是截图。缺乏统一的提交规范导致后续处理困难。人工分类效率低下财务人员需要手动查看每张收据的内容按消费类型如餐饮、交通、办公用品、时间、金额等维度进行分类。这个过程不仅耗时还容易出错。数据提取准确性不足收据上的关键信息商户名称、消费金额、日期、税号等需要人工录入既存在误操作风险又无法保证数据一致性。多源数据整合困难当收据来自不同渠道微信、邮箱、纸质扫描时整合和去重工作复杂容易产生数据孤岛。1.2 LLM在收据处理中的技术优势大语言模型为上述痛点提供了创新的解决方案多模态理解能力现代LLM能够同时处理文本和图像信息可以直接从收据图片中提取结构化数据无需先进行OCR文字识别再处理。上下文感知分类基于收据内容的语义理解LLM可以智能判断消费类别。例如星巴克咖啡会自动归类为餐饮滴滴出行归类为交通。自适应学习机制通过少量样本训练LLM可以学习特定业务的分类规则适应不同企业的报销政策和要求。批量处理能力一个LLM实例可以并行处理大量收据显著提升处理效率降低人力成本。1.3 典型应用场景分析企业费用报销系统员工通过专属链接上传消费凭证系统自动分类并生成报销单财务部门只需审核无需手动录入。个人记账应用用户将日常消费小票拍照上传自动同步到记账软件实现消费行为的可视化分析。税务审计辅助为会计师事务所提供收据智能分类服务快速准备审计材料提高工作效率。零售业数据收集收集客户消费凭证进行市场分析了解消费习惯和偏好。2. 技术架构设计与环境准备2.1 系统整体架构一个完整的智能收据处理系统包含以下核心模块前端界面/API接口 → 文件上传服务 → 存储层(云存储/本地) → OCR识别引擎 → LLM处理模块 → 结果存储 → 管理后台前端交互层提供用户上传界面或API接口支持拖拽上传、批量选择等操作。文件处理层负责接收、验证、存储上传的收据文件支持常见图片格式和PDF文档。OCR识别层将图像中的文字信息提取为结构化文本为后续LLM处理提供输入。LLM分析层核心处理模块对OCR结果进行语义分析提取关键信息并分类。数据存储层保存处理结果和原始文件索引支持查询和统计分析。2.2 开发环境要求基础运行环境操作系统Windows 10/11, macOS 10.14, Ubuntu 18.04Python版本3.8-3.11推荐3.9内存至少8GB处理大量图片时建议16GB存储空间至少10GB可用空间核心依赖包# requirements.txt fastapi0.104.1 uvicorn0.24.0 pydantic2.5.0 python-multipart0.0.6 pillow10.1.0 pytesseract0.3.10 openai1.3.0 transformers4.35.0 torch2.1.0 pdf2image1.16.3 sqlalchemy2.0.23 alembic1.12.1可选LLM后端OpenAI GPT-4V收费识别准确率高本地部署的LLaVA、Qwen-VL等开源多模态模型纯文本LLMGPT-3.5-turbo、ChatGLM等配合OCR预处理2.3 项目目录结构sorted-receipts/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── models/ # 数据模型 │ │ ├── __init__.py │ │ ├── receipt.py # 收据数据模型 │ │ └── response.py # API响应模型 │ ├── services/ # 业务逻辑层 │ │ ├── __init__.py │ │ ├── ocr_service.py # OCR识别服务 │ │ ├── llm_service.py # LLM分析服务 │ │ └── file_service.py # 文件处理服务 │ ├── routers/ # API路由 │ │ ├── __init__.py │ │ ├── upload.py # 文件上传接口 │ │ └── receipts.py # 收据管理接口 │ └── config.py # 配置文件 ├── storage/ # 文件存储目录 │ ├── uploads/ # 原始上传文件 │ └── processed/ # 处理后的文件 ├── tests/ # 测试文件 ├── alembic/ # 数据库迁移 ├── requirements.txt └── README.md3. 核心模块实现详解3.1 文件上传服务实现文件上传是系统的入口需要处理多种格式的收据文件并提供良好的用户体验。基础配置# app/config.py import os from pydantic_settings import BaseSettings class Settings(BaseSettings): # 文件上传配置 MAX_FILE_SIZE: int 10 * 1024 * 1024 # 10MB ALLOWED_EXTENSIONS: set {.jpg, .jpeg, .png, .pdf, .heic} UPLOAD_DIR: str storage/uploads PROCESSED_DIR: str storage/processed # OCR配置 TESSERACT_PATH: str /usr/bin/tesseract # Linux/Mac, Windows需要调整 # LLM配置 OPENAI_API_KEY: str # 可选使用OpenAI时填写 LOCAL_LLM_MODEL: str liuhaotian/llava-v1.5-7b # 本地模型 class Config: env_file .env settings Settings()文件上传接口# app/routers/upload.py import os import uuid from fastapi import APIRouter, UploadFile, File, HTTPException from fastapi.responses import JSONResponse from app.config import settings from app.services.file_service import save_upload_file router APIRouter(prefix/api/upload, tags[upload]) router.post(/receipts) async def upload_receipts(files: list[UploadFile] File(...)): 批量上传收据文件 if len(files) 50: raise HTTPException(status_code400, detail一次最多上传50个文件) results [] for file in files: # 验证文件类型 file_ext os.path.splitext(file.filename)[1].lower() if file_ext not in settings.ALLOWED_EXTENSIONS: results.append({ filename: file.filename, status: rejected, reason: f不支持的文件格式: {file_ext} }) continue # 验证文件大小 content await file.read() if len(content) settings.MAX_FILE_SIZE: results.append({ filename: file.filename, status: rejected, reason: 文件大小超过限制 }) continue # 保存文件 file_id str(uuid.uuid4()) save_path await save_upload_file(content, file_id, file_ext) results.append({ filename: file.filename, file_id: file_id, status: accepted, save_path: save_path }) return JSONResponse(content{results: results})3.2 OCR文字识别模块OCR模块负责从图像中提取文字信息为LLM分析提供文本输入。Tesseract OCR服务# app/services/ocr_service.py import pytesseract from PIL import Image import pdf2image import os from app.config import settings class OCRService: def __init__(self): # 配置Tesseract路径Windows系统需要特殊处理 if os.name nt: # Windows pytesseract.pytesseract.tesseract_cmd rC:\Program Files\Tesseract-OCR\tesseract.exe else: pytesseract.pytesseract.tesseract_cmd settings.TESSERACT_PATH def extract_text_from_image(self, image_path: str) - str: 从图片中提取文字 try: image Image.open(image_path) text pytesseract.image_to_string(image, langchi_simeng) return text.strip() except Exception as e: raise Exception(fOCR识别失败: {str(e)}) def extract_text_from_pdf(self, pdf_path: str) - str: 从PDF中提取文字 try: # 将PDF转换为图片 images pdf2image.convert_from_path(pdf_path) all_text [] for i, image in enumerate(images): text pytesseract.image_to_string(image, langchi_simeng) all_text.append(f第{i1}页:\n{text}) return \n.join(all_text) except Exception as e: raise Exception(fPDF处理失败: {str(e)}) def process_file(self, file_path: str) - dict: 处理单个文件返回OCR结果 file_ext os.path.splitext(file_path)[1].lower() if file_ext in [.pdf]: text self.extract_text_from_pdf(file_path) else: text self.extract_text_from_image(file_path) return { file_path: file_path, extracted_text: text, word_count: len(text.split()) } # 全局实例 ocr_service OCRService()3.3 LLM智能分析模块这是系统的核心利用大语言模型对OCR提取的文本进行智能分析和分类。基础LLM服务类# app/services/llm_service.py import json import logging from typing import Dict, List, Optional from openai import OpenAI from transformers import pipeline logger logging.getLogger(__name__) class LLMReceiptAnalyzer: def __init__(self, use_openai: bool False): self.use_openai use_openai if use_openai: self.client OpenAI(api_keyyour-openai-key) # 实际使用时从配置读取 else: # 使用本地模型示例使用文本分类pipeline self.classifier pipeline( text-classification, modeldistilbert-base-uncased-finetuned-sst-2-english, return_all_scoresTrue ) def analyze_receipt(self, ocr_text: str) - Dict: 分析收据内容提取关键信息并分类 if self.use_openai: return self._analyze_with_openai(ocr_text) else: return self._analyze_with_local_model(ocr_text) def _analyze_with_openai(self, ocr_text: str) - Dict: 使用OpenAI GPT模型分析 prompt f 请分析以下收据文本提取关键信息并按JSON格式返回 收据文本{ocr_text} 请返回以下字段的JSON对象 - merchant_name: 商户名称 - transaction_date: 交易日期YYYY-MM-DD格式 - total_amount: 总金额数字 - currency: 货币类型 - category: 消费类别餐饮、交通、办公、购物、医疗、其他 - items: 商品列表每个商品包含name和price - tax_amount: 税费金额 - confidence: 识别置信度0-1之间 如果某个字段无法识别请设为null。 try: response self.client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.1 ) result_text response.choices[0].message.content # 提取JSON部分 start_idx result_text.find({) end_idx result_text.rfind(}) 1 json_str result_text[start_idx:end_idx] return json.loads(json_str) except Exception as e: logger.error(fOpenAI分析失败: {e}) return self._get_fallback_result(ocr_text) def _analyze_with_local_model(self, ocr_text: str) - Dict: 使用本地模型分析简化版 # 简化的规则引擎实际项目中可以训练专用模型 category self._categorize_receipt(ocr_text) amount self._extract_amount(ocr_text) date self._extract_date(ocr_text) merchant self._extract_merchant(ocr_text) return { merchant_name: merchant, transaction_date: date, total_amount: amount, currency: CNY, category: category, items: [], tax_amount: None, confidence: 0.7 } def _categorize_receipt(self, text: str) - str: 简单的规则分类 text_lower text.lower() categories { 餐饮: [餐厅, 饭店, 咖啡, 奶茶, 快餐, 美食, 酒楼], 交通: [出租车, 滴滴, 地铁, 公交, 加油, 停车, 机票], 办公: [文具, 打印, 复印, 办公用品, 设备], 购物: [商场, 超市, 便利店, 网购, 服装, 电器] } for category, keywords in categories.items(): if any(keyword in text_lower for keyword in keywords): return category return 其他 def _extract_amount(self, text: str) - Optional[float]: 提取金额 import re # 匹配金额模式数字可能的小数点可能的中文元 patterns [ r合计[:]\s*(\d\.?\d*), r金额[:]\s*(\d\.?\d*), r\s*(\d\.?\d*), r¥\s*(\d\.?\d*), r总计[:]\s*(\d\.?\d*) ] for pattern in patterns: match re.search(pattern, text) if match: try: return float(match.group(1)) except ValueError: continue return None def _extract_date(self, text: str) - Optional[str]: 提取日期 import re from datetime import datetime # 多种日期格式匹配 date_patterns [ r(\d{4})年(\d{1,2})月(\d{1,2})日, r(\d{4})-(\d{1,2})-(\d{1,2}), r(\d{4})/(\d{1,2})/(\d{1,2}) ] for pattern in date_patterns: match re.search(pattern, text) if match: try: year, month, day match.groups() date_str f{year}-{month.zfill(2)}-{day.zfill(2)} datetime.strptime(date_str, %Y-%m-%d) # 验证日期有效性 return date_str except ValueError: continue return None def _extract_merchant(self, text: str) - Optional[str]: 提取商户名称 # 简单的商户名称提取逻辑 lines text.split(\n) for line in lines: line line.strip() if len(line) 2 and len(line) 50: # 合理的商户名称长度 # 排除明显不是商户名的行 if not any(keyword in line for keyword in [电话, 地址, 时间, 日期, 编号]): if any(char in line for char in [公司, 店, 超市, 商场, 餐厅]): return line # 如果是首行且包含中文或英文单词 if lines.index(line) 3 and (any(\u4e00 char \u9fff for char in line) or any(word in line for word in [CO, LTD, INC, STORE])): return line return None def _get_fallback_result(self, ocr_text: str) - Dict: 获取降级处理结果 return { merchant_name: None, transaction_date: None, total_amount: None, currency: None, category: 其他, items: [], tax_amount: None, confidence: 0.1 }4. 完整系统集成与测试4.1 主应用入口配置# app/main.py from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles import os from app.routers import upload, receipts from app.config import settings # 创建FastAPI应用 app FastAPI( titleSorted Receipts API, description智能收据分类处理系统, version1.0.0 ) # 配置CORS app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应限制具体域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 挂载静态文件目录 os.makedirs(settings.UPLOAD_DIR, exist_okTrue) os.makedirs(settings.PROCESSED_DIR, exist_okTrue) app.mount(/storage, StaticFiles(directorystorage), namestorage) # 注册路由 app.include_router(upload.router) app.include_router(receipts.router) app.get(/) async def root(): return {message: Sorted Receipts API Server is running} app.get(/health) async def health_check(): return {status: healthy, version: 1.0.0} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)4.2 收据处理工作流# app/services/processing_service.py import asyncio import logging from typing import List from app.services.ocr_service import ocr_service from app.services.llm_service import LLMReceiptAnalyzer from app.models.receipt import Receipt, ReceiptStatus logger logging.getLogger(__name__) class ReceiptProcessingService: def __init__(self): self.llm_analyzer LLMReceiptAnalyzer(use_openaiFalse) # 使用本地模型 async def process_receipts_batch(self, file_paths: List[str]) - List[Receipt]: 批量处理收据文件 tasks [self.process_single_receipt(file_path) for file_path in file_paths] results await asyncio.gather(*tasks, return_exceptionsTrue) processed_receipts [] for result in results: if isinstance(result, Exception): logger.error(f处理失败: {result}) continue processed_receipts.append(result) return processed_receipts async def process_single_receipt(self, file_path: str) - Receipt: 处理单个收据文件 try: # 1. OCR文字提取 ocr_result ocr_service.process_file(file_path) if not ocr_result[extracted_text] or ocr_result[word_count] 5: return Receipt( file_pathfile_path, statusReceiptStatus.FAILED, error_messageOCR未提取到有效文字 ) # 2. LLM智能分析 analysis_result self.llm_analyzer.analyze_receipt(ocr_result[extracted_text]) # 3. 构建收据对象 receipt Receipt( file_pathfile_path, original_textocr_result[extracted_text], merchant_nameanalysis_result.get(merchant_name), transaction_dateanalysis_result.get(transaction_date), total_amountanalysis_result.get(total_amount), currencyanalysis_result.get(currency), categoryanalysis_result.get(category), tax_amountanalysis_result.get(tax_amount), confidenceanalysis_result.get(confidence, 0), statusReceiptStatus.PROCESSED, items_jsonanalysis_result.get(items, []) ) logger.info(f收据处理完成: {file_path}, 分类: {receipt.category}) return receipt except Exception as e: logger.error(f处理收据失败 {file_path}: {e}) return Receipt( file_pathfile_path, statusReceiptStatus.FAILED, error_messagestr(e) )4.3 数据模型定义# app/models/receipt.py from datetime import datetime from typing import Optional, List, Dict, Any from pydantic import BaseModel from enum import Enum class ReceiptStatus(str, Enum): PENDING pending PROCESSING processing PROCESSED processed FAILED failed class ReceiptItem(BaseModel): name: str price: float quantity: int 1 class Receipt(BaseModel): id: Optional[str] None file_path: str original_text: Optional[str] None merchant_name: Optional[str] None transaction_date: Optional[str] None total_amount: Optional[float] None currency: Optional[str] None category: str 其他 tax_amount: Optional[float] None confidence: float 0.0 status: ReceiptStatus ReceiptStatus.PENDING error_message: Optional[str] None created_at: datetime datetime.now() items_json: List[Dict[str, Any]] [] def to_dict(self) - Dict[str, Any]: return { id: self.id, merchant_name: self.merchant_name, transaction_date: self.transaction_date, total_amount: self.total_amount, currency: self.currency, category: self.category, confidence: self.confidence, status: self.status.value }5. 前端界面与用户体验5.1 简单的上传界面虽然Sorted Receipts的核心是API服务但提供一个简单的前端界面可以大大改善用户体验。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title智能收据分类系统/title style .container { max-width: 800px; margin: 0 auto; padding: 20px; } .upload-area { border: 2px dashed #ccc; padding: 40px; text-align: center; margin: 20px 0; border-radius: 10px; } .file-list { margin: 20px 0; } .file-item { padding: 10px; border: 1px solid #ddd; margin: 5px 0; border-radius: 5px; } .progress-bar { height: 20px; background: #f0f0f0; border-radius: 10px; margin: 10px 0; } .progress { height: 100%; background: #4CAF50; border-radius: 10px; width: 0%; transition: width 0.3s; } /style /head body div classcontainer h1智能收据分类系统/h1 div classupload-area iddropZone p拖拽收据文件到这里或点击选择文件/p input typefile idfileInput multiple accept.jpg,.jpeg,.png,.pdf styledisplay: none; button onclickdocument.getElementById(fileInput).click()选择文件/button /div div classfile-list idfileList/div button onclickuploadFiles() iduploadBtn disabled开始处理/button div classprogress-bar div classprogress idprogressBar/div /div div idresults/div /div script let selectedFiles []; // 文件选择处理 document.getElementById(fileInput).addEventListener(change, function(e) { selectedFiles Array.from(e.target.files); updateFileList(); }); // 拖拽功能 const dropZone document.getElementById(dropZone); dropZone.addEventListener(dragover, (e) { e.preventDefault(); dropZone.style.borderColor #4CAF50; }); dropZone.addEventListener(dragleave, () { dropZone.style.borderColor #ccc; }); dropZone.addEventListener(drop, (e) { e.preventDefault(); dropZone.style.borderColor #ccc; selectedFiles Array.from(e.dataTransfer.files); updateFileList(); }); function updateFileList() { const fileList document.getElementById(fileList); const uploadBtn document.getElementById(uploadBtn); fileList.innerHTML selectedFiles.map(file div classfile-item${file.name} (${(file.size/1024/1024).toFixed(2)}MB)/div ).join(); uploadBtn.disabled selectedFiles.length 0; } async function uploadFiles() { const formData new FormData(); selectedFiles.forEach(file formData.append(files, file)); try { const response await fetch(/api/upload/receipts, { method: POST, body: formData }); const result await response.json(); displayResults(result); } catch (error) { alert(上传失败: error.message); } } function displayResults(result) { const resultsDiv document.getElementById(results); resultsDiv.innerHTML h3处理结果/h3 JSON.stringify(result, null, 2); } /script /body /html6. 部署与生产环境配置6.1 Docker容器化部署为了便于部署我们可以使用Docker将整个应用容器化。DockerfileFROM python:3.9-slim # 安装系统依赖 RUN apt-get update apt-get install -y \ tesseract-ocr \ tesseract-ocr-chi-sim \ poppler-utils \ rm -rf /var/lib/apt/lists/* # 设置工作目录 WORKDIR /app # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建存储目录 RUN mkdir -p storage/uploads storage/processed # 暴露端口 EXPOSE 8000 # 启动命令 CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]docker-compose.ymlversion: 3.8 services: sorted-receipts: build: . ports: - 8000:8000 volumes: - ./storage:/app/storage - ./logs:/app/logs environment: - OPENAI_API_KEY${OPENAI_API_KEY} restart: unless-stopped # 可选添加Redis用于任务队列 redis: image: redis:alpine ports: - 6379:6379 restart: unless-stopped # 可选添加PostgreSQL数据库 postgres: image: postgres:13 environment: POSTGRES_DB: receipts POSTGRES_USER: postgres POSTGRES_PASSWORD: password volumes: - postgres_data:/var/lib/postgresql/data restart: unless-stopped volumes: postgres_data:6.2 环境变量配置创建.env文件管理敏感配置# 数据库配置 DATABASE_URLpostgresql://postgres:passwordlocalhost:5432/receipts # OpenAI配置可选 OPENAI_API_KEYyour_openai_api_key_here # 应用配置 DEBUGfalse LOG_LEVELINFO MAX_WORKERS4 # 文件存储 UPLOAD_DIR/app/storage/uploads MAX_FILE_SIZE104857606.3 性能优化建议数据库优化-- 为常用查询字段创建索引 CREATE INDEX idx_receipts_category ON receipts(category); CREATE INDEX idx_receipts_date ON receipts(transaction_date); CREATE INDEX idx_receipts_status ON receipts(status);缓存策略# 使用Redis缓存频繁访问的收据分类结果 import redis import json class ReceiptCache: def __init__(self): self.redis_client redis.Redis(hostlocalhost, port6379, db0) def get_cached_category(self, text_hash: str) - Optional[str]: cached self.redis_client.get(fcategory:{text_hash}) return cached.decode() if cached else None def set_cached_category(self, text_hash: str, category: str, expire: int 3600): self.redis_client.setex(fcategory:{text_hash}, expire, category)7. 常见问题与解决方案7.1 OCR识别准确率问题问题现象中文文字识别错误数字和金额识别不准确排版复杂的收据识别效果差解决方案# 改进的OCR预处理 def preprocess_image_for_ocr(image_path: str) - Image.Image: OCR前图像预处理 image Image.open(image_path) # 转换为灰度图 if image.mode ! L: image image.convert(L) # 增强对比度 from PIL import ImageEnhance enhancer ImageEnhance.Contrast(image) image enhancer.enhance(2.0) # 对比度增强 # 锐化处理 enhancer ImageEnhance.Sharpness(image) image enhancer.enhance(2.0) return image # 使用多语言OCR配置 def improve_ocr_accuracy(image_path: str) - str: 提高OCR准确率的多重尝试 preprocessed_image preprocess_image_for_ocr(image_path) # 尝试多种OCR配置 configs [ --psm 6 -l chi_simeng, # 统一文本块 --psm 4 -l chi_simeng, # 假设有一列文本 --psm 8 -l chi_simeng, # 单个单词 ] best_result for config in configs: try: text pytesseract.image_to_string(preprocessed_image, configconfig) if len(text) len(best_result): best_result text except: continue return best_result7.2 LLM分类准确性优化问题现象消费类别判断错误金额提取不准确商户名称识别偏差优化策略class EnhancedReceiptAnalyzer(LLMReceiptAnalyzer): def __init__(self): super().__init__() # 加载自定义分类规则 self.custom_rules self.load_custom_rules() def load_custom_rules(self) - Dict: 加载业务特定的分类规则 return { 餐饮: { keywords: [餐厅, 饭店, 咖啡, 奶茶, 快餐, 酒楼, 美食, 烧烤, 火锅], merchant_patterns: [r.*餐厅$, r.*饭店$, r.*咖啡, r.*奶茶] }, 交通: { keywords: [出租车, 滴滴, 地铁, 公交, 加油, 停车, 机票, 火车, 高铁], merchant_patterns: [r.*出行, r.*打车, r.*航空, r中国石化, r中国石油] } } def enhanced_categorize(self, text: str, merchant: str) - str: