Skip to content

About

Bank transaction OCR service with metadata and transaction extraction

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository
BranchesTags

Repository files navigation

银行交易流水 OCR 智能体服务

基于 LangGraph + LangChain + FastAPI 构建的银行流水信息提取服务，通过多模态视觉大模型（阿里云百炼 Qwen-VL）从图片/PDF 中自动识别、结构化银行交易流水，并输出标准 CSV。

目录

整体架构

上游系统 (POST /ocr)
       │
       ▼
┌─────────────────────────────────────────────────────┐
│                  FastAPI 服务层                       │
│  ┌──────────────────────────────────────────────┐    │
│  │  POST /ocr          POST /ocr/upload          │    │
│  │  (远程URL方式)       (本地文件上传方式)         │    │
│  └──────────────────────────────────────────────┘    │
└────────────────────────┬────────────────────────────┘
                         │
                         ▼
              ┌──────────────────────┐
              │   URL 修正 & 分类     │
              │  (fix_url / detect)  │
              └──────┬──────────────┘
           ┌─────────┴──────────┐
         图片                  PDF
           │                    │
           │              pdf_to_images()
           │              (PyMuPDF)
           │                    │ (每页转 JPEG)
           └─────────┬──────────┘
                     │ asyncio.gather (并发)
                     ▼
        ┌────────────────────────────────────┐
        │        LangGraph OCR Agent         │
        │                                    │
        │  START                             │
        │   ├─► [extract_transactions]  ─┐  │
        │   └─► [extract_metadata]      ─┤  │
        │          (两个 VLM 节点并行)    │  │
        │                                ▼  │
        │                      [merge_results]  │
        │                                │  │
        │                              END  │
        └────────────────────────────────────┘
                     │
                     ▼
        ┌────────────────────────────────────┐
        │         数据处理层                  │
        │  Markdown → DataFrame              │
        │  → transaction_mapper 标准化       │
        │  → 多文件 DataFrame 合并           │
        └────────────────────────────────────┘
                     │
                     ▼
            返回 base64(CSV) + 统计信息

代码逻辑详解

1. 接口入口 — app/routers/ocr.py

POST /ocr（生产接口）

接收请求体 OCRRequest
  └─ extract_remote_urls()   从 transactions[].files[].remote_url 提取所有 URL
  └─ fix_url()               不以 / 或 http/data: 开头的 URL 补充 /
  └─ asyncio.gather()        并发处理所有 URL（受 MAX_WORKERS 信号量控制）
       └─ _process_single_url()
            ├─ detect_file_type()   通过 URL 后缀 或 data: MIME 判断图片/PDF
            ├─ [图片] → _process_single_image() → Agent → DataFrame
            └─ [PDF]  → download_file() / base64 解码
                      → pdf_to_images()        PDF 拆分为 JPEG 页数组
                      → asyncio.gather()       各页并发调 Agent
                      → 每页 _process_single_image() → DataFrame
  └─ merge_dataframes()      所有 DataFrame 合并为一份
  └─ 返回 base64(CSV)

POST /ocr/upload（本地测试接口）

与 /ocr 逻辑完全相同，区别在于：

  • 跳过远程 URL 下载：直接接收 multipart/form-data 文件字节
  • 根据文件名后缀自动判断图片/PDF 并走相同处理链路
  • 适用于本地验收测试、联调时无需配置影像平台

2. LangGraph Agent — app/services/ocr_agent.py

参考 账单提取核心.yml Dify DSL 的工作流结构，用 LangGraph 原生实现：

AgentState
  image_url        → 输入：图片 URL 或 data:image base64
  transactions_md  → LLM2 输出：Markdown 表格（交易流水正文）
  metadata_json    → LLM3 输出：JSON（账户元数据）
  merged_result    → 合并节点输出：最终 JSON 字符串
  errors           → Annotated[list[str], reducer] 收集并行节点错误

START
  ├──► extract_transactions (LLM2)
  │      System Prompt: 提取全部交易记录 → 输出 Markdown 表格
  │      Model: qwen3-vl-flash (temperature=0.3)
  │
  └──► extract_metadata (LLM3)  ← 与 LLM2 并行执行
         System Prompt: 仅提取表头客户名称/账号/开户行 → 输出 JSON
         Model: qwen3-vl-flash (temperature=0.2)
         ↓
merge_results
  将 Markdown 文本 + JSON 合并为统一结构:
  { "客户名称": ..., "客户账号": ..., "账户所属机构": ...,
    "transactions": "| 列1 | 列2 |..." }
  ↓
END

两个 LLM 节点并行 fan-out，merge_results fan-in 汇聚，利用 LangGraph 的 Annotated reducer 解决并发写入冲突。

3. 数据处理 — app/services/data_processor.py + transaction_mapper.py

Agent JSON
  └─ transactions_md (Markdown 表格文本)
       └─ markdown_to_dataframe()     Markdown → HTML → pd.read_html() → DataFrame
            └─ process_transaction_data()    (完全复用 transaction_mapper.py)
                 ├─ map_fields()             列名模糊匹配 → 标准字段名
                 ├─ clean_dirty_data()       清洗分隔行、全 "-" 行、无效行
                 ├─ split_datetime_fields()  datetime 拆分为 date + time
                 ├─ parse_amount()           金额清洗（去符号、处理负数）
                 ├─ parse_debit_credit()     借贷标志标准化 → 借/贷
                 └─ 填充 account_name/id/bank/serial_id/行号/时间戳
  └─ DataFrames → merge_dataframes()
       ├─ pd.concat()                  合并多文件
       ├─ 重置 rel_line_num            连续行号
       └─ 刷新 create_time/update_time
  └─ dataframe_to_csv_string()        → UTF-8-SIG CSV 字符串 → base64

4. 文件处理 — app/services/file_handler.py

函数 说明
detect_file_type(url) 通过 URL 后缀或 data: MIME 判断 image/pdf
download_file(url) 异步 HTTP 下载远程文件（httpx）
pdf_to_images(bytes) PyMuPDF 渲染 → PIL 压缩 → JPEG 字节列表
image_to_base64_url(bytes) 转 data:image/jpeg;base64,...

5. 日志 — app/logger.py

TimedRotatingFileHandler
  when="midnight"          每天 0 点切割
  backupCount=90           保留 90 天（3 个月）
  suffix="%Y-%m-%d"        文件名格式: app.log.2026-03-04
  编码 = utf-8

日志文件位于 logs/ 目录，同时输出到控制台（INFO 级）。

项目结构

transactionOCR/
├── app/
│   ├── main.py               # FastAPI 应用 & 路由注册
│   ├── config.py             # 环境变量配置（dotenv）
│   ├── models.py             # Pydantic 请求/响应模型
│   ├── logger.py             # 日志配置（3 个月轮转）
│   ├── routers/
│   │   └── ocr.py            # POST /ocr  POST /ocr/upload
│   ├── services/
│   │   ├── url_fixer.py      # URL 修正 & remote_url 提取
│   │   ├── file_handler.py   # 文件下载、类型检测、PDF→图片
│   │   ├── ocr_agent.py      # LangGraph Agent（双 VLM 并行）
│   │   └── data_processor.py # Markdown→DataFrame→CSV、合并
│   └── utils/
│       └── markdown_parser.py # Markdown 表格 → DataFrame
├── tests/
│   └── test_ocr.py           # 单元测试 + 集成测试 + 本地上传测试
├── transaction_mapper.py     # 字段映射核心（复用原有逻辑）
├── requirements.txt
├── .env                      # 本地配置（不提交 Git）
└── logs/                     # 运行时生成

接口说明

POST /ocr — 生产接口（远程 URL、固定回调地址）

请求体：

{
  "orderNO": "BIZ-20260101-001",
  "remoteUrls": ["https://image-server.com/path/to/bill.pdf"],
  "custName": "张三",
  "custNo": "CUST-001",
  "companyName": "某某有限公司",
  "contractNo": "CONTRACT-001",
  "accountNo": "6228480190000000000",
  "accountName": "张三"
}

该接口完成后回调环境变量 CALLBACK_URL 指定的地址。

立即响应（202）：

{
  "orderNO": "BIZ-20260101-001",
  "message": "已受理，处理完成后将回调通知"
}

POST /ocr/callback — 生产接口（请求指定回调地址）

请求字段与 /ocr 一致，并额外传入必填的 callback_url：

{
  "orderNO": "BIZ-20260101-001",
  "remoteUrls": ["https://image-server.com/path/to/bill.pdf"],
  "callback_url": "https://your-domain.com/webhook/result"
}

该接口同样立即返回 202，但处理结果会回调 callback_url，而不是使用固定的 CALLBACK_URL。

PDF 输出的 page_no 使用 PDF 物理页序覆盖模型值；图片使用各自页面提取的页码，未识别则为空。多张图片分别提取元数据，禁止跨图片传播账户信息。

CSV 还会返回账期、合计、本页余额、期末余额、打印次数、网点、印章企业名称、cust_no、contract_no、file_url 和 file_type 等字段。其中 file_url 对应原始 remoteUrls 元素，file_type 用于区分 pdf/jpg/png 等源文件类型。

POST /ocr/sync — 同步接口

请求体与 /ocr 完全一致。接口等待识别完成后直接返回 OCRResponse，不发送回调；处理过程与异步接口共用相同的任务容量、执行信号量、分块并发和重试机制。

curl -X POST http://localhost:8000/ocr/sync \
  -H "Content-Type: application/json" \
  -d '{"orderNO":"SYNC-001","remoteUrls":["https://image-server.com/bill.pdf"],"custNo":"CUST-001","contractNo":"CONTRACT-001"}'

POST /ocr/upload — 本地测试接口（multipart 文件上传）

本地调试、跳过远程 URL 下载，直接上传文件。

curl -X POST http://localhost:8000/ocr/upload \
  -F "files=@WechatIMG127.jpg;type=image/jpeg" \
  -F "files=@上海银行交易流水20260128-1.pdf;type=application/pdf" \
  -F "order_no=LOCAL_TEST_001" \
  -F "custNo=CUST-001" \
  -F "contractNo=CONTRACT-001"

参数与 /ocr 响应结构完全一致。

其他接口

方法 路径 说明
GET / 服务信息
GET /health 健康检查 {"status":"ok"}
GET /docs Swagger UI（FastAPI 自动生成）
GET /redoc ReDoc 文档

配置说明

所有配置通过 .env 文件或环境变量注入：

变量 默认值 说明
LLM_API_KEY sk-... 阿里云百炼 API Key
LLM_BASE_URL https://dashscope.aliyuncs.com/compatible-mode/v1 OpenAI 兼容端点
LLM_EXTRACT_MODEL qwen3.5-35b-a3b 流水提取模型
LLM_META_MODEL qwen3.5-35b-a3b 元数据提取模型
REQUEST_TIMEOUT 1200 LLM 请求超时（秒）
MAX_WORKERS 2 PDF 文件级并发上限
CHUNK_SIZE 5 每个 chunk 的图片数
MAX_CHUNK_WORKERS 12 单请求 chunk 并发上限
GLOBAL_SEMAPHORE_SIZE 8 全局 chunk 并发闸门（跨请求）
BG_TASK_MAX_CONCURRENCY 2 后台任务执行并发上限
BG_TASK_QUEUE_LIMIT 20 后台任务队列容量上限
TO_THREAD_MAX_WORKERS 8 asyncio.to_thread 默认线程池大小
LOG_LEVEL INFO 全局日志级别
RAW_HTTP_LOG_ENABLED false 是否记录原始请求/响应报文
PDF_QUALITY 85 JPEG 压缩质量（1-100）
PDF_ZOOM 2.78 PDF 渲染缩放倍率
LOG_DIR logs 日志目录
LOG_RETENTION_DAYS 90 日志保留天数（3 个月）
DOWNLOAD_MAX_KEEPALIVE_CONNECTIONS 20 下载连接池 keepalive 上限
DOWNLOAD_MAX_CONNECTIONS 100 下载连接池总连接上限
PORT 8000 服务端口

生产建议值（Docker）

