财务审计 Agent 架构
原创
灵阙教研团队
S 精选 提升 |
约 10 分钟阅读
更新于 2026-02-27 AI 导读
财务审计 Agent 架构 从数据采集到报告生成：构建合规驱动的多智能体审计系统 一、审计自动化的本质 财务审计的核心不是"找问题"，而是"有证据地证明没问题"。这一根本特征决定了审计 Agent 的设计必须以证据链为中心：每一个判断都必须可追溯，每一个结论都必须有依据，每一个异常都必须有处置记录。 二、五阶段审计工作流 审计 Agent 五阶段工作流...
财务审计 Agent 架构
从数据采集到报告生成：构建合规驱动的多智能体审计系统
一、审计自动化的本质
财务审计的核心不是"找问题"，而是"有证据地证明没问题"。这一根本特征决定了审计 Agent 的设计必须以证据链为中心：每一个判断都必须可追溯，每一个结论都必须有依据，每一个异常都必须有处置记录。
二、五阶段审计工作流
审计 Agent 五阶段工作流
================================================================
Stage 1: 数据采集
+-----------+ +-----------+ +-----------+
| ERP 系统 | | 银行流水 | | 发票系统 |
+-----------+ +-----------+ +-----------+
| | |
+--------+-------+--------+-------+
|
v
+---------------------------+
| Stage 2: 风险评估 | ← 识别高风险领域
+---------------------------+
|
v
+---------------------------+
| Stage 3: 抽样策略 | ← 分层抽样 / 货币单位抽样
+---------------------------+
|
v
+---------------------------+
| Stage 4: 实质性测试 | ← 规则引擎 + 异常检测
+---------------------------+
|
v
+---------------------------+
| Stage 5: 报告生成 | ← 结构化审计报告
+---------------------------+
三、多智能体协作设计
单一 Agent 无法胜任审计的全部职责。我们设计四个专职 Agent，各司其职，通过 Orchestrator 协调。
多智能体协作架构
=============================================
+------------------+
| Orchestrator |
| (审计主控) |
+------------------+
| | | |
+--------+ | | +--------+
| | | |
v v v v
+-----------+ +----------+ +----------+ +-----------+
| Data | | Rule | | Anomaly | | Report |
| Extractor | | Checker | | Detector | | Generator |
| 数据采集 | | 规则检查 | | 异常检测 | | 报告生成 |
+-----------+ +----------+ +----------+ +-----------+
| | | |
v v v v
原始数据 合规结果 异常清单 审计报告
证据包 测试工作底稿 风险评级 管理层建议
from dataclasses import dataclass, field
from enum import Enum
from datetime import date
from typing import Any
class AuditPhase(Enum):
DATA_COLLECTION = "data_collection"
RISK_ASSESSMENT = "risk_assessment"
SAMPLING = "sampling"
SUBSTANTIVE_TESTING = "substantive_testing"
REPORTING = "reporting"
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class FindingSeverity(Enum):
OBSERVATION = "observation" # 一般观察
DEFICIENCY = "deficiency" # 内控缺陷
SIGNIFICANT = "significant" # 重大缺陷
MATERIAL_WEAKNESS = "material" # 重大薄弱环节
@dataclass
class AuditEvidence:
"""审计证据：每一条发现都必须附带证据。"""
evidence_id: str
source: str # 来源系统
document_ref: str # 原始单据编号
description: str
collected_at: str
collected_by: str # agent_id 或 auditor_name
data_snapshot: dict = field(default_factory=dict)
@dataclass
class AuditFinding:
"""审计发现：问题 + 证据 + 建议的完整链条。"""
finding_id: str
severity: FindingSeverity
category: str # revenue_recognition / expense_approval / ...
title: str
description: str
standard_ref: str # 准则引用 (e.g., CAS 14, CAS 30)
evidence: list[AuditEvidence]
recommendation: str
management_response: str = "" # 管理层回复（人工填写）
risk_level: RiskLevel = RiskLevel.MEDIUM
hitl_required: bool = False # 是否需要人工复核
四、数据采集 Agent
数据采集是审计的基础。Agent 需要从多个异构系统中提取数据，并确保数据完整性和一致性。
from abc import ABC, abstractmethod
class DataExtractor(ABC):
"""数据采集器基类：所有数据源实现统一接口。"""
@abstractmethod
async def extract(self, params: dict) -> dict:
"""提取数据，返回标准化结果。"""
...
@abstractmethod
async def validate_completeness(self, data: dict) -> tuple[bool, str]:
"""验证数据完整性。"""
...
class ERPExtractor(DataExtractor):
"""ERP 系统数据采集：总账、应收、应付、固定资产。"""
async def extract(self, params: dict) -> dict:
period_start = params["period_start"]
period_end = params["period_end"]
modules = params.get("modules", ["gl", "ar", "ap", "fa"])
result = {}
for module in modules:
# 实际实现中通过 ERP API 或数据库直连
result[module] = await self._fetch_module_data(
module, period_start, period_end
)
return result
async def validate_completeness(self, data: dict) -> tuple[bool, str]:
"""检查数据完整性：期间连续性、借贷平衡。"""
gl_data = data.get("gl", {})
total_debit = sum(e.get("debit", 0) for e in gl_data.get("entries", []))
total_credit = sum(e.get("credit", 0) for e in gl_data.get("entries", []))
diff = abs(total_debit - total_credit)
if diff > 0.01: # 允许分位误差
return False, f"trial_balance_mismatch: debit={total_debit}, credit={total_credit}, diff={diff}"
return True, "balanced"
async def _fetch_module_data(self, module: str, start: str, end: str) -> dict:
# 实际实现：数据库查询或 API 调用
return {"module": module, "period": f"{start}~{end}", "entries": []}
五、规则检查 Agent（合规规则引擎）
规则引擎是审计 Agent 的"法规大脑"。我们将中国会计准则（CAS）编码为可执行的检查规则。
@dataclass
class ComplianceRule:
"""合规检查规则：将准则条文编码为可执行逻辑。"""
rule_id: str
standard: str # CAS 14 / CAS 30 / CAS 22 ...
article: str # 具体条款
description: str
check_fn_name: str # 检查函数名
severity: FindingSeverity
parameters: dict = field(default_factory=dict)
# 规则注册表：将中国会计准则映射为可执行规则
COMPLIANCE_RULES: list[ComplianceRule] = [
ComplianceRule(
rule_id="CAS14-R001",
standard="CAS 14 (收入)",
article="第十三条",
description="收入确认须满足五步法：识别合同、识别履约义务、确定交易价格、分摊交易价格、确认收入",
check_fn_name="check_revenue_recognition",
severity=FindingSeverity.SIGNIFICANT,
),
ComplianceRule(
rule_id="CAS30-R001",
standard="CAS 30 (财务报表列报)",
article="第六条",
description="资产和负债不得相互抵销列报，除非其他准则允许或要求",
check_fn_name="check_no_offset",
severity=FindingSeverity.DEFICIENCY,
),
ComplianceRule(
rule_id="CAS22-R001",
standard="CAS 22 (金融工具确认和计量)",
article="第二十二条",
description="金融资产减值须采用预期信用损失模型",
check_fn_name="check_ecl_model",
severity=FindingSeverity.SIGNIFICANT,
),
ComplianceRule(
rule_id="INTERNAL-R001",
standard="内部控制",
article="审批流程",
description="超过5万元的费用支出须经两级审批",
check_fn_name="check_approval_chain",
severity=FindingSeverity.DEFICIENCY,
parameters={"threshold": 50000, "min_approvers": 2},
),
]
class RuleChecker:
"""规则检查引擎：逐条执行合规规则。"""
def __init__(self, rules: list[ComplianceRule]):
self.rules = rules
async def run_all(self, data: dict) -> list[AuditFinding]:
"""执行所有规则检查，返回发现列表。"""
findings = []
for rule in self.rules:
check_fn = getattr(self, rule.check_fn_name, None)
if check_fn is None:
continue
result = await check_fn(data, rule)
if result:
findings.extend(result)
return findings
async def check_approval_chain(
self, data: dict, rule: ComplianceRule
) -> list[AuditFinding]:
"""检查费用审批链：金额超阈值必须有足够审批层级。"""
threshold = rule.parameters.get("threshold", 50000)
min_approvers = rule.parameters.get("min_approvers", 2)
findings = []
expenses = data.get("ap", {}).get("entries", [])
for entry in expenses:
amount = entry.get("amount", 0)
approvers = entry.get("approvers", [])
if amount > threshold and len(approvers) < min_approvers:
findings.append(AuditFinding(
finding_id=f"F-{rule.rule_id}-{entry.get('voucher_no', 'N/A')}",
severity=rule.severity,
category="internal_control",
title=f"费用审批层级不足: {entry.get('voucher_no')}",
description=(
f"凭证 {entry.get('voucher_no')} 金额 {amount:,.2f} 元"
f"超过阈值 {threshold:,.2f} 元，"
f"但仅有 {len(approvers)} 级审批"
f"（要求至少 {min_approvers} 级）"
),
standard_ref=f"{rule.standard} {rule.article}",
evidence=[AuditEvidence(
evidence_id=f"E-{entry.get('voucher_no')}",
source="ERP-AP",
document_ref=entry.get("voucher_no", ""),
description="原始费用凭证",
collected_at=str(date.today()),
collected_by="rule_checker_agent",
data_snapshot=entry,
)],
recommendation=f"补充审批至 {min_approvers} 级后重新入账",
hitl_required=True,
))
return findings
六、异常检测 Agent
规则引擎覆盖"已知的风险"，异常检测覆盖"未知的风险"。
异常检测策略矩阵
=============================================
策略 适用场景 原理
----------- --------------------- ----------------
Benford 定律 费用报销金额分布 首位数字应符合对数分布
时间异常 月末/年末集中入账 日期聚集度偏离正常模式
金额聚类 卡审批阈值的金额 刚好低于审批线的异常聚集
关联方检测 供应商/客户交叉关系 图分析发现隐性关联
趋势偏离 收入/费用月度波动 超出历史均值 +/- 2 sigma
重复检测 重复支付/重复报销 金额+日期+对象三重匹配
import math
from collections import Counter
def benford_test(amounts: list[float], significance: float = 0.05) -> dict:
"""Benford 定律检验：检测金额首位数字分布异常。
正常的财务数据中，首位数字为 1 的概率约 30.1%，
为 9 的概率约 4.6%。偏离此分布可能暗示数据造假。
"""
expected = {d: math.log10(1 + 1 / d) for d in range(1, 10)}
first_digits = []
for amt in amounts:
if amt > 0:
first_digit = int(str(abs(amt)).lstrip("0").lstrip(".")[0])
if 1 <= first_digit <= 9:
first_digits.append(first_digit)
if len(first_digits) < 50:
return {"status": "insufficient_data", "sample_size": len(first_digits)}
observed = Counter(first_digits)
total = len(first_digits)
chi_squared = 0.0
digit_analysis = {}
for d in range(1, 10):
obs_pct = observed.get(d, 0) / total
exp_pct = expected[d]
deviation = obs_pct - exp_pct
chi_squared += (deviation ** 2) / exp_pct
digit_analysis[d] = {
"observed_pct": round(obs_pct * 100, 1),
"expected_pct": round(exp_pct * 100, 1),
"deviation_pct": round(deviation * 100, 1),
}
# 自由度 = 8 (9个数字 - 1), 临界值 alpha=0.05 约 15.51
is_anomalous = chi_squared > 15.51
return {
"status": "anomalous" if is_anomalous else "normal",
"chi_squared": round(chi_squared, 3),
"critical_value": 15.51,
"sample_size": total,
"digit_analysis": digit_analysis,
}
七、Human-in-the-Loop 门禁
审计 Agent 的每一个关键决策点都必须设置人工复核门禁。自动化提升效率，人工保证权威性。
HITL 门禁分布
=============================================
阶段 门禁点 触发条件
--------- ------------------ ----------------
数据采集 数据完整性确认 采集完成后
风险评估 风险等级确认 评估完成后
抽样 样本量 & 范围确认 抽样方案生成后
实质性测试 重大发现复核 severity >= SIGNIFICANT
报告生成 报告终审 报告生成后（强制）
规则:
- severity = CRITICAL/SIGNIFICANT: 强制人工复核
- severity = DEFICIENCY: Agent 自动处理，批量人工抽检
- severity = OBSERVATION: Agent 自动处理，记录备查
八、审计报告生成
# System Prompt: 审计报告生成 Agent
# 角色
你是一名注册会计师，负责根据审计发现生成规范的审计报告。
# 报告结构
1. 审计概况（被审计单位、审计期间、审计范围）
2. 审计依据（中国注册会计师审计准则、CAS 相关条款）
3. 主要发现（按严重级别排序，每条附证据引用）
4. 内部控制评价（设计有效性 + 运行有效性）
5. 审计意见（无保留/保留/否定/无法表示）
6. 管理层建议（可操作的改进措施）
# 写作规范
- 用语必须符合审计报告规范（参照 ISA 700/CAS 1501）
- 每条发现必须引用具体准则条款
- 金额使用千元单位，保留两位小数
- 风险描述必须量化（影响金额/比例/频次）
- 建议必须具体可执行，避免"加强管理"等空话
# 审计意见判断标准
- 所有发现 severity <= DEFICIENCY: 无保留意见
- 存在 SIGNIFICANT 但已纠正: 无保留意见（带强调事项段）
- 存在未纠正的 SIGNIFICANT: 保留意见
- 存在 MATERIAL_WEAKNESS: 否定意见 (需要 HITL 确认)
九、证据链与审计轨迹
证据链完整性
=============================================
原始数据（ERP/银行/发票）
|
v
数据快照（采集时间戳 + hash）
|
v
规则检查结果（规则ID + 输入数据 + 判定逻辑）
|
v
异常检测结果（算法 + 参数 + 统计量）
|
v
审计发现（finding_id -> evidence_id 映射）
|
v
人工复核记录（审计师签名 + 时间 + 意见）
|
v
审计报告（所有发现的汇总 + 意见）
要求:
- 每一步都必须有时间戳和操作者（Agent 或人）
- 数据快照不可变，采集后 hash 锁定
- 所有 Agent 决策必须记录推理过程（reasoning trace）
- 证据链中断 = 审计失败，必须重新采集
十、关键设计决策
为什么用多 Agent 而不是单一 Agent？ 审计的四个环节（采集/检查/检测/报告）的专业知识完全不同。单一 Agent 的 System Prompt 会过长导致能力稀释。多 Agent 设计让每个角色专注自己的领域，Orchestrator 负责协调和决策。
为什么 Benford 检验不能单独作为审计证据？ Benford 定律只是统计指标，偏离不等于舞弊（某些业务天然不符合 Benford 分布）。它的作用是"指方向"——告诉审计师哪些数据值得深入检查，而不是直接下结论。
为什么审计意见必须设置 HITL？ 审计意见是法律文书，CPA 需要对其承担法律责任。Agent 可以准备所有材料、生成初稿、给出建议，但最终签字权必须在人。这不是技术限制，而是制度约束。
为什么数据快照要做 hash 锁定？ 审计证据的核心属性是"不可篡改"。如果被审计方在审计期间修改了数据，而 Agent 没有发现，整个审计结论就不可信。hash 锁定确保采集时点的数据与最终引用的数据完全一致。
为什么规则引擎和异常检测要并行？ 规则引擎只能发现"已知的风险"（违反明确条款的行为）。异常检测发现"未知的风险"（统计上不正常但没有违反具体条款的行为）。两者互补，缺一不可。
Maurice | maurice_wen@proton.me