基于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 本质上是一个微服务架构的应用，主要包含三个模块：

1. 前端上传接口：提供一个稳定的HTTP链接，接收多文件上传
2. 文件预处理管道：对上传的图片或PDF进行格式转换、文字提取预处理
3. LLM分类引擎：调用大语言模型分析票据内容，按规则分类

2.2 LLM在票据识别中的特殊价值

你可能会问：OCR（光学字符识别）技术已经很成熟，为什么还需要LLM？关键在于语义理解与上下文判断。

举例说明：

• 一张餐饮小票上既有食物项也有服务费，OCR只能提取文字，LLM能判断这属于"业务招待"还是"团队聚餐"
• 一张模糊的出租车票，OCR可能识别失败，LLM可以根据残留的"出租"字样和金额范围推断类别
• 跨语言票据（如英文发票中的"Tax"和中文发票的"税"），LLM能统一归类到"税费"科目

这种理解能力来自于LLM在大量文本数据上的预训练，让它能够处理OCR输出中的噪声、不完整信息和多义性。

2.3 关键术语解释

3. 环境准备与前置条件

在开始部署前，请确保你的环境满足以下要求：

3.1 系统与环境要求

• 操作系统：Linux (Ubuntu 20.04+ 推荐) 或 macOS，Windows可通过WSL运行
• Python版本：3.8-3.11（3.9+推荐，避免使用已停止支持的版本）
• 内存：至少4GB可用内存（如果本地运行LLM需要更多）
• 网络：能正常访问PyPI和可能的LLM API服务

3.2 核心依赖说明

Sorted Receipts 依赖几个关键库，各自的作用如下：

# 核心依赖功能说明 - fastapi：提供上传接口和Web界面 - pydantic：数据验证和设置管理 - python-multipart：处理文件上传 - pillow：图像预处理（尺寸调整、格式转换） - pytesseract或easyocr：OCR文字提取 - openai或llama-cpp-python：LLM集成

重要提醒：如果你计划使用云端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.txt

4.2 第二步：配置文件设置

项目使用环境变量管理配置，创建.env文件：

# 创建配置文件 cp .env.example .env

编辑.env文件，根据你的需求调整以下关键配置：

# LLM服务配置（以OpenAI为例） LLM_PROVIDER=openai OPENAI_API_KEY=your_api_key_here LLM_MODEL=gpt-3.5-turbo # 应用基础配置 UPLOAD_FOLDER=./uploads MAX_FILE_SIZE=10485760 # 10MB ALLOWED_EXTENSIONS=pdf,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_code=400, 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_extensions

5.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_ok=True) 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', encoding='utf-8') as f: json.dump({ "batch_id": batch_id, "processed_at": datetime.now().isoformat(), "results": results }, f, ensure_ascii=False, indent=2) 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='', encoding='utf-8') as f: if results: fieldnames = results[0].keys() writer = csv.DictWriter(f, fieldnames=fieldnames) 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 "files=@taxi_receipt.jpg" \ -F "files=@restaurant_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. 常见问题与排查思路

在实际部署中，你可能会遇到以下典型问题：

7.1 性能优化建议

对于生产环境使用，建议进行以下优化：

# 文件处理并发控制 from concurrent.futures import ThreadPoolExecutor import asyncio class ProcessingPool: def __init__(self, max_workers: int = 3): self.executor = ThreadPoolExecutor(max_workers=max_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_exceptions=True) return results

8. 最佳实践与工程建议

8.1 分类规则设计原则

基于实际项目经验，分类规则的设计应该遵循：

1. 互斥性优先：类别之间界限清晰，避免重叠
2. 容错性设计：为模糊票据设置"待确认"类别
3. 可扩展结构：预留自定义字段，适应不同财务需求

{ "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 生产环境部署建议

对于企业级使用，建议采用以下架构：

1. 使用Docker容器化部署，保证环境一致性
2. 配置反向代理（Nginx）处理静态文件和SSL
3. 使用Redis作为任务队列和缓存
4. 设置监控告警，监控服务健康状态
5. 定期备份分类规则和处理结果

# 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, json=results, 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）服务于业务目标——在这里是降低报销流程的摩擦成本。在实际应用中，建议先从一个小团队开始试点，收集反馈，迭代优化分类规则，再逐步推广到更大范围。