企业官网建设流程全解析

在日常财务管理和报销流程中，处理纸质或电子收据往往令人头疼。无论是个人记账还是企业报销，收集、分类、汇总各类消费凭证都需要耗费大量时间。特别是当多个客户或团队成员需要提交收据时，传统的邮件附件、微信群文件等方式容易造成遗漏和混乱。

最近在技术社区看到一个很有意思的项目——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.04+
• Python版本：3.8-3.11（推荐3.9）
• 内存：至少8GB，处理大量图片时建议16GB+
• 存储空间：至少10GB可用空间

核心依赖包：

# requirements.txt fastapi==0.104.1 uvicorn==0.24.0 pydantic==2.5.0 python-multipart==0.0.6 pillow==10.1.0 pytesseract==0.3.10 openai==1.3.0 transformers==4.35.0 torch==2.1.0 pdf2image==1.16.3 sqlalchemy==2.0.23 alembic==1.12.1

可选LLM后端：

• OpenAI GPT-4V（收费，识别准确率高）
• 本地部署的LLaVA、Qwen-VL等开源多模态模型
• 纯文本LLM（GPT-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.md

3. 核心模块实现详解

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_code=400, 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 = r'C:\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, lang='chi_sim+eng') return text.strip() except Exception as e: raise Exception(f"OCR识别失败: {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, lang='chi_sim+eng') all_text.append(f"第{i+1}页:\n{text}") return "\n".join(all_text) except Exception as e: raise Exception(f"PDF处理失败: {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_key="your-openai-key") # 实际使用时从配置读取 else: # 使用本地模型（示例使用文本分类pipeline） self.classifier = pipeline( "text-classification", model="distilbert-base-uncased-finetuned-sst-2-english", return_all_scores=True ) 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( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], temperature=0.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(f"OpenAI分析失败: {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( title="Sorted Receipts API", description="智能收据分类处理系统", version="1.0.0" ) # 配置CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境应限制具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 挂载静态文件目录 os.makedirs(settings.UPLOAD_DIR, exist_ok=True) os.makedirs(settings.PROCESSED_DIR, exist_ok=True) app.mount("/storage", StaticFiles(directory="storage"), name="storage") # 注册路由 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, host="0.0.0.0", port=8000)

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_openai=False) # 使用本地模型 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_exceptions=True) 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_path=file_path, status=ReceiptStatus.FAILED, error_message="OCR未提取到有效文字" ) # 2. LLM智能分析 analysis_result = self.llm_analyzer.analyze_receipt(ocr_result['extracted_text']) # 3. 构建收据对象 receipt = Receipt( file_path=file_path, original_text=ocr_result['extracted_text'], merchant_name=analysis_result.get('merchant_name'), transaction_date=analysis_result.get('transaction_date'), total_amount=analysis_result.get('total_amount'), currency=analysis_result.get('currency'), category=analysis_result.get('category'), tax_amount=analysis_result.get('tax_amount'), confidence=analysis_result.get('confidence', 0), status=ReceiptStatus.PROCESSED, items_json=analysis_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_path=file_path, status=ReceiptStatus.FAILED, error_message=str(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 lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.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 class="container"> <h1>智能收据分类系统</h1> <div class="upload-area" id="dropZone"> <p>拖拽收据文件到这里，或点击选择文件</p> <input type="file" id="fileInput" multiple accept=".jpg,.jpeg,.png,.pdf" style="display: none;"> <button onclick="document.getElementById('fileInput').click()">选择文件</button> </div> <div class="file-list" id="fileList"></div> <button onclick="uploadFiles()" id="uploadBtn" disabled>开始处理</button> <div class="progress-bar"> <div class="progress" id="progressBar"></div> </div> <div id="results"></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 class="file-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> </html>

6. 部署与生产环境配置

6.1 Docker容器化部署

为了便于部署，我们可以使用Docker将整个应用容器化。

Dockerfile：

FROM 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.yml：

version: '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_URL=postgresql://postgres:password@localhost:5432/receipts # OpenAI配置（可选） OPENAI_API_KEY=your_openai_api_key_here # 应用配置 DEBUG=false LOG_LEVEL=INFO MAX_WORKERS=4 # 文件存储 UPLOAD_DIR=/app/storage/uploads MAX_FILE_SIZE=10485760

6.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(host='localhost', port=6379, db=0) def get_cached_category(self, text_hash: str) -> Optional[str]: cached = self.redis_client.get(f"category:{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(f"category:{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_sim+eng', # 统一文本块 '--psm 4 -l chi_sim+eng', # 假设有一列文本 '--psm 8 -l chi_sim+eng', # 单个单词 ] best_result = "" for config in configs: try: text = pytesseract.image_to_string(preprocessed_image, config=config) if len(text) > len(best_result): best_result = text except: continue return best_result

7.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: