refactor: 架构重组 — doc/bot → core/infra,新增 Agent 调度模块

- src/doc/ 拆分为 src/core/extraction/, matching/, validation/(核心业务逻辑)
- src/bot/ 重命名为 src/infra/browser/(浏览器自动化基础设施)
- fill_consumable_doc.py → src/infra/documents/consumable.py
- 新增 Agent 调度模块:coordinator.py, events.py, session.py,重构 orchestrator.py
- 更新 AGENTS.md、README.md 及所有子目录 README
This commit is contained in:
wandering
2026-07-02 18:36:19 +08:00
parent e9e0252638
commit 9043886b6f
68 changed files with 2965 additions and 1548 deletions
+21
View File
@@ -0,0 +1,21 @@
---
last_reviewed: 2026-06-15
---
# src/core — 核心业务逻辑
项目的核心业务层,负责信息提取、金额匹配和信息校验。此层不依赖 Web 框架或浏览器自动化等基础设施。
## 子模块
| 目录 | 说明 |
|------|------|
| `extraction/` | 文档信息提取:PDF/图片 → LLM 多模态识别 → 结构化数据 |
| `matching/` | 发票与支付记录按金额匹配(一对一 / 一对多) |
| `validation/` | 声明式信息完整性校验,规则从 JSON 配置文件加载 |
## 设计原则
- **零外部依赖**:不依赖 Flask、Playwright 等框架
- **接口契约**:每个子模块通过 `__init__.py` 导出稳定的对外接口
- **错误传播**:明确的异常层次,便于上层统一处理
+4
View File
@@ -0,0 +1,4 @@
"""核心业务逻辑模块
提供信息提取、规则校验和发票匹配功能。
"""
+29
View File
@@ -0,0 +1,29 @@
---
last_reviewed: 2026-06-15
---
# src/core/extraction — 信息提取
从 PDF 发票和图片中提取结构化数据,是系统数据流的起点。
## 文件
| 文件 | 职责 |
|------|------|
| `extractor.py` | 编排入口:扫描目录 → 逐文件提取 → 分类(发票/支付记录/申请单)→ 金额匹配 |
| `llm_extractor.py` | LLM 多模态提取核心:统一文档提取、差旅/普通信息提取、缓存管理、SSE 流式事件 |
## 对外接口
| 函数 | 说明 |
|------|------|
| `extract_invoices(directory)` | 统一提取入口,返回 `(payment_records, applications, groups)` |
| `extract_document(file_path)` | 从单个图片/PDF 提取信息 |
| `extract_travel_info(source_dir)` | 综合发票和匹配结果提取差旅信息 |
| `extract_normal_info(source_dir)` | 提取普通发票报销信息 |
| `load_cache(source_dir)` | 加载缓存的结构化数据 |
| `llm_query_text(...)` | 纯文本 LLM 查询(供 Agent 调度使用) |
## 缓存机制
提取结果缓存在 `.invoice_cache/` 目录中,文件名与源文件同名(`发票1.pdf` → `.invoice_cache/发票1.json`),避免重复调用 LLM。
+42
View File
@@ -0,0 +1,42 @@
"""信息提取模块
提供发票/文档结构化提取、LLM 辅助提取等功能。
"""
from .extractor import (
EXTRACTION_PARALLEL_COUNT,
FILE_EVENTS_LOG,
SUPPORTED_EXTENSIONS,
extract_invoices,
)
from .llm_extractor import (
CACHE_DIR_NAME,
build_extraction_user_message,
extract_document,
extract_normal_info,
extract_travel_info,
llm_query_text,
load_cache,
load_match_result,
merge_supplement_into_info,
parse_json_response,
process_user_supplement,
)
__all__ = [
"CACHE_DIR_NAME",
"build_extraction_user_message",
"extract_document",
"extract_normal_info",
"extract_travel_info",
"load_cache",
"load_match_result",
"llm_query_text",
"merge_supplement_into_info",
"parse_json_response",
"process_user_supplement",
"EXTRACTION_PARALLEL_COUNT",
"FILE_EVENTS_LOG",
"SUPPORTED_EXTENSIONS",
"extract_invoices",
]
+358
View File
@@ -0,0 +1,358 @@
"""发票提取编排
统一扫描目录下所有文件(PDF + 图片),通过 LLM 提取结构化数据,
根据 LLM 返回的「invoice_type」字段自动分类为发票/支付记录/出差事前申请单。
对外接口:
extract_invoices(directory) -> tuple[list[dict], list[dict], dict]
错误传播规则:
- 单个文件提取失败: 记录日志 + SSE error 事件,继续处理下一个文件
- 全部文件提取失败: raise ExtractionError,携带失败文件列表和原始错误
- 缓存读取失败: 静默降级,尝试重新提取
SSE 文件进度事件:
在 source_dir 下写入 file_events.log,每行一个 JSON 对象:
- {"type": "file_progress", "file": "...", "status": "processing"}
- {"type": "file_progress", "file": "...", "status": "done", "summary": {...}}
- {"type": "file_progress", "file": "...", "status": "cached"}
- {"type": "file_progress", "file": "...", "status": "error", "error": "..."}
"""
import json
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any
from ... import get_logger
from ...core.matching import match_invoices_to_cards
from ...exceptions import ExtractionError
from ...infra.documents.invoice import CACHE_DIR_NAME, classify_invoice_batch
from .llm_extractor import extract_document
log = get_logger("extractor")
# SSE 文件进度事件日志文件名
FILE_EVENTS_LOG = "file_events.log"
# 并行提取文件数,可通过环境变量 EXTRACTION_PARALLEL_COUNT 配置
EXTRACTION_PARALLEL_COUNT = int(os.environ.get("EXTRACTION_PARALLEL_COUNT", "3"))
# 支持的文件扩展名
SUPPORTED_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".bmp", ".webp"}
def _emit_file_event(source_dir: Path, file_name: str, status: str, **kwargs: Any) -> None:
"""向 file_events.log 追加一行 JSON 事件(线程安全,失败时静默忽略)"""
event = {
"type": "file_progress",
"file": file_name,
"status": status,
**kwargs,
}
try:
event_path = source_dir / FILE_EVENTS_LOG
with open(event_path, "a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
except Exception:
pass
def _get_cache_dir(source_dir: Path) -> Path:
"""获取缓存目录路径"""
cache_dir = source_dir / CACHE_DIR_NAME
cache_dir.mkdir(exist_ok=True)
return cache_dir
def _get_json_path(file_path: Path, cache_dir: Path) -> Path:
"""根据文件路径生成对应的 JSON 缓存路径(包含后缀名以区分同名的 PDF/图片)"""
return cache_dir / f"{file_path.stem}{file_path.suffix}.json"
def _save_to_cache(file_path: Path, extracted_data: dict[str, Any], cache_dir: Path) -> Path:
"""将提取结果保存到 JSON 缓存文件,并记录源文件路径和后缀名"""
json_path = _get_json_path(file_path, cache_dir)
cache_data = {
"source_file": str(file_path),
"source_filename": file_path.name,
"source_extension": file_path.suffix.lower(),
"extracted_data": extracted_data,
}
with open(json_path, "w", encoding="utf-8") as f:
json.dump(cache_data, f, ensure_ascii=False, indent=2)
log.info(f"提取结果已缓存: {json_path.name}")
return json_path
def _load_from_cache(json_path: Path, expected_extension: str | None = None) -> dict[str, Any] | None:
"""从 JSON 缓存文件加载提取结果,可选校验后缀名一致性"""
if not json_path.exists():
return None
try:
with open(json_path, encoding="utf-8") as f:
cache_data: dict[str, Any] = json.load(f)
# 校验后缀名是否一致,防止同名不同后缀的文件误命中缓存
if expected_extension and cache_data.get("source_extension", "").lower() != expected_extension.lower():
return None
result: dict[str, Any] | None = cache_data.get("extracted_data")
return result
except Exception as e:
log.warning(f"读取缓存失败 {json_path.name}: {e}")
return None
def _build_summary(data: dict[str, Any]) -> dict[str, str]:
"""从提取结果构建前端展示摘要(人类可读格式)。
返回所有非内部字段(排除 _source_file 等下划线前缀字段),
并将 invoice_type 转换为中文标签。列表/字典类型的值会展开为可读文本。
"""
invoice_type_map = {
"train": "高铁票",
"hotel": "酒店住宿",
"general": "普通发票",
"payment": "支付记录",
"application": "出差申请单",
}
summary = {}
for key, value in data.items():
# 跳过内部字段
if key.startswith("_"):
continue
# 跳过空值
if value is None or value == "":
continue
# invoice_type 转为中文标签
if key == "invoice_type":
summary["invoice_type_label"] = invoice_type_map.get(str(value), str(value))
elif isinstance(value, list):
# 列表展开为多行可读文本
if len(value) == 0:
continue
parts = []
for item in value:
if isinstance(item, dict):
# 字典项用 "key: value" 格式拼接
pair_parts = [f"{k}: {v}" for k, v in item.items()]
parts.append(" | ".join(pair_parts))
else:
parts.append(str(item))
summary[key] = "\n".join(parts)
elif isinstance(value, dict):
# 字典展开为 "key: value" 格式
pair_parts = [f"{k}: {v}" for k, v in value.items()]
summary[key] = " | ".join(pair_parts)
else:
summary[key] = str(value)
return summary
def _extract_document(
file_path: Path,
cache_dir: Path,
source_dir: Path,
) -> tuple[dict[str, str] | None, str | None]:
"""提取单个文件的结构化信息,优先使用缓存。
根据 LLM 返回的「invoice_type」字段自动分类:
- "payment" -> 支付记录
- "application" -> 申请单
- 有 "invoice_number" -> 发票
- 其他 -> 无法识别
Args:
file_path: 文件路径(PDF 或图片)。
cache_dir: JSON 缓存目录。
source_dir: 源目录(用于写入 SSE 进度事件)。
Returns:
(提取结果字典或 None, 错误信息或 None)。
"""
file_name = file_path.name
json_path = _get_json_path(file_path, cache_dir)
cached = _load_from_cache(json_path, expected_extension=file_path.suffix.lower())
if cached:
cached["_source_file"] = file_name
log.info(f"使用缓存: {file_name}")
_emit_file_event(source_dir, file_name, "cached")
return cached, None
# 发送处理中事件
_emit_file_event(source_dir, file_name, "processing")
log.info(f"使用多模态提取: {file_name}")
try:
result = extract_document(file_path)
if result:
result["_source_file"] = file_name
_save_to_cache(file_path, result, cache_dir)
# 发送完成事件(含摘要)
summary = _build_summary(result)
_emit_file_event(source_dir, file_name, "done", summary=summary)
return result, None
except Exception as e:
err_msg = str(e)
log.warning(f"多模态提取失败: {file_name} ({err_msg})")
_emit_file_event(source_dir, file_name, "error", error=err_msg)
return None, err_msg
return None, None
def _find_all_files(directory: str) -> list[Path]:
"""扫描目录下所有支持的文件(PDF + 图片)"""
dir_path = Path(directory)
files = [f for f in dir_path.iterdir() if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS]
return sorted(files)
def _save_match_result(payment_records: list[dict[str, Any]], cache_dir: Path) -> None:
"""将发票与支付记录的匹配结果保存到缓存。"""
match_data: dict[str, list[dict[str, Any]]] = {}
for record in payment_records:
card_source = record.get("_source_file", "")
matched = record.get("_matched_invoices", [])
card_amount = record.get("card_amount", "")
if matched:
invoice_list = [
{
"file": inv.get("_source_file", ""),
"type": inv.get("invoice_type", ""),
"amount": inv.get("total_amount", inv.get("total_amount", "")),
}
for inv in matched
]
if card_source:
match_data[f"{card_source} (¥{card_amount})"] = invoice_list
else:
key = "__unmatched__"
if key not in match_data:
match_data[key] = []
match_data[key].extend(invoice_list)
if not match_data:
return
match_path = cache_dir / "match_result.json"
with open(match_path, "w", encoding="utf-8") as f:
json.dump(match_data, f, ensure_ascii=False, indent=2)
log.info(f"匹配结果已缓存: {match_path.name}")
def extract_invoices(
directory: str = ".",
) -> tuple[list[dict[str, str]], list[dict[str, str]], dict[str, list[dict[str, str]]]]:
"""扫描目录下所有文件,提取信息并匹配支付记录
统一使用 LLM 提取,根据返回的「invoice_type」自动分类:
- 发票(有 invoice_number)-> 参与金额匹配
- 支付记录(invoice_type="payment")-> 参与金额匹配
- 出差事前申请单 -> 单独存储,不参与匹配
Returns:
(payment_records, applications, groups):
- payment_records: 支付记录列表(仅包含真实发票,不含申请单)
- applications: 出差事前申请单列表(单独存储,不参与支付匹配)
- groups: 按文档类型分组的字典
{'travel': [差旅发票], 'general': [普通发票], 'application': [出差事前申请单]}
Raises:
ExtractionError: 当所有文件提取均失败时抛出,携带失败文件列表和原始错误。
"""
source_dir = Path(directory)
cache_dir = _get_cache_dir(source_dir)
# 清空上次的文件进度事件
try:
(source_dir / FILE_EVENTS_LOG).unlink(missing_ok=True)
except Exception:
pass
all_files = _find_all_files(directory)
if not all_files:
log.warning("未找到支持的文件")
return [], [], {"travel": [], "general": [], "application": []}
log.info(f"发现 {len(all_files)} 个文件")
all_invoices = []
all_cards = []
applications = []
# 记录失败文件及其错误信息
failed_files: list[tuple[str, str]] = []
max_workers = max(1, EXTRACTION_PARALLEL_COUNT)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_file = {executor.submit(_extract_document, fp, cache_dir, source_dir): fp for fp in all_files}
for future in as_completed(future_to_file):
file_path = future_to_file[future]
try:
result, err = future.result()
except Exception as e:
err_msg = str(e)
failed_files.append((file_path.name, err_msg))
log.warning(f"未能解析: {file_path.name} ({err_msg})")
continue
if not result:
if err:
failed_files.append((file_path.name, err))
log.warning(f"未能解析: {file_path.name}")
continue
inv_type = result.get("invoice_type", "")
if inv_type == "application":
applications.append(result)
log.info(f"[{inv_type}] 已解析: {file_path.name}")
elif inv_type == "payment":
all_cards.append(result)
log.info(f"[{inv_type}] 已解析: {file_path.name}")
elif result.get("invoice_number"):
all_invoices.append(result)
log.info(f"[{inv_type}] 已解析: {file_path.name}")
else:
log.warning(f"无法分类: {file_path.name} (invoice_type={inv_type})")
# 全部文件提取失败时抛出异常,携带原始错误信息
if failed_files and not all_invoices and not all_cards and not applications:
failed_names = [name for name, _ in failed_files]
error_details = {name: err for name, err in failed_files}
raise ExtractionError(
f"所有 {len(all_files)} 个文件提取均失败",
failed_files=failed_names,
details=error_details,
)
log.info(f"分类结果: 发票 {len(all_invoices)} 张, 支付记录 {len(all_cards)} 条, 申请单 {len(applications)} 份")
if not all_invoices:
log.warning("未成功解析任何发票")
payment_records = match_invoices_to_cards(all_invoices, all_cards)
_save_match_result(payment_records, cache_dir)
# 构建分类
all_documents: list[dict[str, str]] = []
for record in payment_records:
all_documents.extend(record.get("_matched_invoices", []))
all_documents.extend(applications)
groups = classify_invoice_batch(all_documents)
log.info(
f"文档分类: 差旅发票 {len(groups['travel'])} 张, "
f"普通发票 {len(groups['general'])} 张, "
f"出差事前申请单 {len(groups['application'])} 份"
)
return payment_records, applications, groups
+591
View File
@@ -0,0 +1,591 @@
"""LLM 信息提取
使用 LLM 从 PDF 文本/图片、支付截图中提取结构化数据。
## 功能模块
- **统一文档提取**:使用一套提示词,LLM 自行判断文档类型(发票/支付记录/出差事前申请单等),支持 JSON 格式输出。
- **差旅信息提取**:综合多张发票、支付记录和匹配结果,提取出差事由、地点、时间等差旅相关信息。
- **缓存管理**:支持从 `.invoice_cache/` 目录加载已提取的结构化数据和匹配结果,避免重复处理。
- **SSE 流式事件**:`extract_travel_info` 和 `extract_normal_info` 在调用 LLM 时,向 `source_dir/llm_stream.log` 写入流式事件(start/reasoning/chunk/end/error),前端通过 SSE 实时展示 AI 思考过程与正式回答。
## 对外接口
- `extract_document(file_path) -> dict` — 统一入口:从任意图片/PDF 提取信息
- `extract_travel_info(source_dir) -> dict` — 综合发票和匹配结果提取差旅信息
- `extract_normal_info(source_dir) -> dict` — 提取普通发票报销信息
- `load_cache(source_dir) -> dict` — 加载缓存的结构化数据
- `load_match_result(source_dir) -> dict` — 加载发票与支付记录的匹配结果
- `llm_query_text(system_prompt, text, source_dir) -> str` — 纯文本 LLM 查询(供 Agent 调度使用)
- `parse_json_response(text) -> dict` — 从 LLM 响应中提取 JSON(供 Agent 调度使用)
- `build_extraction_user_message(cache_map, match_result) -> str` — 构建提取请求的用户消息(供 Agent 调度使用)
## SSE 流式事件协议
`llm_stream.log` 每行一个 JSON 对象:
- `{"type": "llm_stream", "phase": "start", "label": "..."}` — LLM 调用开始
- `{"type": "llm_stream", "phase": "reasoning", "text": "..."}` — 模型原生推理/思考片段(来自 `thinking_delta`)
- `{"type": "llm_stream", "phase": "chunk", "text": "..."}` — 流式文本片段(正式回答)
- `{"type": "llm_stream", "phase": "end", "label": "..."}` — LLM 调用结束
- `{"type": "llm_stream", "phase": "error", "error": "..."}` — LLM 调用失败
"""
from __future__ import annotations
import base64
import json
from pathlib import Path
from typing import Any, cast
from ... import get_logger
from ...infra.documents.invoice import CACHE_DIR_NAME
from ...infra.llm.prompt import (
build_invoice_system_prompt,
build_normal_info_system_prompt,
build_supplement_system_prompt,
build_travel_info_system_prompt,
)
log = get_logger("llm_extractor")
# Re-export CACHE_DIR_NAME for convenience
__all__ = [
"CACHE_DIR_NAME",
"extract_document",
"extract_travel_info",
"extract_normal_info",
"load_cache",
"load_match_result",
"llm_query_text",
"parse_json_response",
"build_extraction_user_message",
]
# SSE LLM 流式事件日志文件名
LLM_STREAM_LOG = "llm_stream.log"
def _emit_llm_stream(source_dir: Path, phase: str, **kwargs: Any) -> None:
"""向 llm_stream.log 追加一行 JSON 事件(线程安全,失败时静默忽略)
Args:
source_dir: 会话目录路径。
phase: 事件阶段 ("start" / "chunk" / "end" / "error")。
**kwargs: 额外字段 (text, label, error 等)。
"""
event = {"type": "llm_stream", "phase": phase, **kwargs}
try:
event_path = source_dir / LLM_STREAM_LOG
with open(event_path, "a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
except Exception:
pass
def _create_llm() -> Any:
"""根据配置文件创建 LLM 实例。"""
try:
from llama_index.llms.openai_like import OpenAILike
except ImportError:
log.error("缺少 llama-index-llms-openai-like,请执行: uv pip install llama-index-llms-openai-like")
raise
from ...config import get_llm_config
llm_config = get_llm_config()
return OpenAILike(
model=llm_config["model"],
api_base=llm_config["api_base"],
api_key=llm_config.get("api_key", "lm-studio"),
temperature=0.1,
max_tokens=65535,
request_timeout=600.0,
is_chat_model=True,
)
def parse_json_response(text: str) -> dict[str, Any]:
"""从 LLM 响应中提取 JSON,处理可能的 Markdown 包裹。
Args:
text: LLM 响应文本。
Returns:
解析后的字典。
"""
text = text.strip()
# 处理 ```json ... ``` 包裹
if "```" in text:
start = text.find("```") + 3
end = text.find("```", start)
if end > start:
text = text[start:end].strip()
# 去掉可能的前缀 (如 "json")
if text.lower().startswith("json"):
text = text[4:].strip()
return cast(dict[str, Any], json.loads(text))
# ------------------------------------------------------------------
# 统一文档提取(多模态,直接传图片给 LLM)
# ------------------------------------------------------------------
def _image_to_base64(image_path: Path) -> str:
"""将图片文件读取为 base64 字符串。"""
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def _stream_llm_response(
llm: Any,
messages: list[Any],
source_dir: Path | None,
reasoning_effort: str,
log_label: str = "LLM",
) -> str:
"""流式调用 LLM 并写入 SSE 事件(供 llm_query_text 和 _llm_query_multimodal 共用)。
Args:
llm: LLM 实例。
messages: 消息列表。
source_dir: 会话目录(可选,传入时启用 SSE 流式事件写入)。
reasoning_effort: 推理努力级别。
log_label: 日志标签(用于区分"纯文本"和"多模态")。
Returns:
LLM 响应文本。
"""
try:
if source_dir:
_emit_llm_stream(source_dir, "start", label="正在分析文件...")
parts = []
for resp in llm.stream_chat(
messages,
temperature=0.1,
extra_body={"reasoning_effort": reasoning_effort},
):
delta = resp.delta
if delta:
parts.append(delta)
if source_dir:
_emit_llm_stream(source_dir, "chunk", text=delta)
thinking = getattr(resp, "additional_kwargs", {}) or {}
thinking_delta = thinking.get("thinking_delta", "")
if thinking_delta and source_dir:
_emit_llm_stream(source_dir, "reasoning", text=thinking_delta)
text = "".join(parts)
log.info("%s请求完成,响应总长度: %d 字符", log_label, len(text))
if source_dir:
_emit_llm_stream(source_dir, "end", label="分析完成")
return text
except Exception as e:
log.error("%s请求失败: %s", log_label, e)
if source_dir:
_emit_llm_stream(source_dir, "error", error=str(e))
raise
def llm_query_text(
system_prompt: str,
text: str,
reasoning_effort: str = "none",
source_dir: Path | None = None,
) -> str:
"""发送纯文本请求到 LLM(供 Agent 调度使用)。
Args:
system_prompt: 系统提示词。
text: 用户文本。
reasoning_effort: 推理努力级别。
source_dir: 会话目录(可选,传入时启用 SSE 流式事件写入)。
Returns:
LLM 响应文本。
"""
from llama_index.core.base.llms.types import TextBlock
from llama_index.core.llms import ChatMessage
from ...config import get_llm_config
messages = [
ChatMessage(role="system", content=system_prompt),
ChatMessage(role="user", blocks=[TextBlock(text=text)]),
]
llm_config = get_llm_config()
llm = _create_llm()
log.info(
"开始请求 LLM (model=%s, base=%s)",
llm_config["model"],
llm_config["api_base"],
)
return _stream_llm_response(llm, messages, source_dir, reasoning_effort, log_label="LLM")
def extract_document(file_path: Path) -> dict[str, Any]:
"""统一文档提取入口:从任意图片/PDF 中提取结构化信息。
LLM 会根据统一提示词自行判断文档类型(发票/支付记录/出差事前申请单等)。
Args:
file_path: 文件路径(支持 PDF 和图片格式)。
Returns:
包含提取字段的字典。
"""
from ...infra.documents.pdf import render_pdf_to_images
system_prompt = build_invoice_system_prompt()
user_text = f"请分析以下财务文档并提取信息:\n\n文件名: {file_path.name}"
# PDF 先渲染为图片
suffix = file_path.suffix.lower()
if suffix == ".pdf":
image_b64s = render_pdf_to_images(file_path)
else:
image_b64s = [_image_to_base64(file_path)]
if not image_b64s:
log.warning(f"文件渲染为空: {file_path.name}")
return {}
try:
response = _llm_query_multimodal(system_prompt, user_text, image_b64s)
result = parse_json_response(response)
log.info("LLM 文档提取成功: %s", file_path.name)
return result
except Exception as e:
log.error("LLM 文档提取失败: %s (%s)", file_path.name, e)
raise
def _llm_query_multimodal(
system_prompt: str,
text: str | None = None,
image_b64s: list[str] | None = None,
blocks: list[Any] | None = None,
reasoning_effort: str = "none",
source_dir: Path | None = None,
) -> str:
"""发送多模态请求到 LLM(内部使用)。
Args:
system_prompt: 系统提示词。
text: 用户文本(与 image_b64s 配合使用,文本在前、图片在后)。
image_b64s: base64 编码的图片列表。
blocks: 预构建的内容块列表(TextBlock/ImageBlock),传入时忽略 text 和 image_b64s。
source_dir: 会话目录(可选,传入时启用 SSE 流式事件写入)。
Returns:
LLM 响应文本。
"""
from llama_index.core.base.llms.types import ImageBlock, TextBlock
from llama_index.core.llms import ChatMessage
from ...config import get_llm_config
if blocks is not None:
final_blocks = blocks
else:
text = text or ""
image_b64s = image_b64s or []
final_blocks = [TextBlock(text=text)]
for img_b64 in image_b64s:
final_blocks.append(
ImageBlock(
url=f"data:image/jpeg;base64,{img_b64}",
detail="high",
)
)
messages = [
ChatMessage(role="system", content=system_prompt),
ChatMessage(role="user", blocks=final_blocks),
]
llm_config = get_llm_config()
llm = _create_llm()
log.info(
"开始请求 LLM 多模态 (model=%s, base=%s, blocks=%d)",
llm_config["model"],
llm_config["api_base"],
len(final_blocks),
)
return _stream_llm_response(llm, messages, source_dir, reasoning_effort, log_label="LLM多模态")
def load_cache(source_dir: Path) -> dict[str, Any]:
"""从 JSON 缓存目录加载结构化数据,构建 source filename -> 缓存数据的映射。
Args:
source_dir: 源文件目录(包含 .invoice_cache 子目录)。
Returns:
{source_filename: extracted_data} 字典。
额外包含 "travel_info" 键(如果 travel_info.json 存在)。
"""
cache_map: dict[str, Any] = {}
cache_dir = source_dir / CACHE_DIR_NAME
if not cache_dir.exists():
return cache_map
for json_path in sorted(cache_dir.glob("*.json")):
try:
with open(json_path, encoding="utf-8") as f:
cache_data = json.load(f)
if json_path.name in ("travel_info.json", "normal_info.json"):
cache_map[json_path.name.replace(".json", "")] = cache_data
continue
extracted = cache_data.get("extracted_data", {})
src_file = extracted.get("_source_file", "")
if src_file:
cache_map[src_file] = extracted
except Exception as e:
log.warning(f"读取缓存失败 {json_path.name}: {e}")
return cache_map
def load_match_result(source_dir: Path) -> dict[str, list[dict[str, Any]]]:
"""从 JSON 缓存目录加载发票与支付记录的匹配结果。
Args:
source_dir: 源文件目录(包含 .invoice_cache 子目录)。
Returns:
{支付记录源文件 (含金额): [发票信息列表]} 字典。
每个发票信息包含 file, type, amount 字段。
"""
cache_dir = source_dir / CACHE_DIR_NAME
match_path = cache_dir / "match_result.json"
if not match_path.exists():
return {}
try:
with open(match_path, encoding="utf-8") as f:
result: dict[str, list[dict[str, Any]]] = json.load(f)
return result
except Exception as e:
log.warning(f"读取匹配结果缓存失败: {e}")
return {}
def build_extraction_user_message(
cache_map: dict[str, Any],
match_result: dict[str, list[dict[str, Any]]],
previous_analysis: dict[str, Any] | None = None,
) -> str:
"""构建提取请求的用户消息(供 Agent 调度使用)。
Args:
cache_map: 缓存数据映射。
match_result: 匹配结果。
previous_analysis: 上一轮 LLM 分析结果(可选,补充文件时传入作为历史上下文)。
Returns:
拼接好的用户消息字符串。
"""
parts = [
"以下是本次报销的所有源文件及其提取出的结构化数据。"
"每个源文件的数据来自 OCR 识别和发票信息提取,已按文件名分组展示。"
]
if previous_analysis:
parts.append(
"【上一轮分析结果】"
"以下是上一轮 LLM 对已有文件的分析结果。"
"注意:用户可能已补充新文件,请综合所有数据(含新文件)重新分析。"
"如果新文件填补了之前的信息缺失,请相应更新分析结果。\n"
+ json.dumps(previous_analysis, ensure_ascii=False, indent=2)
)
if match_result:
parts.append(
"【发票与支付记录匹配结果】"
"以下数据已将发票信息与对应的支付记录进行关联匹配,"
"用于判断每笔支付对应的发票和商户信息。\n" + json.dumps(match_result, ensure_ascii=False, indent=2)
)
for filename, extracted in cache_map.items():
parts.append(
f"【源文件: {filename}】"
"以下为从该文件提取的结构化发票/支付/申请单数据。\n" + json.dumps(extracted, ensure_ascii=False, indent=2)
)
parts.append("\n=== 请返回 JSON 格式结果 ===")
return "\n".join(parts)
def _extract_info(
source_dir: Path | None,
system_prompt: str,
info_type: str,
) -> dict[str, Any]:
"""通用的信息提取函数:加载缓存、构建消息、调用 LLM 并解析 JSON。
extract_travel_info 和 extract_normal_info 的公共实现。
Args:
source_dir: 源文件目录(必填,包含 .invoice_cache 子目录)。
system_prompt: 系统提示词。
info_type: 信息类型标签("差旅" 或 "普通发票"),用于日志。
Returns:
LLM 提取的结构化信息字典。
"""
if not source_dir:
log.warning("未提供 source_dir,无法加载缓存数据")
return {}
cache_map = load_cache(source_dir)
match_result = load_match_result(source_dir)
user_message = build_extraction_user_message(cache_map, match_result)
log.info("开始构建%s信息提取请求,缓存条目: %d, 匹配结果: %d", info_type, len(cache_map), len(match_result))
try:
response = llm_query_text(
system_prompt=system_prompt,
text=user_message,
reasoning_effort="low",
source_dir=source_dir,
)
result = parse_json_response(response)
log.info("LLM %s信息提取成功", info_type)
return result
except Exception as e:
log.error("LLM %s信息提取失败: %s", info_type, e)
raise
def extract_travel_info(
source_dir: Path | None = None,
) -> dict[str, Any]:
"""根据差旅发票(bot 格式),让 LLM 提取出差相关信息。
纯提取,不包含校验逻辑。校验由 Agent 层调度。
Args:
source_dir: 源文件目录(必填,包含 .invoice_cache 子目录)。
Returns:
包含出差事由、地点、交通工具、时间、住宿信息等字段的字典。
"""
return _extract_info(source_dir, build_travel_info_system_prompt(), "差旅")
def extract_normal_info(
source_dir: Path | None = None,
) -> dict[str, Any]:
"""根据普通发票(非差旅),让 LLM 提取报销相关信息。
纯提取,不包含校验逻辑。校验由 Agent 层调度。
Args:
source_dir: 源文件目录(必填,包含 .invoice_cache 子目录)。
Returns:
包含报销说明、发票总数、总金额、支付方式、附件清单等字段的字典。
"""
return _extract_info(source_dir, build_normal_info_system_prompt(), "普通发票")
# ------------------------------------------------------------------
# 用户补充信息处理
# ------------------------------------------------------------------
def process_user_supplement(
user_text: str,
extracted_info: dict[str, Any],
invoice_type: str,
source_dir: Path | None = None,
) -> dict[str, Any]:
"""让用户补充的文字信息通过 LLM 分析,返回需要更新的字段。
Args:
user_text: 用户输入的文字。
extracted_info: 当前已提取的报销信息。
invoice_type: "travel" 或 "normal"。
source_dir: 会话目录(可选,传入时启用 SSE 流式事件写入)。
Returns:
包含 updated_fields, changes, confidence, unparsed_info 的字典。
"""
system_prompt = build_supplement_system_prompt()
parts = [
f"发票类型: {'差旅报销' if invoice_type == 'travel' else '普通报销'}",
"",
"以下是当前已提取的报销信息:",
json.dumps(extracted_info, ensure_ascii=False, indent=2),
"",
f"用户补充信息:{user_text}",
"",
"=== 请分析用户输入并返回需要更新的字段 ===",
]
user_message = "\n".join(parts)
try:
response = llm_query_text(
system_prompt=system_prompt,
text=user_message,
reasoning_effort="low",
source_dir=source_dir,
)
result = parse_json_response(response)
log.info("LLM 补充信息分析完成")
return result
except Exception as e:
log.error("LLM 补充信息分析失败: %s", e)
return {
"updated_fields": {},
"changes": [],
"confidence": 0.0,
"unparsed_info": f"分析失败: {e}",
}
def merge_supplement_into_info(
extracted_info: dict[str, Any],
updated_fields: dict[str, Any],
) -> dict[str, Any]:
"""将 LLM 返回的更新字段合并到已提取的信息中。
支持点号路径(如 basic_info.travel_purpose)表示嵌套更新。
Args:
extracted_info: 当前已提取的报销信息。
updated_fields: LLM 返回的需要更新的字段。
Returns:
更新后的报销信息。
"""
import copy
result = copy.deepcopy(extracted_info)
for field_path, value in updated_fields.items():
parts = field_path.split(".")
current = result
for part in parts[:-1]:
if part not in current:
current[part] = {}
current = current[part]
current[parts[-1]] = value
log.info("更新字段 %s = %s", field_path, value)
return result
+27
View File
@@ -0,0 +1,27 @@
---
last_reviewed: 2026-06-15
---
# src/core/matching — 金额匹配
将提取到的发票数据与支付记录(刷卡截图)按金额进行匹配。
## 文件
| 文件 | 职责 |
|------|------|
| `matcher.py` | 匹配引擎:一对一匹配、一对多贪心匹配、未匹配发票处理 |
## 匹配策略
| 场景 | 策略 |
|------|------|
| 发票数 == 支付记录数 | 一对一匹配:按金额降序配对,相对容差内即匹配 |
| 发票数 > 支付记录数 | 一对多匹配:贪心算法凑金额,相对容差 3% |
| 文件名匹配 | 最高优先级:文件名(不含后缀)一致时直接匹配 |
| 未匹配发票 | 单独列为一条支付记录,`remark` 标记为 `"unmatched"` |
## 业务约束
- 发票总金额 >= 支付总金额
- 输出以支付记录为主键的结果列表
+8
View File
@@ -0,0 +1,8 @@
"""匹配模块
提供发票与支付记录的金额匹配功能。
"""
from .matcher import match_invoices_to_cards
__all__ = ["match_invoices_to_cards"]
+409
View File
@@ -0,0 +1,409 @@
"""发票与支付记录匹配
将提取到的发票数据与支付记录进行金额匹配,
输出以支付记录为主键的结果列表。
支付记录由统一提取模块(extractor)根据 LLM 返回的「invoice_type」字段分类而来,
不再按文件类型假设文档类型。
## 业务约束
- 发票数 >= 付款记录数(最少一张发票对应一张付款记录)
- 发票总金额 >= 付款总金额(发票只能比付款多,不能少)
- 若发票数 == 付款数,走一对一匹配,无需一对多
## 匹配流程
1. 接收分类好的发票和支付记录列表
2. 解析发票和刷卡记录的金额,进行总额校验
- 发票总额 < 刷卡总额时发出 warning
3. 按金额降序排序
4. 文件名匹配(最高优先级):发票和刷卡记录的文件名(不含后缀)一致时直接匹配
5. 根据数量关系选择匹配策略:
- 数量相等 → 一对一匹配:按金额从大到小依次配对,相对容差内即匹配
- 发票更多 → 一对多匹配:对每张刷卡记录贪心凑金额,相对容差内结束
6. 构建以支付记录为主键的结果列表
7. 未匹配的发票单独作为一条记录(无刷卡信息)
8. 清理内部字段,输出支付记录列表
## 容差计算
使用相对容差(默认 3%),以刷卡金额为基准:
- ¥2900 发票 vs ¥2850 刷卡 → 差 ¥50,容差 ¥85.5 → 匹配成功
- ¥100 发票 vs ¥95 刷卡 → 差 ¥5,容差 ¥3.0 → 不匹配(需精确匹配或调整)
## 一对多匹配细节
- 对每张刷卡记录,维护 remaining(剩余待匹配金额)
- 遍历未分配的发票(按金额降序):
- 若发票金额 + 容差 >= remaining,视为最后一张,匹配后退出
- 否则发票金额不超过 remaining + 容差即可匹配
- 匹配后 remaining 为负且超出容差时回滚最后一张发票
- 每张发票只会被分配一次
## 对外接口
match_invoices_to_cards(invoices, cards, tolerance) -> list[dict]
"""
from pathlib import Path
from typing import Any
from ... import get_logger
log = get_logger("matcher")
def _safe_float(value: str | None, default: float = 0.0) -> float:
"""安全转换为浮点数"""
if value is None or str(value).strip() == "":
return default
try:
return float(str(value).replace(",", ""))
except (ValueError, TypeError):
return default
def _build_invoice_summary(invoices: list[dict[str, Any]]) -> str:
"""将多张发票信息汇总为备注字符串"""
parts = []
for inv in invoices:
inv_type = inv.get("invoice_type", "unknown")
amount = inv.get("total_amount", "unknown")
if inv_type == "train":
label = inv.get("person_name") or inv.get("invoice_number", "unknown")
elif inv_type == "hotel":
label = "hotel"
else:
label = inv.get("item_name") or inv.get("invoice_number", "unknown")
parts.append(f"{inv_type}[{label}]¥{amount}")
return " | ".join(parts)
def _relative_tolerance(base: float, rate: float = 0.03) -> float:
"""根据基准金额计算相对容差(默认 3%)"""
return abs(base) * rate
def match_invoices_to_cards(
invoices: list[dict[str, Any]],
cards: list[dict[str, Any]] | None = None,
tolerance: float = 0.03,
) -> list[dict[str, Any]]:
"""将发票与支付记录按金额匹配,输出以支付记录为主键的结果列表
支付记录由统一提取模块(extractor)根据 LLM 返回的「invoice_type」字段分类而来。
业务约束:
- 发票数 >= 付款记录数
- 发票总金额 >= 付款总金额(发票只能比付款多,不能少)
- 若发票数 == 付款数,一对一匹配,无需一对多
Args:
invoices: 发票列表,需包含 "total_amount" 字段
cards: 支付记录列表,需包含 "card_amount" 字段(由 extractor 分类提供)
tolerance: 金额匹配容差比例(默认 0.03 = 3%)
Returns:
以支付记录为主键的结果列表,每条记录包含:
- card_date, card_no, card_amount(支付信息)
- 关联发票列表(_matched_invoices)
- 发票详情备注
- 未匹配发票单独作为一条无刷卡信息的记录
"""
if not cards:
log.warning("无支付记录可供匹配,发票将保持原状")
return _invoices_to_records(invoices)
# 解析金额
for card in cards:
card["_amount"] = _safe_float(card.get("card_amount"))
for inv in invoices:
inv["_amount"] = _safe_float(inv.get("total_amount"))
# 数据校验
total_invoices = sum(inv["_amount"] for inv in invoices)
total_cards = sum(card["_amount"] for card in cards)
log.info(
f"金额校验: 发票总额 ¥{total_invoices:.2f}, 刷卡总额 ¥{total_cards:.2f}, "
f"发票数 {len(invoices)}, 刷卡数 {len(cards)}"
)
total_tolerance = _relative_tolerance(max(total_invoices, total_cards), tolerance)
if total_invoices < total_cards - total_tolerance:
log.warning(
f"发票总额 (¥{total_invoices:.2f}) 小于刷卡总额 (¥{total_cards:.2f}), "
f"超出容差 {tolerance * 100:.0f}%, 匹配结果可能有偏差"
)
# 按金额降序排序
cards.sort(key=lambda c: c["_amount"], reverse=True)
invoices.sort(key=lambda i: i["_amount"], reverse=True)
# 执行匹配,返回 {card_index: [invoice_indices]} 的映射
card_to_invoices = _match(cards, invoices, tolerance)
# 构建支付记录列表
records = _build_payment_records(cards, invoices, card_to_invoices)
# 清理内部字段
for inv in invoices:
inv.pop("_amount", None)
for card in cards:
card.pop("_amount", None)
# 统计
matched_invoices = sum(len(inv_list) for inv_list in card_to_invoices.values())
unmatched_count = len(invoices) - matched_invoices
log.info(f"匹配完成: {len(records)} 条支付记录, {matched_invoices}/{len(invoices)} 张发票已关联")
if unmatched_count:
log.info(f"未匹配发票: {unmatched_count} 张(已单独列为记录)")
return records
def _match(
cards: list[dict[str, Any]],
invoices: list[dict[str, Any]],
tolerance: float,
) -> dict[int, list[int]]:
"""执行匹配,返回 {card_index: [invoice_indices]} 的映射
tolerance 为相对容差比例(如 0.03 表示 3%)
匹配优先级(从高到低):
1. 文件名匹配:发票和刷卡记录的文件名(不含后缀)一致时直接匹配
2. 精确匹配:金额差 <= 0.01 元
3. 一对一 / 一对多贪心匹配:按金额容差匹配
"""
result: dict[int, list[int]] = {}
assigned: set[int] = set()
# ---- 阶段 0:文件名匹配(最高优先级)----
_match_by_filename(invoices, cards, assigned, result)
if len(invoices) == len(cards):
_match_one_to_one(invoices, cards, tolerance, assigned, result)
else:
_match_one_to_many(invoices, cards, tolerance, assigned, result)
return result
def _match_by_filename(
invoices: list[dict[str, Any]],
cards: list[dict[str, Any]],
assigned: set[int],
result: dict[int, list[int]],
) -> None:
"""文件名匹配:发票和刷卡记录的文件名(不含后缀)一致时直接匹配"""
for card_idx, card in enumerate(cards):
card_name = card.get("_source_file", "")
if not card_name:
continue
card_stem = Path(card_name).stem
for idx, inv in enumerate(invoices):
if idx in assigned:
continue
inv_name = inv.get("_source_file", "")
if not inv_name:
continue
inv_stem = Path(inv_name).stem
if inv_stem == card_stem:
assigned.add(idx)
result[card_idx] = [idx]
log.info(f"[文件名匹配] {inv_name} ↔ {card_name}")
break
def _match_one_to_one(
invoices: list[dict[str, Any]],
cards: list[dict[str, Any]],
tolerance: float,
assigned: set[int],
result: dict[int, list[int]],
) -> None:
"""一对一匹配:发票数等于刷卡数,对每张刷卡记录寻找金额最接近的未分配发票"""
for card_idx, card in enumerate(cards):
if card_idx in result:
continue
card_amount = card["_amount"]
card_tol = _relative_tolerance(card_amount, tolerance)
# 在未分配的发票中找金额最接近的
best_idx = -1
best_diff = float("inf")
for idx, inv in enumerate(invoices):
if idx in assigned:
continue
diff = abs(inv["_amount"] - card_amount)
if diff < best_diff:
best_diff = diff
best_idx = idx
if best_idx >= 0 and best_diff <= card_tol:
inv = invoices[best_idx]
assigned.add(best_idx)
result[card_idx] = [best_idx]
log.info(
f"[一对一] {inv.get('invoice_number', 'unknown')} ¥{inv['_amount']:.2f} "
f"↔ {card.get('_source_file', 'unknown')} ¥{card_amount:.2f}"
)
elif best_idx >= 0:
inv = invoices[best_idx]
log.warning(
f"[一对一] 金额偏差超出容差: "
f"{inv.get('invoice_number', 'unknown')} ¥{inv['_amount']:.2f} "
f"vs ¥{card_amount:.2f} (差 ¥{best_diff:.2f}, 容差 ¥{card_tol:.2f})"
)
def _match_one_to_many(
invoices: list[dict[str, Any]],
cards: list[dict[str, Any]],
tolerance: float,
assigned: set[int],
result: dict[int, list[int]],
) -> None:
"""一对多匹配:一张刷卡可能对应多张发票,按金额从大到小贪心匹配"""
# ---- 阶段 1:精确匹配(金额差 <= 0.01 元视为相等)----
exact_tolerance = 0.01
for card_idx, card in enumerate(cards):
card_amount = card["_amount"]
if card_amount <= 0:
continue
for idx, inv in enumerate(invoices):
if idx in assigned:
continue
inv_amount = inv["_amount"]
if inv_amount <= 0:
continue
if abs(inv_amount - card_amount) <= exact_tolerance:
assigned.add(idx)
result[card_idx] = [idx]
log.info(
f"[一对多-精确] {inv.get('invoice_number', 'unknown')} ¥{inv_amount:.2f} "
f"↔ {card.get('_source_file', 'unknown')} ¥{card_amount:.2f}"
)
break
# ---- 阶段 2:贪心匹配(仅处理未精确匹配的刷卡记录)----
for card_idx, card in enumerate(cards):
if card_idx in result:
continue
card_amount = card["_amount"]
if card_amount <= 0:
continue
card_tol = _relative_tolerance(card_amount, tolerance)
remaining = card_amount
matched_indices: list[int] = []
for idx, inv in enumerate(invoices):
if idx in assigned:
continue
if remaining <= card_tol:
break
inv_amount = inv["_amount"]
if inv_amount <= 0:
continue
if inv_amount + card_tol >= remaining:
is_match = True
else:
is_match = inv_amount <= remaining + card_tol
if is_match:
assigned.add(idx)
matched_indices.append(idx)
remaining -= inv_amount
if remaining <= card_tol:
break
# 回滚:如果匹配后 remaining 为负且超出容差
if remaining < -card_tol and matched_indices:
last_idx = matched_indices.pop()
assigned.discard(last_idx)
remaining += invoices[last_idx]["_amount"]
# 记录匹配结果
if matched_indices:
result[card_idx] = matched_indices
for idx in matched_indices:
inv = invoices[idx]
log.info(
f"[一对多-贪心] {inv.get('invoice_number', 'unknown')} ¥{inv['_amount']:.2f} "
f"→ {card.get('_source_file', 'unknown')} ¥{card['_amount']:.2f}"
)
def _build_payment_records(
cards: list[dict[str, Any]],
invoices: list[dict[str, Any]],
card_to_invoices: dict[int, list[int]],
) -> list[dict[str, Any]]:
"""构建以支付记录为主键的结果列表"""
records: list[dict[str, Any]] = []
for card_idx, inv_indices in card_to_invoices.items():
card = cards[card_idx]
matched_invs = [invoices[idx] for idx in inv_indices]
record = {
"card_date": card.get("card_date", ""),
"card_no": card.get("card_no", ""),
"card_amount": str(card["_amount"]),
"relative_invoice_count": str(len(matched_invs)),
"invoice_detail": _build_invoice_summary(matched_invs),
"remark": "",
"_source_file": card.get("_source_file", ""),
"_matched_invoices": matched_invs,
}
records.append(record)
# 未匹配的发票,单独作为记录
matched_indices = set()
for inv_indices in card_to_invoices.values():
matched_indices.update(inv_indices)
unmatched = [inv for idx, inv in enumerate(invoices) if idx not in matched_indices]
for inv in unmatched:
record = {
"card_date": "",
"card_no": "",
"card_amount": "",
"relative_invoice_count": "1",
"invoice_detail": _build_invoice_summary([inv]),
"remark": "unmatched",
"_matched_invoices": [inv],
}
records.append(record)
return records
def _invoices_to_records(invoices: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""无刷卡记录时,将每张发票转为独立记录"""
records = []
for inv in invoices:
record = {
"card_date": "",
"card_no": "",
"card_amount": "",
"relative_invoice_count": "1",
"invoice_detail": _build_invoice_summary([inv]),
"remark": "",
"_matched_invoices": [inv],
}
records.append(record)
return records
+34
View File
@@ -0,0 +1,34 @@
---
last_reviewed: 2026-06-15
---
# src/core/validation — 信息校验
对 LLM 提取的报销信息进行声明式规则校验,判断是否满足填报要求。
## 文件
| 文件 | 职责 |
|------|------|
| `validator.py` | 校验引擎:加载 JSON 规则配置 → 遍历字段/数组 → 输出校验报告 |
## 设计特点
- **规则与引擎分离**:校验规则存储在 `config/validation_rules.json`,引擎只负责执行
- **统一路径定位**:使用 `path` 列表定位嵌套字段,如 `["basic_info", "travel_purpose"]`
- **自定义校验**:支持 `custom_check` 函数(日期格式、正数检查等)
- **数组元素校验**:支持 `min_items` 最小数量 + 每个元素的必填字段
## 校验规则类型
| 类型 | 用途 | 配置项 |
|------|------|--------|
| `fields` | 顶层单值字段 | `path`, `required`, `custom_check`, `check_empty` |
| `arrays` | 数组字段 | `path`, `min_items`, `element_fields` |
## 对外接口
| 函数 | 说明 |
|------|------|
| `validate(info, invoice_type)` | 执行校验,返回 `ValidationReport` |
| `get_missing_fields(report)` | 提取缺失字段列表 |
+28
View File
@@ -0,0 +1,28 @@
"""校验模块
提供报销信息的规则级校验功能。
"""
from .validator import (
ArrayRule,
FieldRule,
ValidationReport,
ValidationRules,
get_validation_rules,
reload_validation_rules,
validate_extracted_info,
validate_normal_info,
validate_travel_info,
)
__all__ = [
"validate_extracted_info",
"validate_travel_info",
"validate_normal_info",
"ValidationReport",
"FieldRule",
"ArrayRule",
"ValidationRules",
"get_validation_rules",
"reload_validation_rules",
]
+537
View File
@@ -0,0 +1,537 @@
"""信息完整性校验器
对 LLM 提取的报销信息进行规则级校验,判断是否满足填报要求。
校验规则从 JSON 配置文件加载,支持声明式配置。
设计理念:
使用声明式规则配置,将校验规则与校验逻辑分离,提高可读性和可维护性。
"""
from __future__ import annotations
import json
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, TypedDict
from ... import get_logger
log = get_logger("validator")
# ------------------------------------------------------------------
# 日期格式
# ------------------------------------------------------------------
DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")
def _is_valid_date(value: str) -> bool:
"""检查日期格式是否为 YYYY-MM-DD。"""
return bool(DATE_PATTERN.match(value))
def _is_positive_number(value: Any) -> bool:
"""检查值是否为正数(整数或浮点数)。"""
return isinstance(value, int | float) and value > 0
def _is_positive_integer(value: Any) -> bool:
"""检查值是否为正整数。"""
return isinstance(value, int) and value > 0
# 自定义校验函数注册表
_CUSTOM_CHECKS: dict[str, Callable[[Any], bool]] = {
"is_valid_date": _is_valid_date,
"is_positive_number": _is_positive_number,
"is_positive_integer": _is_positive_integer,
}
# ------------------------------------------------------------------
# 规则定义
# ------------------------------------------------------------------
class FieldRule(TypedDict, total=False):
"""字段校验规则(统一使用 path 定位)"""
path: list[str] # 字段路径(统一定位方式)
required: bool = True # 是否必填(默认必填)
check_empty: bool = True # 是否检查空字符串(默认检查)
custom_check: str | Callable[[Any], bool] | None = None # 自定义校验函数(名称或函数)
description: str = "" # 字段描述(用于生成友好提示)
class ArrayRule(TypedDict, total=False):
"""数组校验规则"""
path: list[str] # 数组路径
min_items: int = 1 # 最小元素数量
element_fields: list[str | FieldRule] = [] # 元素字段规则
description: str = "" # 数组描述
class ValidationRules(TypedDict):
"""校验规则集合"""
fields: list[FieldRule] # 字段规则列表
arrays: list[ArrayRule] # 数组规则列表
class ValidationConfig(TypedDict):
"""校验配置结构"""
version: str
custom_checks: dict[str, str]
travel: ValidationRules
normal: ValidationRules
# ------------------------------------------------------------------
# 配置加载
# ------------------------------------------------------------------
_CONFIG_PATH = Path(__file__).parent.parent.parent / "config" / "validation_rules.json"
_cached_rules: ValidationConfig | None = None
def _load_validation_config() -> ValidationConfig:
"""加载校验规则配置文件。"""
global _cached_rules
if _cached_rules is not None:
return _cached_rules
if not _CONFIG_PATH.exists():
log.warning("校验规则配置文件不存在: %s,使用内置默认规则", _CONFIG_PATH)
return _load_default_rules()
try:
with open(_CONFIG_PATH, encoding="utf-8") as f:
config = json.load(f)
_cached_rules = _resolve_custom_checks(config)
log.info("校验规则配置加载成功")
return _cached_rules
except Exception as e:
log.error("加载校验规则配置失败: %s,使用内置默认规则", e)
return _load_default_rules()
def _resolve_custom_checks(config: dict[str, Any]) -> ValidationConfig:
"""解析配置中的自定义校验函数名称,替换为实际函数引用。"""
def resolve_rule(rule: dict[str, Any]) -> dict[str, Any]:
if "custom_check" in rule and isinstance(rule["custom_check"], str):
check_name = rule["custom_check"]
if check_name in _CUSTOM_CHECKS:
rule["custom_check"] = _CUSTOM_CHECKS[check_name]
else:
log.warning("未知的自定义校验函数: %s", check_name)
rule["custom_check"] = None
return rule
# 解析 travel 规则的 fields
for field_rule in config.get("travel", {}).get("fields", []):
resolve_rule(field_rule)
# 解析 element_fields
for array_rule in config.get("travel", {}).get("arrays", []):
for elem_field in array_rule.get("element_fields", []):
if isinstance(elem_field, dict):
resolve_rule(elem_field)
# 解析 normal 规则的 fields
for field_rule in config.get("normal", {}).get("fields", []):
resolve_rule(field_rule)
# 解析 element_fields
for array_rule in config.get("normal", {}).get("arrays", []):
for elem_field in array_rule.get("element_fields", []):
if isinstance(elem_field, dict):
resolve_rule(elem_field)
return config # type: ignore[return-value]
def _load_default_rules() -> ValidationConfig:
"""返回内置的默认校验规则(当配置文件不存在时使用)。"""
return {
"version": "1.0",
"custom_checks": {},
"travel": {
"fields": [
{"path": ["basic_info", "travel_purpose"], "description": "出差事由"},
{"path": ["basic_info", "travel_location"], "description": "出差地点"},
{"path": ["basic_info", "start_date"], "custom_check": _is_valid_date, "description": "出差开始日期"},
{"path": ["basic_info", "end_date"], "custom_check": _is_valid_date, "description": "出差结束日期"},
],
"arrays": [
{
"path": ["reimbursement_details", "transport_fee"],
"min_items": 1,
"element_fields": [
{"path": ["vehicle_type"], "description": "交通工具类型"},
{"path": ["start_date"], "custom_check": _is_valid_date, "description": "出发日期"},
{"path": ["end_date"], "custom_check": _is_valid_date, "description": "到达日期"},
{"path": ["departure_place"], "description": "出发地"},
{"path": ["arrival_place"], "description": "目的地"},
{"path": ["amount"], "custom_check": _is_positive_number, "description": "金额"},
{"path": ["bill_count"], "custom_check": _is_positive_integer, "description": "票据张数"},
{"path": ["remark"], "check_empty": False, "description": "备注说明"},
],
"description": "交通费用明细",
},
{
"path": ["payment_methods"],
"min_items": 1,
"element_fields": [
{"path": ["card_date"], "custom_check": _is_valid_date, "description": "刷卡日期"},
{"path": ["card_amount"], "custom_check": _is_positive_number, "description": "支付金额"},
{"path": ["merchant"], "description": "商户名称"},
{"path": ["remark"], "check_empty": False, "description": "备注"},
],
"description": "支付方式记录",
},
{
"path": ["subsidy_list"],
"min_items": 1,
"element_fields": [
{"path": ["person_id"], "description": "人员工号"},
{"path": ["person_name"], "description": "人员姓名"},
{"path": ["start_date"], "custom_check": _is_valid_date, "description": "补助开始日期"},
{"path": ["end_date"], "custom_check": _is_valid_date, "description": "补助结束日期"},
{"path": ["days"], "custom_check": _is_positive_integer, "description": "补助天数"},
],
"description": "补助清单",
},
{
"path": ["attachments"],
"min_items": 0,
"element_fields": [
{"path": ["filename"], "description": "文件名"},
{"path": ["attachment_type"], "description": "附件类型"},
],
"description": "附件列表",
},
],
},
"normal": {
"fields": [
{"path": ["basic_info", "reimbursement_description"], "description": "报销事由"},
{
"path": ["reimbursement_details", "total_invoices"],
"custom_check": _is_positive_integer,
"description": "发票总数",
},
{
"path": ["reimbursement_details", "total_amount"],
"custom_check": _is_positive_number,
"description": "总金额",
},
],
"arrays": [
{
"path": ["payment_methods"],
"min_items": 1,
"element_fields": [
{"path": ["card_date"], "custom_check": _is_valid_date, "description": "刷卡日期"},
{"path": ["card_amount"], "custom_check": _is_positive_number, "description": "支付金额"},
{"path": ["merchant"], "description": "商户名称"},
{"path": ["remark"], "check_empty": False, "description": "备注"},
],
"description": "支付方式记录",
},
{
"path": ["attachments"],
"min_items": 0,
"element_fields": [
{"path": ["filename"], "description": "文件名"},
{"path": ["attachment_type"], "description": "附件类型"},
],
"description": "附件列表",
},
],
},
}
def get_validation_rules(invoice_type: str) -> ValidationRules:
"""获取指定发票类型的校验规则。
Args:
invoice_type: 发票类型,'travel' 或 'normal'。
Returns:
对应的校验规则。
"""
config = _load_validation_config()
return config.get(invoice_type, config.get("travel", {})) # type: ignore[return-value]
# ------------------------------------------------------------------
# 数据模型
# ------------------------------------------------------------------
@dataclass
class ValidationReport:
"""校验结果报告"""
valid: bool
missing_fields: list[str] = field(default_factory=list)
missing_materials: list[str] = field(default_factory=list)
confidence: float = 0.0
suggestion: str = ""
# ------------------------------------------------------------------
# 通用校验引擎
# ------------------------------------------------------------------
def _check_field(
data: dict[str, Any],
rule: FieldRule,
) -> tuple[bool, str]:
"""根据字段规则检查字段。"""
path = rule["path"]
check_empty = rule.get("check_empty", True)
custom_check = rule.get("custom_check")
current = data
for key in path:
if not isinstance(current, dict):
return (False, ".".join(path))
if key not in current:
return (False, ".".join(path))
current = current[key]
if check_empty and isinstance(current, str) and not current.strip():
return (False, ".".join(path))
if custom_check is not None and not custom_check(current):
return (False, ".".join(path))
return (True, ".".join(path))
def _check_array(
data: dict[str, Any],
rule: ArrayRule,
) -> tuple[list[str], int, int]:
"""根据数组规则检查数组。
Returns:
(缺失字段列表, 总检查数, 通过检查数)
"""
missing: list[str] = []
path = rule["path"]
min_items = rule.get("min_items", 1)
element_fields = rule.get("element_fields", [])
path_str = ".".join(path)
total_checks = 1 # 数组存在性和最小数量检查
passed_checks = 0
# 遍历路径获取数组
current = data
for key in path:
if not isinstance(current, dict) or key not in current:
return ([path_str], total_checks, passed_checks)
current = current[key]
# 检查数组是否满足最小数量要求
if not isinstance(current, list) or len(current) < min_items:
return ([path_str], total_checks, passed_checks)
passed_checks += 1 # 数组检查通过
# 检查数组元素的字段
if element_fields:
for i, item in enumerate(current):
if not isinstance(item, dict):
missing.append(f"{path_str}[{i}]")
total_checks += len(element_fields)
continue
for field_rule in element_fields:
total_checks += 1
# 支持两种格式:简单字符串格式 和 详细规则格式
if isinstance(field_rule, str):
field_rule_dict: FieldRule = {"path": [field_rule]}
else:
field_rule_dict = field_rule
# 复用 _check_field 函数检查元素字段
ok, _ = _check_field(item, field_rule_dict)
if ok:
passed_checks += 1
else:
field_path_str = ".".join(field_rule_dict["path"])
missing.append(f"{path_str}[{i}].{field_path_str}")
return missing, total_checks, passed_checks
def _validate_with_rules(data: dict[str, Any], rules: ValidationRules) -> tuple[list[str], int, int]:
"""使用规则配置进行校验。"""
missing: list[str] = []
total_checks = 0
passed_checks = 0
# 校验字段规则
for rule in rules.get("fields", []):
total_checks += 1
ok, field_path = _check_field(data, rule)
if ok:
passed_checks += 1
else:
missing.append(field_path)
# 校验数组规则
for rule in rules.get("arrays", []):
array_missing, array_total, array_passed = _check_array(data, rule)
total_checks += array_total
passed_checks += array_passed
missing.extend(array_missing)
return missing, total_checks, passed_checks
# ------------------------------------------------------------------
# 校验入口
# ------------------------------------------------------------------
def validate_travel_info(data: dict[str, Any]) -> ValidationReport:
"""校验差旅报销信息的完整性。"""
rules = get_validation_rules("travel")
missing, total_checks, passed_checks = _validate_with_rules(data, rules)
confidence = passed_checks / total_checks if total_checks > 0 else 0.0
missing_materials = _infer_missing_materials(missing, data)
suggestion = _build_suggestion(missing, missing_materials)
return ValidationReport(
valid=len(missing) == 0,
missing_fields=missing,
missing_materials=missing_materials,
confidence=round(confidence, 2),
suggestion=suggestion,
)
def validate_normal_info(data: dict[str, Any]) -> ValidationReport:
"""校验普通报销信息的完整性。"""
rules = get_validation_rules("normal")
missing, total_checks, passed_checks = _validate_with_rules(data, rules)
confidence = passed_checks / total_checks if total_checks > 0 else 0.0
missing_materials = _infer_missing_materials(missing, data)
suggestion = _build_suggestion(missing, missing_materials)
return ValidationReport(
valid=len(missing) == 0,
missing_fields=missing,
missing_materials=missing_materials,
confidence=round(confidence, 2),
suggestion=suggestion,
)
# ------------------------------------------------------------------
# 缺失材料推断
# ------------------------------------------------------------------
def _infer_missing_materials(
missing_fields: list[str],
data: dict[str, Any],
) -> list[str]:
"""根据缺失字段推断可能需要补充的材料类型。"""
materials: list[str] = []
field_set = set(missing_fields)
if any("start_date" in f or "end_date" in f for f in field_set):
if "basic_info.start_date" in field_set or "basic_info.end_date" in field_set:
materials.append("出差事前申请单")
if "basic_info.travel_purpose" in field_set:
materials.append("出差事前申请单")
if "basic_info.travel_location" in field_set:
materials.append("交通工具发票")
if "payment_methods" in field_set or any("payment_methods[" in f for f in field_set):
materials.append("支付记录截图")
if any("transport_fee" in f for f in field_set):
materials.append("交通工具发票")
if any("subsidy_list" in f for f in field_set):
materials.append("出差事前申请单")
if "basic_info.reimbursement_description" in field_set:
materials.append("发票或支付记录")
return list(dict.fromkeys(materials))
def _build_suggestion(
missing_fields: list[str],
missing_materials: list[str],
) -> str:
"""生成用户友好的建议信息。"""
if not missing_fields:
return ""
if missing_materials:
material_names = "、".join(missing_materials)
return f"信息不完整,请补充上传:{material_names}"
return f"信息不完整,缺少 {len(missing_fields)} 个字段"
# ------------------------------------------------------------------
# 统一入口
# ------------------------------------------------------------------
def validate_extracted_info(
data: dict[str, Any],
invoice_type: str = "travel",
) -> ValidationReport:
"""校验提取信息的完整性。
Args:
data: LLM 提取的结构化信息。
invoice_type: 发票类型,'travel' 或 'normal'。
Returns:
校验报告。
"""
log.info("开始校验 %s 报销信息完整性", invoice_type)
if invoice_type == "travel":
report = validate_travel_info(data)
else:
report = validate_normal_info(data)
status = "通过" if report.valid else "未通过"
log.info(
"校验结果: %s (置信度: %.0f%%, 缺失字段: %d)",
status,
report.confidence * 100,
len(report.missing_fields),
)
return report
def reload_validation_rules() -> None:
"""重新加载校验规则配置(用于运行时热更新)。"""
global _cached_rules
_cached_rules = None
_load_validation_config()
log.info("校验规则已重新加载")