做银行与金融机构的内部审计项目时最折磨人的往往不是算法选型而是流程本身。审计工作并不是“拿一个大模型问几个问题”那么简单它要完成材料收集、要素提取、政策比对、风险分级、底稿归档等一连串动作每一步都附带合规约束和留痕要求。传统规则脚本只能覆盖“能明确写死”的逻辑遇到合同语义、签章完整性、材料一致性这类模糊判断就不得不依赖人工逐份翻阅。最近我们在内部审计场景里搭建了一条 AI workflow把 PDF 解析、LLM 结构化抽取、规则校验、风险评分、人工复核完整串成了一条可追踪的流水线。这篇文章会把整套方案拆开来讲包括业务背景、工作流设计、完整可运行的 Python 代码、配置文件、常见坑点以及生产落地建议。无论你是金融行业的数据/风控/审计系统开发还是想在银行类场景落地 AI 应用都可以照着这套思路做一次最小验证。1. 为什么要为银行内部审计设计 AI 工作流1.1 内部审计到底在审什么银行内部审计与外部监管审计不一样它更像是银行内部的“体检系统”。审计人员会定期或不定期抽查业务材料验证流程是否规范、风险是否可控、操作是否违规。典型的内部审计场景包括对零售信贷业务做抽样检查核对贷款合同、收入证明、抵押材料是否完整合规。对公司客户授信材料做风险复核检查审批权限、授信额度、利率定价是否符合内部政策。对柜面业务凭证做一致性核验确认客户签名、证件信息、交易流水是否存在异常。对反洗钱可疑交易报告做复核判断上报依据是否充分。这些场景有一个共同特点材料多、字段多、规则细。审计员往往需要在短时间内阅读大量 PDF、图片、扫描件再对照内部制度一条条核对。这个过程既考验业务理解也考验耐心最容易出现的问题是漏看字段、规则理解不一致、底稿记录不完整。1.2 传统审计流程的局限传统方案主要有两条路。第一种是纯人工核对优点是判断灵活能处理语义和上下文但效率低、成本高而且不同审计员的判断标准可能不一致。第二种是写固定规则脚本比如用正则表达式提取合同编号、用 Excel 公式比对金额这种方式确定性强、执行快但只能覆盖“能提前枚举”的逻辑。实际审计材料往往比规则枚举复杂得多。比如一份收入证明上写的是“年薪 30 万”而贷款申请金额对应的月供压力明显超出合理区间这个判断需要模型理解语义又比如合同盖章位置偏淡、印章被遮挡传统 OCR 很难直接判断“是否存在有效签章”这又涉及图像特征和语义理解的结合。这些场景说明固定规则引擎不够用纯人工又累中间存在一个很大的智能自动化空间。1.3 AI Workflow 在这里扮演什么角色AI Workflow 并不是“调用一个 AI 接口就结束”而是把多个能力编排成一条可追踪、可重跑、可审计的业务流水线。以贷款材料合规性检查为例一条典型的 AI 工作流可以这样设计输入材料PDF/图片 ↓ 文档解析PDF 文本提取 / OCR ↓ 字段抽取LLM 抽取合同编号、客户名、金额、利率等 ↓ 规则校验金额上限、利率上限、材料完整性 ↓ 风险评分与分级 ↓ 人工复核队列 / 审计底稿归档每一步都有明确的输入输出任何一步失败都可以单独重试不会因为一个文件解析失败就导致整批案件停滞。更重要的是每一步都有日志和状态记录符合金融行业的留痕要求。1.4 AI Agent 与 Workflow 的边界最近 AI Agent 很火很多团队会直接把审计流程做成一个“什么都能干的 Agent”。但从银行审计角度看Agent 的自主性必须被限制。Agent 适合在有限范围内做动态规划比如根据风险等级选择不同的检查分支但审计场景中Agent 不能随意读取无关数据、不能修改底稿、不能绕过人工复核。更稳妥的做法是把 Agent 封装成工作流里的一个节点同时给它定义清楚工具边界。比如允许 Agent 调用“材料检索工具”“政策查询工具”但不允许它调用“删除案件工具”“直接修改评级工具”。每次工具调用都要记录参数与返回结果最终高风险判定必须经过人工确认。简单说在银行内部审计里AI 是“辅助放大器”不是“替代决策者”。2. 环境准备与项目结构2.1 技术栈选型为了兼顾可读性与可运行性本文的示例基于以下技术栈Python 3.10 及以上主要在 Linux 服务器上运行。FastAPI 作为可选的服务暴露层方便将工作流包装为内部接口。pdfplumber 用于提取 PDF 文本。Pydantic 用于定义结构化输出模型校验 LLM 抽取结果。OpenAPI 兼容客户端例如 OpenAI SDK调用大模型接口可切换到内部部署模型。SQLite 作为轻量状态存储生产环境可替换为 PostgreSQL。PyYAML 读取配置文件。如果你的公司内部大模型只提供了兼容 HTTP 接口也可以直接用requests替换 OpenAI SDK核心逻辑不变。2.2 示例项目结构建议先按分层思想组织代码不要把所有逻辑写在一个文件里。一个最小可运行的项目结构如下audit_workflow/ ├── config.yaml # 全局配置 ├── requirements.txt # 依赖列表 ├── main.py # 命令行入口 ├── src/ │ ├── __init__.py │ ├── pdf_parser.py # 文档解析模块 │ ├── schemas.py # Pydantic 数据模型 │ ├── llm_extractor.py # LLM 字段抽取模块 │ ├── compliance_rules.py # 规则校验模块 │ ├── risk_scorer.py # 风险评分模块 │ ├── workflow.py # 工作流编排核心 │ └── storage.py # 状态与结果存储 └── data/ ├── input/ # 放入待检查的 PDF └── output/ # 生成结果报告在真实项目中这个结构还可以继续拆出prompts/目录存放提示词模板、tests/目录存放单元测试、migrations/目录存放数据库变更脚本。这里先保持精简方便看清主体逻辑。2.3 依赖安装requirements.txt中建议不要锁死过细的版本先用范围约束避免因为某个小版本升级导致连锁冲突fastapi0.110,0.120 uvicorn0.29,0.40 pydantic2.6,3.0 pyyaml6.0,7.0 pdfplumber0.11,0.13 openai1.30,2.0 python-multipart0.0.9然后执行安装命令pip install -r requirements.txt如果只跑命令行示例不启动 FastAPI可以把fastapi、uvicorn、python-multipart去掉。不同公司内网环境差异较大建议实测后再统一版本。3. 核心原理工作流分层与关键设计3.1 工作流分层的价值把 AI 审计流程拆成分层不只是代码整洁的问题更是为了可控性。数据接入层负责把各种格式的文档转成统一文本信息抽取层把非结构化文本转成结构化字段规则校验层把结构化字段与制度规则做确定性比对风险分级层结合规则结果与模型置信度输出风险等级人工复核层保证关键结论有人兜底。这种分层让每一层都可以独立测试。文档解析结果差单独调 OCR 或解析参数字段抽不准只改提示词和 Pydantic 模型规则误报只改规则配置。如果没有分层所有逻辑混在一起出了问题几乎无法定位。3.2 为什么不能只用一个 Prompt有人会问“既然大模型能读文档能不能把全套材料丢给大模型让它直接输出一份审计结论” 理论上可以工程上不推荐。首先是上下文限制。一份贷款材料可能包含几十页合同、证明、银行流水全部塞进上下文既浪费 token也会降低抽取准确率。工作流可以先做分页抽取、关键段落定位再聚合结果。其次是稳定性。大模型输出天然有不确定性完整风险结论如果一次性生成很难校验对错。分层之后每一层都有结构化输出例如“抽取字段是否完整”“规则是否命中”这些中间结果可以追踪也方便人工复核。第三是审计要求。银行审计要求结论可解释、可追溯到原始材料。如果只给一个 Prompt 生成整体判断审计员无法快速定位“这个高风险评级到底是因为哪一条规则”产生的。分层之后结论可以关联到具体字段和规则比如“贷款金额 120 万超过内部上限 100 万”这样审计底稿才有说服力。3.3 人在环路Human-in-the-loopAI 工作流在银行落地的核心原则不是“全自动”而是“人机协同”。系统自动完成初筛和标注审计员把精力集中在高风险案件上。本文示例会在风险评分后增加一个人工复核队列。当风险分数超过阈值时案件状态变为 “PENDING_REVIEW”由审计员检查 AI 生成的发现项后确认或驳回。所有人工操作都要记录操作人、操作时间、操作内容最终形成闭环。3.4 提示词与结构化输出设计为了让 LLM 输出可被程序消费必须定义结构化输出格式。推荐用 JSON 输出再用 Pydantic 做校验。提示词里至少包含三部分任务描述说明当前要抽取哪些字段、字段的业务含义。输出格式明确要求返回 JSON并给出 JSON Schema。输出要求要求模型无法确定时输出null或unknown不要强行编造。示例提示词片段如下你是一个银行信贷材料审核助手。请从贷款合同中抽取以下字段 contract_no, customer_name, loan_amount, interest_rate, loan_term_months, signed_date, has_stamp, has_signature 要求 1. 输出 JSON不要输出其他解释文本。 2. 金额单位统一为元利率统一为小数。 3. 如果文档中无法找到某个字段请输出 null。给两个 few-shot 示例会让效果明显提升尤其是金额、日期这类格式容易混乱的字段。后续可以把提示词模板抽到独立目录方便运营同学调整。4. 完整实战贷款材料合规性 AI 检查工作流4.1 场景说明假设我们有若干份贷款申请材料 PDF需要完成以下检查是否包含合同、身份证、收入证明三类必要材料。合同中的客户姓名、身份证号、金额、利率、期限是否完整。贷款金额是否超过内部上限示例为 100 万。年化利率是否超过内部上限示例为 15%。贷款期限是否超过上限示例为 60 个月。合同是否包含签章。输出风险评分和建议动作自动通过 / 人工复核 / 退回补充材料。4.2 创建项目结构在任意目录执行mkdir -p audit_workflow/src audit_workflow/data/input audit_workflow/data/output然后把上一节的项目结构里的文件逐步创建出来。4.3 编写配置文件在audit_workflow/config.yaml中写入storage: input_dir: ./data/input output_dir: ./data/output db_path: ./data/audit_workflow.db model: base_url: http://localhost:8000/v1 api_key: sk-local-test model_name: auditor-llm temperature: 0.1 timeout: 60 compliance: max_loan_amount: 1000000 max_interest_rate: 0.15 max_loan_term_months: 60 high_risk_threshold: 80 review_threshold: 60 extraction: chunk_size: 1500配置项说明input_dir是待审材料目录每个 PDF 对应一个案件。db_path是 SQLite 数据库路径用于记录流程状态和结果。base_url和api_key是大模型服务地址与密钥。如果使用公司内部模型通常也是 OpenAI 兼容格式按内部文档替换即可。chunk_size用于处理超长文档时的文本分段避免上下文过长。4.4 定义结构化数据模型文件路径src/schemas.pyfrom pydantic import BaseModel, Field from typing import Optional class LoanContract(BaseModel): contract_no: Optional[str] Field(defaultNone, description合同编号) customer_name: Optional[str] Field(defaultNone, description客户姓名) id_card: Optional[str] Field(defaultNone, description身份证号) loan_amount: Optional[float] Field(defaultNone, description贷款金额元) interest_rate: Optional[float] Field(defaultNone, description年化利率小数) loan_term_months: Optional[int] Field(defaultNone, description贷款期限月) signed_date: Optional[str] Field(defaultNone, description签署日期) has_stamp: Optional[bool] Field(defaultNone, description是否有盖章) has_signature: Optional[bool] Field(defaultNone, description是否有签名) class Finding(BaseModel): rule_code: str Field(description命中的规则编码) message: str Field(description规则命中说明) evidence: str Field(description证据原文片段) severity: str Field(description风险级别high/medium/low) class AuditResult(BaseModel): case_id: str Field(description案件编号通常来自文件名) contract: LoanContract Field(description抽取的合同字段) findings: list[Finding] Field(default_factorylist, description规则发现项) risk_score: int Field(ge0, le100, description风险分数) risk_level: str Field(description风险等级high/medium/low) action: str Field(description建议动作pass/review/return)这里把抽取结果、规则命中、最终评级拆成三个模型有利于后续每个节点的独立测试和存储。4.5 文档解析模块文件路径src/pdf_parser.pyimport pdfplumber from pathlib import Path def extract_text_from_pdf(pdf_path: str) - str: 从 PDF 文件中提取文本内容。 text_parts [] with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: page_text page.extract_text() if page_text: text_parts.append(page_text) return \n.join(text_parts) def extract_text_from_file(file_path: str) - str: 根据文件后缀选择解析方式便于后续扩展 OCR 能力。 suffix Path(file_path).suffix.lower() if suffix .pdf: return extract_text_from_pdf(file_path) # 如果是图片或扫描件建议接入 OCR 服务本文示例只处理 PDF 文本 raise ValueError(f暂不支持的文件类型: {suffix})如果材料是扫描件pdfplumber提取不到文本此时需要接入 OCR。常见方案是调用内部 OCR 服务或使用 PaddleOCR 等开源工具。把解析逻辑独立成模块后续替换 OCR 就不会影响其他层。4.6 LLM 字段抽取模块文件路径src/llm_extractor.pyimport json import yaml from openai import OpenAI from .schemas import LoanContract PROMPT_TEMPLATE 你是一个银行信贷材料审核助手。 请从以下贷款合同文本中抽取字段只输出 JSON不要输出解释。 字段包括 contract_no合同编号 customer_name客户姓名 id_card身份证号 loan_amount贷款金额单位元 interest_rate年化利率小数形式例如 0.12 表示 12% loan_term_months贷款期限单位月 signed_date签署日期格式 YYYY-MM-DD has_stamp是否有盖章true/false has_signature是否有签名true/false 注意 1. 金额必须转为数值去掉单位。 2. 利率必须转为小数。 3. 找不到的字段输出 null。 合同文本 {text} class LLMExtractor: def __init__(self, config: dict): self.client OpenAI( base_urlconfig[model][base_url], api_keyconfig[model][api_key], timeoutconfig[model].get(timeout, 60), ) self.model_name config[model][model_name] self.temperature config[model].get(temperature, 0.1) def extract(self, text: str) - LoanContract: prompt PROMPT_TEMPLATE.format(texttext[:3000]) resp self.client.chat.completions.create( modelself.model_name, messages[{role: user, content: prompt}], temperatureself.temperature, ) content resp.choices[0].message.content.strip() # 兼容模型返回带 json 标记的情况 if content.startswith(): content content.strip() if content.startswith(json): content content[4:] data json.loads(content) return LoanContract(**data)这里有一个很容易踩的坑模型偶尔会返回带 Markdown 标记的 JSON比如json ... 必须做兼容处理。另外text[:3000]是为了防止单次传入内容过长在真实项目中可以配合分块抽取与汇总。4.7 合规规则校验模块文件路径src/compliance_rules.pyfrom .schemas import LoanContract, Finding class ComplianceRule: def __init__(self, code: str, severity: str, desc: str): self.code code self.severity severity self.desc desc def check(self, contract: LoanContract) - Finding | None: raise NotImplementedError class MaxAmountRule(ComplianceRule): def __init__(self, max_amount: float): super().__init__(AMOUNT_LIMIT, high, 贷款金额超上限) self.max_amount max_amount def check(self, contract: LoanContract) - Finding | None: if contract.loan_amount and contract.loan_amount self.max_amount: return Finding( rule_codeself.code, messageself.desc, evidencef贷款金额 {contract.loan_amount} 元超过上限 {self.max_amount} 元, severityself.severity, ) return None class MaxInterestRule(ComplianceRule): def __init__(self, max_rate: float): super().__init__(INTEREST_LIMIT, high, 利率超上限) self.max_rate max_rate def check(self, contract: LoanContract) - Finding | None: if contract.interest_rate and contract.interest_rate self.max_rate: return Finding( rule_codeself.code, messageself.desc, evidencef年化利率 {contract.interest_rate}超过上限 {self.max_rate}, severityself.severity, ) return None class RequiredFieldRule(ComplianceRule): def __init__(self): super().__init__(FIELD_MISSING, medium, 关键字段缺失) self.required_fields [ contract_no, customer_name, loan_amount, interest_rate, loan_term_months, ] def check(self, contract: LoanContract) - Finding | None: missing [f for f in self.required_fields if getattr(contract, f) is None] if missing: return Finding( rule_codeself.code, messageself.desc, evidencef缺失字段: {, .join(missing)}, severityself.severity, ) return None class SignatureStampRule(ComplianceRule): def __init__(self): super().__init__(SIGN_MISSING, high, 缺少签章) def check(self, contract: LoanContract) - Finding | None: if contract.has_signature is False or contract.has_stamp is False: return Finding( rule_codeself.code, messageself.desc, evidencefhas_signature{contract.has_signature}, has_stamp{contract.has_stamp}, severityself.severity, ) return None这种实现方式把每条规则做成一个类后续新增规则不需要改工作流主逻辑只需要在装配层注册。4.8 风险评分模块文件路径src/risk_scorer.pyfrom .schemas import AuditResult, Finding def calculate_risk_score(findings: list[Finding]) - int: 基于命中规则的严重程度计算风险分。 severity_weight { high: 40, medium: 20, low: 10, } score 0 for finding in findings: score severity_weight.get(finding.severity, 5) return min(score, 100) def determine_risk_level(score: int, high_threshold: int, review_threshold: int) - str: if score high_threshold: return high if score review_threshold: return medium return low风险评分可以采用更复杂的加权模型但起步阶段不需要太复杂。这个示例的核心思路是“规则命中越多、越严重风险分越高”规则命中片段会一起进入人工复核队列。4.9 工作流编排模块文件路径src/workflow.pyfrom pathlib import Path from .pdf_parser import extract_text_from_file from .llm_extractor import LLMExtractor from .compliance_rules import ( MaxAmountRule, MaxInterestRule, RequiredFieldRule, SignatureStampRule, ) from .risk_scorer import calculate_risk_score, determine_risk_level from .schemas import AuditResult from .storage import Storage class AuditWorkflow: def __init__(self, config: dict): self.config config self.extractor LLMExtractor(config) self.rules [ MaxAmountRule(config[compliance][max_loan_amount]), MaxInterestRule(config[compliance][max_interest_rate]), RequiredFieldRule(), SignatureStampRule(), ] self.storage Storage(config[storage][db_path]) def run_case(self, file_path: str) - AuditResult: case_id Path(file_path).stem text extract_text_from_file(file_path) contract self.extractor.extract(text) findings [] for rule in self.rules: finding rule.check(contract) if finding: findings.append(finding) score calculate_risk_score(findings) high_threshold self.config[compliance][high_risk_threshold] review_threshold self.config[compliance][review_threshold] level determine_risk_level(score, high_threshold, review_threshold) if score review_threshold: action review elif score high_threshold: action return else: action pass result AuditResult( case_idcase_id, contractcontract, findingsfindings, risk_scorescore, risk_levellevel, actionaction, ) self.storage.save_result(result) return result这个模块把前面的解析、抽取、规则、评分串起来是工作流的核心主干。4.10 状态存储模块文件路径src/storage.pyimport json import sqlite3 from pathlib import Path from .schemas import AuditResult class Storage: def __init__(self, db_path: str): Path(db_path).parent.mkdir(parentsTrue, exist_okTrue) self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS audit_result ( case_id TEXT PRIMARY KEY, result_json TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def save_result(self, result: AuditResult): self.conn.execute( INSERT OR REPLACE INTO audit_result (case_id, result_json) VALUES (?, ?), (result.case_id, result.model_dump_json()), ) self.conn.commit()存储层先把结果保存为 JSON后续如果要接人工复核系统可以在表里增加review_status、reviewer、review_comment等字段。4.11 命令行入口文件路径main.pyimport argparse import sys from pathlib import Path import yaml from src.workflow import AuditWorkflow def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): parser argparse.ArgumentParser(description内部审计 AI 工作流) parser.add_argument(--config, defaultconfig.yaml, help配置文件路径) parser.add_argument(--input, defaultdata/input, help待审材料目录) args parser.parse_args() config load_config(args.config) workflow AuditWorkflow(config) input_dir Path(args.input) pdf_files list(input_dir.glob(*.pdf)) if not pdf_files: print(没有找到 PDF 文件请放入 data/input 目录) sys.exit(1) for pdf_file in pdf_files: result workflow.run_case(str(pdf_file)) print( * 50) print(f案件: {result.case_id}) print(f风险分数: {result.risk_score}) print(f风险等级: {result.risk_level}) print(f建议动作: {result.action}) for finding in result.findings: print(f - [{finding.severity}] {finding.rule_code}: {finding.message}) print(f 证据: {finding.evidence}) if __name__ __main__: main()4.12 运行与验证把示例 PDF 放进data/input/目录后执行cd audit_workflow python main.py --config config.yaml --input data/input假设有一份合同文本中的贷款金额为 120 万、利率为 0.16预期输出类似 案件: loan_case_001 风险分数: 80 风险等级: high 建议动作: review - [high] AMOUNT_LIMIT: 贷款金额超上限 证据: 贷款金额 1200000 元超过上限 1000000 元 - [high] INTEREST_LIMIT: 利率超上限 证据: 年化利率 0.16超过上限 0.15 - [high] SIGN_MISSING: 缺少签章 证据: has_signatureFalse, has_stampFalse如果案件规则全部通过风险分数为 0建议动作就是pass。所有结果都会写入 SQLite 数据库中的audit_result表方便后续接入复核界面或报表系统。5. 常见问题与排查思路问题现象常见原因解决思路LLM 抽取字段不稳定提示词缺少示例、温度过高增加 few-shot 示例把 temperature 调到 0.1 以下PDF 提取文本为空文件是扫描件或图片型 PDF接入 OCR 能力先做文字识别再抽取模型返回 JSON 解析失败输出带 Markdown 标记或额外解释代码中兼容 json 包裹并记录原始输出便于排查规则误报率高规则阈值与实际业务不匹配先跑一批历史案件做校准根据误报率调整阈值长文档抽取效果差单次输入内容过长丢失上下文分段抽取并汇总或用 RAG 定位关键段落人工复核没有闭环工作流只输出结果没有状态流转在存储层增加 review_status、reviewer 字段下面重点展开两个高频问题。5.1 LLM 输出不稳定这是最常见的问题。模型偶尔会把 JSON 包在 Markdown 代码块里偶尔会漏字段偶尔会把利率 12% 直接写成12。解决思路不是强求模型“永远正确”而是从三个层面兜底提示词层明确输出格式给两个示例。代码层解析 JSON 失败时记录原始输出并支持重试一次。校验层Pydantic 模型对缺失字段做识别缺失时进入人工复核而不是静默失败。5.2 扫描件解析为空银行审计材料里大量存在扫描件、照片、盖章件。pdfplumber只能提取数字 PDF 的文本层对扫描件无能为力。如果遇到这种情况需要引入 OCR。生产环境建议调用公司内部 OCR 服务或用 PaddleOCR 做本地识别。在架构上你只需要修改pdf_parser.py把图片型 PDF 转成图片后走 OCR 流程。6. 最佳实践与工程建议6.1 数据安全与权限最小化银行数据安全是最高优先级。AI 工作流涉及的材料往往包含个人身份信息、信贷记录、企业财务数据在立项前必须通过内部数据安全评估。模型部署建议优先使用本地化或私有化环境避免敏感数据流向未授权外部服务。权限管理上要遵循最小权限原则。工作流服务只允许读取指定目录的文件不允许扫描整个文件系统数据库连接只授予INSERT、SELECT、UPDATE权限不给DELETE权限调用模型接口使用独立 API Key并设置调用限额和审计日志。6.2 审计留痕与可追溯性这里必须强调一点AI 输出不能直接作为审计结论只能作为“辅助发现”。每个发现项都要保留三样东西原始材料案件对应的 PDF 原文不能只存截断文本。证据片段规则命中时引用的文本上下文便于审计员对照原文。操作日志谁在什么时间运行了哪条流程输入输出是什么。建议在数据库表中增加created_at、operator_id、request_id等字段方便追溯。如果使用商业大模型 API要注意请求日志脱敏后再落库避免把完整身份证号、手机号写进日志。6.3 模型选择与评估不要盲目追求大参数模型。内部审计场景里字段抽取和规则命中的准确率比“文采”重要得多。建议先用同一批脱敏历史案件比较 2-3 个候选模型的表现。评估指标可以设置抽取字段准确率模型抽取结果与人工标注结果的一致程度。规则命中准确率模型抽取字段经过规则校验后是否与人工审计判断一致。人工复核通过率AI 标记为高风险并转入人工队列的案件中有多少被人工确认为真正风险。每次换模型或改提示词都要在这套评估集上回归一次避免“改了一个规则带崩另一个规则”。6.4 人审闭环设计人审是银行审计 AI 工作流里不可省略的一环。生产环境中不要只把结果打印在终端而是接入一个简单的复审界面或工单系统。审计员可以查看 AI 发现项、原始材料、证据片段并作出“确认风险”“误报”“需补充材料”三类判断。这些判断会回流到样本库成为后续优化模型和规则的训练数据。闭环数据的价值在于它能让系统越用越准。初期误报率高很正常关键是让每一条误报都被标记出来然后分析是规则阈值问题、抽取字段问题还是提示词问题。6.5 代码层面的工程建议所有配置项放配置文件不要硬编码在代码里。提示词单独管理方便业务人员调整避免重新发版。每个模块提供独立测试用例尤其是规则校验模块要做到“改一条规则只影响一条规则”。文件解析和 LLM 抽取失败时设置重试机制但重试次数不要超过 2 次。输出和日志中禁止打印完整身份证号、卡号、手机号必须做脱敏处理。7. 从原型到生产下一步怎么走前面这套示例已经可以跑通一条完整的 AI 审计工作流但离生产环境还有一段路。优先级比较高的几个方向是把 SQLite 换成 PostgreSQL增加字段状态和操作日志表支持多人并发审核。把命令行入口扩展成 FastAPI 服务方便内部系统调用和前端展示。在提示词之外引入 RAG把内部制度文件、历史审计底稿变成可检索的知识库帮助模型回答“某项业务是否符合某条制度”这类开放问题。对 Agent 能力做更严格的安全封装只允许调用白名单工具并对工具调用做全量日志审计。对一批历史案件做离线评估建立准召率基线再决定是否需要引入更大模型或调整工作流结构。从落地策略上看不建议一开始就做一个覆盖全行的“超级审计工作流”。更务实的做法是选一个标准化程度较高、材料格式相对统一的场景先试点比如“零售信贷合同抽样检查”跑通后再逐步扩展到对公业务、反洗钱、柜面凭证等领域。如果本文对你有帮助可以先收藏备用。后续你可以根据自己公司的模型接口、数据目录和内部制度把这套工作流改造成适合自己业务的版本。真正动手跑一遍比看十篇文章都管用。
做银行与金融机构的内部审计项目时最折磨人的往往不是算法选型而是流程本身。审计工作并不是“拿一个大模型问几个问题”那么简单它要完成材料收集、要素提取、政策比对、风险分级、底稿归档等一连串动作每一步都附带合规约束和留痕要求。传统规则脚本只能覆盖“能明确写死”的逻辑遇到合同语义、签章完整性、材料一致性这类模糊判断就不得不依赖人工逐份翻阅。最近我们在内部审计场景里搭建了一条 AI workflow把 PDF 解析、LLM 结构化抽取、规则校验、风险评分、人工复核完整串成了一条可追踪的流水线。这篇文章会把整套方案拆开来讲包括业务背景、工作流设计、完整可运行的 Python 代码、配置文件、常见坑点以及生产落地建议。无论你是金融行业的数据/风控/审计系统开发还是想在银行类场景落地 AI 应用都可以照着这套思路做一次最小验证。1. 为什么要为银行内部审计设计 AI 工作流1.1 内部审计到底在审什么银行内部审计与外部监管审计不一样它更像是银行内部的“体检系统”。审计人员会定期或不定期抽查业务材料验证流程是否规范、风险是否可控、操作是否违规。典型的内部审计场景包括对零售信贷业务做抽样检查核对贷款合同、收入证明、抵押材料是否完整合规。对公司客户授信材料做风险复核检查审批权限、授信额度、利率定价是否符合内部政策。对柜面业务凭证做一致性核验确认客户签名、证件信息、交易流水是否存在异常。对反洗钱可疑交易报告做复核判断上报依据是否充分。这些场景有一个共同特点材料多、字段多、规则细。审计员往往需要在短时间内阅读大量 PDF、图片、扫描件再对照内部制度一条条核对。这个过程既考验业务理解也考验耐心最容易出现的问题是漏看字段、规则理解不一致、底稿记录不完整。1.2 传统审计流程的局限传统方案主要有两条路。第一种是纯人工核对优点是判断灵活能处理语义和上下文但效率低、成本高而且不同审计员的判断标准可能不一致。第二种是写固定规则脚本比如用正则表达式提取合同编号、用 Excel 公式比对金额这种方式确定性强、执行快但只能覆盖“能提前枚举”的逻辑。实际审计材料往往比规则枚举复杂得多。比如一份收入证明上写的是“年薪 30 万”而贷款申请金额对应的月供压力明显超出合理区间这个判断需要模型理解语义又比如合同盖章位置偏淡、印章被遮挡传统 OCR 很难直接判断“是否存在有效签章”这又涉及图像特征和语义理解的结合。这些场景说明固定规则引擎不够用纯人工又累中间存在一个很大的智能自动化空间。1.3 AI Workflow 在这里扮演什么角色AI Workflow 并不是“调用一个 AI 接口就结束”而是把多个能力编排成一条可追踪、可重跑、可审计的业务流水线。以贷款材料合规性检查为例一条典型的 AI 工作流可以这样设计输入材料PDF/图片 ↓ 文档解析PDF 文本提取 / OCR ↓ 字段抽取LLM 抽取合同编号、客户名、金额、利率等 ↓ 规则校验金额上限、利率上限、材料完整性 ↓ 风险评分与分级 ↓ 人工复核队列 / 审计底稿归档每一步都有明确的输入输出任何一步失败都可以单独重试不会因为一个文件解析失败就导致整批案件停滞。更重要的是每一步都有日志和状态记录符合金融行业的留痕要求。1.4 AI Agent 与 Workflow 的边界最近 AI Agent 很火很多团队会直接把审计流程做成一个“什么都能干的 Agent”。但从银行审计角度看Agent 的自主性必须被限制。Agent 适合在有限范围内做动态规划比如根据风险等级选择不同的检查分支但审计场景中Agent 不能随意读取无关数据、不能修改底稿、不能绕过人工复核。更稳妥的做法是把 Agent 封装成工作流里的一个节点同时给它定义清楚工具边界。比如允许 Agent 调用“材料检索工具”“政策查询工具”但不允许它调用“删除案件工具”“直接修改评级工具”。每次工具调用都要记录参数与返回结果最终高风险判定必须经过人工确认。简单说在银行内部审计里AI 是“辅助放大器”不是“替代决策者”。2. 环境准备与项目结构2.1 技术栈选型为了兼顾可读性与可运行性本文的示例基于以下技术栈Python 3.10 及以上主要在 Linux 服务器上运行。FastAPI 作为可选的服务暴露层方便将工作流包装为内部接口。pdfplumber 用于提取 PDF 文本。Pydantic 用于定义结构化输出模型校验 LLM 抽取结果。OpenAPI 兼容客户端例如 OpenAI SDK调用大模型接口可切换到内部部署模型。SQLite 作为轻量状态存储生产环境可替换为 PostgreSQL。PyYAML 读取配置文件。如果你的公司内部大模型只提供了兼容 HTTP 接口也可以直接用requests替换 OpenAI SDK核心逻辑不变。2.2 示例项目结构建议先按分层思想组织代码不要把所有逻辑写在一个文件里。一个最小可运行的项目结构如下audit_workflow/ ├── config.yaml # 全局配置 ├── requirements.txt # 依赖列表 ├── main.py # 命令行入口 ├── src/ │ ├── __init__.py │ ├── pdf_parser.py # 文档解析模块 │ ├── schemas.py # Pydantic 数据模型 │ ├── llm_extractor.py # LLM 字段抽取模块 │ ├── compliance_rules.py # 规则校验模块 │ ├── risk_scorer.py # 风险评分模块 │ ├── workflow.py # 工作流编排核心 │ └── storage.py # 状态与结果存储 └── data/ ├── input/ # 放入待检查的 PDF └── output/ # 生成结果报告在真实项目中这个结构还可以继续拆出prompts/目录存放提示词模板、tests/目录存放单元测试、migrations/目录存放数据库变更脚本。这里先保持精简方便看清主体逻辑。2.3 依赖安装requirements.txt中建议不要锁死过细的版本先用范围约束避免因为某个小版本升级导致连锁冲突fastapi0.110,0.120 uvicorn0.29,0.40 pydantic2.6,3.0 pyyaml6.0,7.0 pdfplumber0.11,0.13 openai1.30,2.0 python-multipart0.0.9然后执行安装命令pip install -r requirements.txt如果只跑命令行示例不启动 FastAPI可以把fastapi、uvicorn、python-multipart去掉。不同公司内网环境差异较大建议实测后再统一版本。3. 核心原理工作流分层与关键设计3.1 工作流分层的价值把 AI 审计流程拆成分层不只是代码整洁的问题更是为了可控性。数据接入层负责把各种格式的文档转成统一文本信息抽取层把非结构化文本转成结构化字段规则校验层把结构化字段与制度规则做确定性比对风险分级层结合规则结果与模型置信度输出风险等级人工复核层保证关键结论有人兜底。这种分层让每一层都可以独立测试。文档解析结果差单独调 OCR 或解析参数字段抽不准只改提示词和 Pydantic 模型规则误报只改规则配置。如果没有分层所有逻辑混在一起出了问题几乎无法定位。3.2 为什么不能只用一个 Prompt有人会问“既然大模型能读文档能不能把全套材料丢给大模型让它直接输出一份审计结论” 理论上可以工程上不推荐。首先是上下文限制。一份贷款材料可能包含几十页合同、证明、银行流水全部塞进上下文既浪费 token也会降低抽取准确率。工作流可以先做分页抽取、关键段落定位再聚合结果。其次是稳定性。大模型输出天然有不确定性完整风险结论如果一次性生成很难校验对错。分层之后每一层都有结构化输出例如“抽取字段是否完整”“规则是否命中”这些中间结果可以追踪也方便人工复核。第三是审计要求。银行审计要求结论可解释、可追溯到原始材料。如果只给一个 Prompt 生成整体判断审计员无法快速定位“这个高风险评级到底是因为哪一条规则”产生的。分层之后结论可以关联到具体字段和规则比如“贷款金额 120 万超过内部上限 100 万”这样审计底稿才有说服力。3.3 人在环路Human-in-the-loopAI 工作流在银行落地的核心原则不是“全自动”而是“人机协同”。系统自动完成初筛和标注审计员把精力集中在高风险案件上。本文示例会在风险评分后增加一个人工复核队列。当风险分数超过阈值时案件状态变为 “PENDING_REVIEW”由审计员检查 AI 生成的发现项后确认或驳回。所有人工操作都要记录操作人、操作时间、操作内容最终形成闭环。3.4 提示词与结构化输出设计为了让 LLM 输出可被程序消费必须定义结构化输出格式。推荐用 JSON 输出再用 Pydantic 做校验。提示词里至少包含三部分任务描述说明当前要抽取哪些字段、字段的业务含义。输出格式明确要求返回 JSON并给出 JSON Schema。输出要求要求模型无法确定时输出null或unknown不要强行编造。示例提示词片段如下你是一个银行信贷材料审核助手。请从贷款合同中抽取以下字段 contract_no, customer_name, loan_amount, interest_rate, loan_term_months, signed_date, has_stamp, has_signature 要求 1. 输出 JSON不要输出其他解释文本。 2. 金额单位统一为元利率统一为小数。 3. 如果文档中无法找到某个字段请输出 null。给两个 few-shot 示例会让效果明显提升尤其是金额、日期这类格式容易混乱的字段。后续可以把提示词模板抽到独立目录方便运营同学调整。4. 完整实战贷款材料合规性 AI 检查工作流4.1 场景说明假设我们有若干份贷款申请材料 PDF需要完成以下检查是否包含合同、身份证、收入证明三类必要材料。合同中的客户姓名、身份证号、金额、利率、期限是否完整。贷款金额是否超过内部上限示例为 100 万。年化利率是否超过内部上限示例为 15%。贷款期限是否超过上限示例为 60 个月。合同是否包含签章。输出风险评分和建议动作自动通过 / 人工复核 / 退回补充材料。4.2 创建项目结构在任意目录执行mkdir -p audit_workflow/src audit_workflow/data/input audit_workflow/data/output然后把上一节的项目结构里的文件逐步创建出来。4.3 编写配置文件在audit_workflow/config.yaml中写入storage: input_dir: ./data/input output_dir: ./data/output db_path: ./data/audit_workflow.db model: base_url: http://localhost:8000/v1 api_key: sk-local-test model_name: auditor-llm temperature: 0.1 timeout: 60 compliance: max_loan_amount: 1000000 max_interest_rate: 0.15 max_loan_term_months: 60 high_risk_threshold: 80 review_threshold: 60 extraction: chunk_size: 1500配置项说明input_dir是待审材料目录每个 PDF 对应一个案件。db_path是 SQLite 数据库路径用于记录流程状态和结果。base_url和api_key是大模型服务地址与密钥。如果使用公司内部模型通常也是 OpenAI 兼容格式按内部文档替换即可。chunk_size用于处理超长文档时的文本分段避免上下文过长。4.4 定义结构化数据模型文件路径src/schemas.pyfrom pydantic import BaseModel, Field from typing import Optional class LoanContract(BaseModel): contract_no: Optional[str] Field(defaultNone, description合同编号) customer_name: Optional[str] Field(defaultNone, description客户姓名) id_card: Optional[str] Field(defaultNone, description身份证号) loan_amount: Optional[float] Field(defaultNone, description贷款金额元) interest_rate: Optional[float] Field(defaultNone, description年化利率小数) loan_term_months: Optional[int] Field(defaultNone, description贷款期限月) signed_date: Optional[str] Field(defaultNone, description签署日期) has_stamp: Optional[bool] Field(defaultNone, description是否有盖章) has_signature: Optional[bool] Field(defaultNone, description是否有签名) class Finding(BaseModel): rule_code: str Field(description命中的规则编码) message: str Field(description规则命中说明) evidence: str Field(description证据原文片段) severity: str Field(description风险级别high/medium/low) class AuditResult(BaseModel): case_id: str Field(description案件编号通常来自文件名) contract: LoanContract Field(description抽取的合同字段) findings: list[Finding] Field(default_factorylist, description规则发现项) risk_score: int Field(ge0, le100, description风险分数) risk_level: str Field(description风险等级high/medium/low) action: str Field(description建议动作pass/review/return)这里把抽取结果、规则命中、最终评级拆成三个模型有利于后续每个节点的独立测试和存储。4.5 文档解析模块文件路径src/pdf_parser.pyimport pdfplumber from pathlib import Path def extract_text_from_pdf(pdf_path: str) - str: 从 PDF 文件中提取文本内容。 text_parts [] with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: page_text page.extract_text() if page_text: text_parts.append(page_text) return \n.join(text_parts) def extract_text_from_file(file_path: str) - str: 根据文件后缀选择解析方式便于后续扩展 OCR 能力。 suffix Path(file_path).suffix.lower() if suffix .pdf: return extract_text_from_pdf(file_path) # 如果是图片或扫描件建议接入 OCR 服务本文示例只处理 PDF 文本 raise ValueError(f暂不支持的文件类型: {suffix})如果材料是扫描件pdfplumber提取不到文本此时需要接入 OCR。常见方案是调用内部 OCR 服务或使用 PaddleOCR 等开源工具。把解析逻辑独立成模块后续替换 OCR 就不会影响其他层。4.6 LLM 字段抽取模块文件路径src/llm_extractor.pyimport json import yaml from openai import OpenAI from .schemas import LoanContract PROMPT_TEMPLATE 你是一个银行信贷材料审核助手。 请从以下贷款合同文本中抽取字段只输出 JSON不要输出解释。 字段包括 contract_no合同编号 customer_name客户姓名 id_card身份证号 loan_amount贷款金额单位元 interest_rate年化利率小数形式例如 0.12 表示 12% loan_term_months贷款期限单位月 signed_date签署日期格式 YYYY-MM-DD has_stamp是否有盖章true/false has_signature是否有签名true/false 注意 1. 金额必须转为数值去掉单位。 2. 利率必须转为小数。 3. 找不到的字段输出 null。 合同文本 {text} class LLMExtractor: def __init__(self, config: dict): self.client OpenAI( base_urlconfig[model][base_url], api_keyconfig[model][api_key], timeoutconfig[model].get(timeout, 60), ) self.model_name config[model][model_name] self.temperature config[model].get(temperature, 0.1) def extract(self, text: str) - LoanContract: prompt PROMPT_TEMPLATE.format(texttext[:3000]) resp self.client.chat.completions.create( modelself.model_name, messages[{role: user, content: prompt}], temperatureself.temperature, ) content resp.choices[0].message.content.strip() # 兼容模型返回带 json 标记的情况 if content.startswith(): content content.strip() if content.startswith(json): content content[4:] data json.loads(content) return LoanContract(**data)这里有一个很容易踩的坑模型偶尔会返回带 Markdown 标记的 JSON比如json ... 必须做兼容处理。另外text[:3000]是为了防止单次传入内容过长在真实项目中可以配合分块抽取与汇总。4.7 合规规则校验模块文件路径src/compliance_rules.pyfrom .schemas import LoanContract, Finding class ComplianceRule: def __init__(self, code: str, severity: str, desc: str): self.code code self.severity severity self.desc desc def check(self, contract: LoanContract) - Finding | None: raise NotImplementedError class MaxAmountRule(ComplianceRule): def __init__(self, max_amount: float): super().__init__(AMOUNT_LIMIT, high, 贷款金额超上限) self.max_amount max_amount def check(self, contract: LoanContract) - Finding | None: if contract.loan_amount and contract.loan_amount self.max_amount: return Finding( rule_codeself.code, messageself.desc, evidencef贷款金额 {contract.loan_amount} 元超过上限 {self.max_amount} 元, severityself.severity, ) return None class MaxInterestRule(ComplianceRule): def __init__(self, max_rate: float): super().__init__(INTEREST_LIMIT, high, 利率超上限) self.max_rate max_rate def check(self, contract: LoanContract) - Finding | None: if contract.interest_rate and contract.interest_rate self.max_rate: return Finding( rule_codeself.code, messageself.desc, evidencef年化利率 {contract.interest_rate}超过上限 {self.max_rate}, severityself.severity, ) return None class RequiredFieldRule(ComplianceRule): def __init__(self): super().__init__(FIELD_MISSING, medium, 关键字段缺失) self.required_fields [ contract_no, customer_name, loan_amount, interest_rate, loan_term_months, ] def check(self, contract: LoanContract) - Finding | None: missing [f for f in self.required_fields if getattr(contract, f) is None] if missing: return Finding( rule_codeself.code, messageself.desc, evidencef缺失字段: {, .join(missing)}, severityself.severity, ) return None class SignatureStampRule(ComplianceRule): def __init__(self): super().__init__(SIGN_MISSING, high, 缺少签章) def check(self, contract: LoanContract) - Finding | None: if contract.has_signature is False or contract.has_stamp is False: return Finding( rule_codeself.code, messageself.desc, evidencefhas_signature{contract.has_signature}, has_stamp{contract.has_stamp}, severityself.severity, ) return None这种实现方式把每条规则做成一个类后续新增规则不需要改工作流主逻辑只需要在装配层注册。4.8 风险评分模块文件路径src/risk_scorer.pyfrom .schemas import AuditResult, Finding def calculate_risk_score(findings: list[Finding]) - int: 基于命中规则的严重程度计算风险分。 severity_weight { high: 40, medium: 20, low: 10, } score 0 for finding in findings: score severity_weight.get(finding.severity, 5) return min(score, 100) def determine_risk_level(score: int, high_threshold: int, review_threshold: int) - str: if score high_threshold: return high if score review_threshold: return medium return low风险评分可以采用更复杂的加权模型但起步阶段不需要太复杂。这个示例的核心思路是“规则命中越多、越严重风险分越高”规则命中片段会一起进入人工复核队列。4.9 工作流编排模块文件路径src/workflow.pyfrom pathlib import Path from .pdf_parser import extract_text_from_file from .llm_extractor import LLMExtractor from .compliance_rules import ( MaxAmountRule, MaxInterestRule, RequiredFieldRule, SignatureStampRule, ) from .risk_scorer import calculate_risk_score, determine_risk_level from .schemas import AuditResult from .storage import Storage class AuditWorkflow: def __init__(self, config: dict): self.config config self.extractor LLMExtractor(config) self.rules [ MaxAmountRule(config[compliance][max_loan_amount]), MaxInterestRule(config[compliance][max_interest_rate]), RequiredFieldRule(), SignatureStampRule(), ] self.storage Storage(config[storage][db_path]) def run_case(self, file_path: str) - AuditResult: case_id Path(file_path).stem text extract_text_from_file(file_path) contract self.extractor.extract(text) findings [] for rule in self.rules: finding rule.check(contract) if finding: findings.append(finding) score calculate_risk_score(findings) high_threshold self.config[compliance][high_risk_threshold] review_threshold self.config[compliance][review_threshold] level determine_risk_level(score, high_threshold, review_threshold) if score review_threshold: action review elif score high_threshold: action return else: action pass result AuditResult( case_idcase_id, contractcontract, findingsfindings, risk_scorescore, risk_levellevel, actionaction, ) self.storage.save_result(result) return result这个模块把前面的解析、抽取、规则、评分串起来是工作流的核心主干。4.10 状态存储模块文件路径src/storage.pyimport json import sqlite3 from pathlib import Path from .schemas import AuditResult class Storage: def __init__(self, db_path: str): Path(db_path).parent.mkdir(parentsTrue, exist_okTrue) self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS audit_result ( case_id TEXT PRIMARY KEY, result_json TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def save_result(self, result: AuditResult): self.conn.execute( INSERT OR REPLACE INTO audit_result (case_id, result_json) VALUES (?, ?), (result.case_id, result.model_dump_json()), ) self.conn.commit()存储层先把结果保存为 JSON后续如果要接人工复核系统可以在表里增加review_status、reviewer、review_comment等字段。4.11 命令行入口文件路径main.pyimport argparse import sys from pathlib import Path import yaml from src.workflow import AuditWorkflow def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): parser argparse.ArgumentParser(description内部审计 AI 工作流) parser.add_argument(--config, defaultconfig.yaml, help配置文件路径) parser.add_argument(--input, defaultdata/input, help待审材料目录) args parser.parse_args() config load_config(args.config) workflow AuditWorkflow(config) input_dir Path(args.input) pdf_files list(input_dir.glob(*.pdf)) if not pdf_files: print(没有找到 PDF 文件请放入 data/input 目录) sys.exit(1) for pdf_file in pdf_files: result workflow.run_case(str(pdf_file)) print( * 50) print(f案件: {result.case_id}) print(f风险分数: {result.risk_score}) print(f风险等级: {result.risk_level}) print(f建议动作: {result.action}) for finding in result.findings: print(f - [{finding.severity}] {finding.rule_code}: {finding.message}) print(f 证据: {finding.evidence}) if __name__ __main__: main()4.12 运行与验证把示例 PDF 放进data/input/目录后执行cd audit_workflow python main.py --config config.yaml --input data/input假设有一份合同文本中的贷款金额为 120 万、利率为 0.16预期输出类似 案件: loan_case_001 风险分数: 80 风险等级: high 建议动作: review - [high] AMOUNT_LIMIT: 贷款金额超上限 证据: 贷款金额 1200000 元超过上限 1000000 元 - [high] INTEREST_LIMIT: 利率超上限 证据: 年化利率 0.16超过上限 0.15 - [high] SIGN_MISSING: 缺少签章 证据: has_signatureFalse, has_stampFalse如果案件规则全部通过风险分数为 0建议动作就是pass。所有结果都会写入 SQLite 数据库中的audit_result表方便后续接入复核界面或报表系统。5. 常见问题与排查思路问题现象常见原因解决思路LLM 抽取字段不稳定提示词缺少示例、温度过高增加 few-shot 示例把 temperature 调到 0.1 以下PDF 提取文本为空文件是扫描件或图片型 PDF接入 OCR 能力先做文字识别再抽取模型返回 JSON 解析失败输出带 Markdown 标记或额外解释代码中兼容 json 包裹并记录原始输出便于排查规则误报率高规则阈值与实际业务不匹配先跑一批历史案件做校准根据误报率调整阈值长文档抽取效果差单次输入内容过长丢失上下文分段抽取并汇总或用 RAG 定位关键段落人工复核没有闭环工作流只输出结果没有状态流转在存储层增加 review_status、reviewer 字段下面重点展开两个高频问题。5.1 LLM 输出不稳定这是最常见的问题。模型偶尔会把 JSON 包在 Markdown 代码块里偶尔会漏字段偶尔会把利率 12% 直接写成12。解决思路不是强求模型“永远正确”而是从三个层面兜底提示词层明确输出格式给两个示例。代码层解析 JSON 失败时记录原始输出并支持重试一次。校验层Pydantic 模型对缺失字段做识别缺失时进入人工复核而不是静默失败。5.2 扫描件解析为空银行审计材料里大量存在扫描件、照片、盖章件。pdfplumber只能提取数字 PDF 的文本层对扫描件无能为力。如果遇到这种情况需要引入 OCR。生产环境建议调用公司内部 OCR 服务或用 PaddleOCR 做本地识别。在架构上你只需要修改pdf_parser.py把图片型 PDF 转成图片后走 OCR 流程。6. 最佳实践与工程建议6.1 数据安全与权限最小化银行数据安全是最高优先级。AI 工作流涉及的材料往往包含个人身份信息、信贷记录、企业财务数据在立项前必须通过内部数据安全评估。模型部署建议优先使用本地化或私有化环境避免敏感数据流向未授权外部服务。权限管理上要遵循最小权限原则。工作流服务只允许读取指定目录的文件不允许扫描整个文件系统数据库连接只授予INSERT、SELECT、UPDATE权限不给DELETE权限调用模型接口使用独立 API Key并设置调用限额和审计日志。6.2 审计留痕与可追溯性这里必须强调一点AI 输出不能直接作为审计结论只能作为“辅助发现”。每个发现项都要保留三样东西原始材料案件对应的 PDF 原文不能只存截断文本。证据片段规则命中时引用的文本上下文便于审计员对照原文。操作日志谁在什么时间运行了哪条流程输入输出是什么。建议在数据库表中增加created_at、operator_id、request_id等字段方便追溯。如果使用商业大模型 API要注意请求日志脱敏后再落库避免把完整身份证号、手机号写进日志。6.3 模型选择与评估不要盲目追求大参数模型。内部审计场景里字段抽取和规则命中的准确率比“文采”重要得多。建议先用同一批脱敏历史案件比较 2-3 个候选模型的表现。评估指标可以设置抽取字段准确率模型抽取结果与人工标注结果的一致程度。规则命中准确率模型抽取字段经过规则校验后是否与人工审计判断一致。人工复核通过率AI 标记为高风险并转入人工队列的案件中有多少被人工确认为真正风险。每次换模型或改提示词都要在这套评估集上回归一次避免“改了一个规则带崩另一个规则”。6.4 人审闭环设计人审是银行审计 AI 工作流里不可省略的一环。生产环境中不要只把结果打印在终端而是接入一个简单的复审界面或工单系统。审计员可以查看 AI 发现项、原始材料、证据片段并作出“确认风险”“误报”“需补充材料”三类判断。这些判断会回流到样本库成为后续优化模型和规则的训练数据。闭环数据的价值在于它能让系统越用越准。初期误报率高很正常关键是让每一条误报都被标记出来然后分析是规则阈值问题、抽取字段问题还是提示词问题。6.5 代码层面的工程建议所有配置项放配置文件不要硬编码在代码里。提示词单独管理方便业务人员调整避免重新发版。每个模块提供独立测试用例尤其是规则校验模块要做到“改一条规则只影响一条规则”。文件解析和 LLM 抽取失败时设置重试机制但重试次数不要超过 2 次。输出和日志中禁止打印完整身份证号、卡号、手机号必须做脱敏处理。7. 从原型到生产下一步怎么走前面这套示例已经可以跑通一条完整的 AI 审计工作流但离生产环境还有一段路。优先级比较高的几个方向是把 SQLite 换成 PostgreSQL增加字段状态和操作日志表支持多人并发审核。把命令行入口扩展成 FastAPI 服务方便内部系统调用和前端展示。在提示词之外引入 RAG把内部制度文件、历史审计底稿变成可检索的知识库帮助模型回答“某项业务是否符合某条制度”这类开放问题。对 Agent 能力做更严格的安全封装只允许调用白名单工具并对工具调用做全量日志审计。对一批历史案件做离线评估建立准召率基线再决定是否需要引入更大模型或调整工作流结构。从落地策略上看不建议一开始就做一个覆盖全行的“超级审计工作流”。更务实的做法是选一个标准化程度较高、材料格式相对统一的场景先试点比如“零售信贷合同抽样检查”跑通后再逐步扩展到对公业务、反洗钱、柜面凭证等领域。如果本文对你有帮助可以先收藏备用。后续你可以根据自己公司的模型接口、数据目录和内部制度把这套工作流改造成适合自己业务的版本。真正动手跑一遍比看十篇文章都管用。
![[光学原理与应用-521]：对光的错误理解与纠偏](http://pic.xiahunao.cn/yaotu/[光学原理与应用-521]：对光的错误理解与纠偏)
