财务审计 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）
 - 证据链中断 = 审计失败，必须重新采集

十、关键设计决策

1. 为什么用多 Agent 而不是单一 Agent？ 审计的四个环节（采集/检查/检测/报告）的专业知识完全不同。单一 Agent 的 System Prompt 会过长导致能力稀释。多 Agent 设计让每个角色专注自己的领域，Orchestrator 负责协调和决策。

2. 为什么 Benford 检验不能单独作为审计证据？ Benford 定律只是统计指标，偏离不等于舞弊（某些业务天然不符合 Benford 分布）。它的作用是"指方向"——告诉审计师哪些数据值得深入检查，而不是直接下结论。

3. 为什么审计意见必须设置 HITL？ 审计意见是法律文书，CPA 需要对其承担法律责任。Agent 可以准备所有材料、生成初稿、给出建议，但最终签字权必须在人。这不是技术限制，而是制度约束。

4. 为什么数据快照要做 hash 锁定？ 审计证据的核心属性是"不可篡改"。如果被审计方在审计期间修改了数据，而 Agent 没有发现，整个审计结论就不可信。hash 锁定确保采集时点的数据与最终引用的数据完全一致。

5. 为什么规则引擎和异常检测要并行？ 规则引擎只能发现"已知的风险"（违反明确条款的行为）。异常检测发现"未知的风险"（统计上不正常但没有违反具体条款的行为）。两者互补，缺一不可。

Maurice | [email protected]