参数 建议值 备注
WORKERS 2 先保守，避免多进程放大竞争
MAX_CHUNK_WORKERS 4~8 视 LLM 吞吐与 CPU 调整
GLOBAL_SEMAPHORE_SIZE 4~8 控制全局并发洪峰
BG_TASK_QUEUE_LIMIT 20~50 过载时快速拒绝避免雪崩
TO_THREAD_MAX_WORKERS 4~12 建议与容器 CPU 核数联动
LOG_LEVEL INFO 排障时短时切到 DEBUG
RAW_HTTP_LOG_ENABLED false 仅短时排障开启

快速启动

1. 安装依赖

pip install -r requirements.txt

2. 配置 API Key

复制 .env.example 为 .env，并填入你自己的 API Key：

LLM_API_KEY=your_api_key_here

3. 启动服务

python -m app.main
# 服务在 http://0.0.0.0:8000 启动

或使用 uvicorn 多 worker 模式（生产环境）：

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

4. 验证服务

curl http://localhost:8000/health
# {"status":"ok","service":"transaction-ocr-agent"}

测试指南

单元测试（无需 LLM，秒级完成）

python -m pytest tests/test_ocr.py -v -k "not Integration and not LocalUpload"

覆盖：健康检查、请求模型验证、URL 修正、文件类型检测、Markdown 解析、PDF 转图片、数据处理。

本地文件全流程测试（推荐，绕过远程 URL）

使用当前目录下的 WechatIMG127.jpg、WechatIMG128.jpg、WechatIMG129.png、上海银行交易流水20260128-1.pdf 直接调用 /ocr/upload。

# 运行全部本地上传测试
RUN_INTEGRATION=1 python -m pytest tests/test_ocr.py::TestLocalUpload -v -s

# 只测单张图片
RUN_INTEGRATION=1 python -m pytest tests/test_ocr.py::TestLocalUpload::test_upload_single_image -v -s

# 只测 PDF
RUN_INTEGRATION=1 python -m pytest tests/test_ocr.py::TestLocalUpload::test_upload_single_pdf -v -s

# 图片 + PDF 多文件并发
RUN_INTEGRATION=1 python -m pytest tests/test_ocr.py::TestLocalUpload::test_upload_multiple_files -v -s

也可以直接用 curl 测试：

# 单张图片
curl -X POST http://localhost:8000/ocr/upload \
  -F "files=@WechatIMG127.jpg" \
  -F "order_no=TEST001" | python3 -c "
import sys,json,base64
d=json.load(sys.stdin)
print('records:', d['total_records'])
if d.get('csv_data'):
    print(base64.b64decode(d['csv_data']).decode('utf-8-sig')[:500])
"

# PDF
curl -X POST http://localhost:8000/ocr/upload \
  -F "files=@上海银行交易流水20260128-1.pdf" \
  -F "order_no=TEST_PDF"

生产接口集成测试（远程 URL 方式）

RUN_INTEGRATION=1 python -m pytest tests/test_ocr.py::TestOCRAgentIntegration -v -s

测试结果汇总（已验证）

测试类 测试数 结果 说明
TestHealthCheck 2 ✅ 基础接口
TestRequestValidation 2 ✅ 参数校验
TestURLFixer 3 ✅ URL 修正逻辑
TestFileTypeDetection 4 ✅ 文件类型检测
TestMarkdownParser 3 ✅ Markdown 表格解析
TestPDFConversion 1 ✅ PDF→JPEG 转换
TestDataProcessor 2 ✅ 字段映射 & 合并
TestLocalUpload 4 ✅ 本地文件全流程（需 LLM）
TestOCRAgentIntegration 4 ✅ 远程 URL 全流程（需 LLM）
合计 25 ✅ 全通过

日志说明

服务运行后日志保存在 logs/ 目录：

logs/
├── app.log              # 当天日志（实时追加）
├── app.log.2026-03-04   # 历史日志（按天存档）
└── app.log.2026-03-03

# 实时查看日志
tail -f logs/app.log

# 搜索特定订单
grep "order_no=BIZ-001" logs/app.log

# 统计成功/失败
grep -c "处理完成" logs/app.log
grep -c "处理失败" logs/app.log

日志默认保留 90 天（3 个月），可通过 LOG_RETENTION_DAYS 调整。

About

Bank transaction OCR service with metadata and transaction extraction

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages