如何構建具備審計軌跡、存取控制及人手監督的金融 AI 智能體

Here’s a truth about building AI agents for finance that nobody puts in the tutorial: the agent intelligence is maybe 30 percent of the work. The audit trail, the access controls, the human oversight checkpoints, the compliance documentation, that’s the other 70 percent. And that 70 percent is what determines whether your agent runs in production or gets killed in the compliance review before it processes a single transaction.

We learned this the expensive way on our first finance AI agent deployment. Built a beautiful underwriting agent. Elegant reasoning. Clean tool use. Impressive accuracy on test data. The compliance team looked at it for two hours and rejected it because they couldn’t answer one question: “If a regulator asks why the agent recommended denying this loan application, can you show them the exact data and reasoning that produced that recommendation?”

We couldn’t. The agent worked. It just didn’t prove it worked. And in finance, working without proving it is the same as not working.

This article is the guide we wish existed when we started. It covers the audit trail architecture, the access control design, the human oversight implementation, and the code patterns that make a finance AI agent not just functional but auditable, controllable, and compliant.

If you’re building agents that touch money, customer data, or regulatory workflows, this is the engineering that separates a demo from a deployment.

The audit trail isn’t logging. It’s the primary deliverable

Most developers think of audit trails as application logging. Write some events to a log file. Query them when something goes wrong. That’s not what finance compliance requires.

A finance audit trail must capture what data the agent received, what reasoning it applied, what tools it called, what each tool returned, what decision it reached, how confident it was, and what action it took, for every single transaction. And it must store this immutably for seven to ten years in a format that a non-technical regulator can review.

Here’s the data model:

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
import hashlib
import json


class EvaluationType(Enum):
    DETERMINISTIC = "deterministic"
    AI_ASSISTED = "ai_assisted"
    HUMAN_REVIEWED = "human_reviewed"


class ActionType(Enum):
    AUTO_APPROVED = "auto_approved"
    AUTO_DENIED = "auto_denied"
    ESCALATED = "escalated"
    PENDING_REVIEW = "pending_review"


@dataclass
class AuditRecord:
    """Immutable record of every agent decision."""
    record_id: str
    timestamp: datetime
    agent_id: str
    agent_version: str

    # What was evaluated
    transaction_id: str
    transaction_type: str
    input_data_hash: str          # SHA-256 of exact input
    input_data_snapshot: dict      # Frozen copy of input state

    # How it was evaluated
    evaluation_type: EvaluationType
    rules_applied: list[str]       # Rule IDs for deterministic checks
    model_version: Optional[str]   # For AI-assisted evaluations
    prompt_hash: Optional[str]     # For reproducibility

    # What was concluded
    decision: ActionType
    confidence: Optional[float]
    reasoning_chain: list[dict]    # Step-by-step reasoning

    # What happened next
    escalated_to: Optional[str]
    human_reviewer: Optional[str]
    human_decision: Optional[str]
    human_override_reason: Optional[str]
    human_decision_timestamp: Optional[datetime]

    # Integrity
    previous_record_hash: Optional[str]  # Chain integrity
    record_hash: str = field(init=False)

    def __post_init__(self):
        """Compute hash for tamper detection."""
        content = json.dumps({
            'record_id': self.record_id,
            'transaction_id': self.transaction_id,
            'input_data_hash': self.input_data_hash,
            'decision': self.decision.value,
            'timestamp': self.timestamp.isoformat(),
            'previous_hash': self.previous_record_hash
        }, sort_keys=True)
        self.record_hash = hashlib.sha256(content.encode()).hexdigest()

Two fields in this model deserve specific attention.

input_data_hash and input_data_snapshot serve different purposes. The hash proves that the data hasn’t been modified since the decision was made. The snapshot preserves the exact data state at decision time, because the source data in the ERP may change after the agent evaluated it, and the regulator needs to see what the agent saw, not what the database currently contains.

previous_record_hash creates a chain where tampering with any historical record breaks the hash chain on every subsequent record. This isn’t blockchain, it’s a simple integrity mechanism that makes after-the-fact modification detectable.

The storage architecture matters:

class AuditStore:
    """Append-only audit storage with hot and cold tiers."""

    def __init__(self, hot_store, cold_store, retention_years=7):
        self.hot = hot_store      # PostgreSQL - last 90 days, fast queries
        self.cold = cold_store    # S3 + Parquet - long-term, cost-efficient
        self.retention = retention_years

    async def append(self, record: AuditRecord) -> str:
        """Write to hot store. Never update. Never delete."""
        # Verify chain integrity
        last_record = await self.hot.get_latest(record.agent_id)
        if last_record and record.previous_record_hash != last_record.record_hash:
            raise IntegrityError("Chain broken - previous hash mismatch")

        await self.hot.insert(record)
        return record.record_hash

    async def archive(self, older_than_days=90):
        """Move aged records to cold storage. Hot store stays fast."""
        records = await self.hot.get_older_than(older_than_days)
        await self.cold.write_parquet(records)
        # Mark as archived in hot store - never delete
        await self.hot.mark_archived(records)

The append-only constraint is non-negotiable. No updates. No deletes. No “soft deletes.” If a record needs correction, you append a correction record that references the original. The original stays forever.

The access control layer: least privilege isn’t optional

The agent’s service account should have exactly the permissions it needs and nothing more. This sounds obvious. In practice, it’s the control that most teams get wrong because scoping permissions precisely requires understanding every data access and system interaction the agent performs.

class AgentPermissions:
    """Define and enforce what the agent can access and do."""

    def __init__(self, agent_role: str):
        self.role = agent_role
        self.permissions = self._load_permissions(agent_role)

    def _load_permissions(self, role: str) -> dict:
        """Each agent role has explicit read/write scopes."""
        roles = {
            "ap_processor": {
                "read": [
                    "invoices",
                    "purchase_orders", 
                    "goods_receipts",
                    "vendor_master",        # Read vendor info
                    "approval_matrix",      # Read routing rules
                ],
                "write": [
                    "invoices.status",       # Update processing status
                    "invoices.match_result", # Record match outcome
                    "approval_queue",        # Route for approval
                ],
                "blocked": [
                    "payments.execute",     # Cannot trigger payments
                    "vendor_master.modify", # Cannot change vendor data
                    "general_ledger",       # No GL access
                    "payroll",              # No payroll access
                    "treasury",             # No treasury access
                ],
                "amount_ceiling": 100_000,  # Escalate above this
            },
            "underwriting_analyst": {
                "read": [
                    "loan_applications",
                    "credit_bureau_data",
                    "financial_documents",
                    "compliance_rules",
                ],
                "write": [
                    "applications.risk_assessment",
                    "applications.recommendation",
                    "escalation_queue",
                ],
                "blocked": [
                    "applications.approve",  # Cannot approve loans
                    "applications.deny",     # Cannot deny loans
                    "disbursement",          # No access to funds
                ],
                "amount_ceiling": None,  # All amounts escalate
            }
        }
        return roles.get(role, {"read": [], "write": [], "blocked": []})

    def check_access(self, operation: str, resource: str) -> bool:
        """Enforce permission before every system interaction."""
        if resource in self.permissions.get("blocked", []):
            raise PermissionDenied(
                f"Agent role '{self.role}' is blocked from '{resource}'"
            )

        allowed_resources = self.permissions.get(operation, [])
        if resource not in allowed_resources:
            raise PermissionDenied(
                f"Agent role '{self.role}' lacks '{operation}' on '{resource}'"
            )
        return True

The blocked list is as important as the read and write lists. Explicitly defining what the agent cannot access, even when those resources exist in the same system, creates a documented boundary that compliance teams can review. An AP processing agent that has no path to the payroll system, the treasury system, or the general ledger is an agent whose blast radius is bounded by design.

The amount_ceiling field is the operational authority limit. Transactions above this threshold escalate to a human regardless of how confident the agent is. The ceiling is a business decision, not a technical one, set by the finance leadership based on the organisation’s risk tolerance.

Human oversight: the checkpoint architecture

Human-in-the-loop for finance agents isn’t “a human can review if they want.” It’s a set of defined checkpoints where the agent must stop, present its work, and receive human authorisation before proceeding.

class OversightCheckpoint:
    """Defines when and how the agent pauses for human review."""

    def __init__(self, rules: list[dict]):
        self.rules = rules

    async def evaluate(
        self, 
        transaction: dict,
        agent_assessment: dict,
        audit_trail: list[AuditRecord]
    ) -> CheckpointDecision:
        """Determine if human review is required."""

        for rule in self.rules:
            if self._matches(rule, transaction, agent_assessment):
                return CheckpointDecision(
                    requires_human=True,
                    reason=rule["reason"],
                    reviewer_role=rule["reviewer_role"],
                    context=self._assemble_context(
                        transaction, agent_assessment, audit_trail
                    ),
                    sla_hours=rule.get("sla_hours", 24)
                )

        return CheckpointDecision(requires_human=False)

    def _matches(self, rule, transaction, assessment) -> bool:
        """Check if this transaction triggers a checkpoint."""
        checks = {
            "amount_exceeds": lambda: (
                transaction.get("amount", 0) > rule.get("threshold", 0)
            ),
            "confidence_below": lambda: (
                assessment.get("confidence", 1.0) < rule.get("threshold", 0.85)
            ),
            "new_counterparty": lambda: (
                transaction.get("counterparty_history_days", 999) < 
                rule.get("minimum_days", 90)
            ),
            "exception_type": lambda: (
                assessment.get("exception_type") in 
                rule.get("escalation_types", [])
            ),
            "regulatory_flag": lambda: (
                assessment.get("regulatory_flags", []) != []
            ),
        }

        check_fn = checks.get(rule["condition"])
        return check_fn() if check_fn else False

    def _assemble_context(self, transaction, assessment, trail) -> dict:
        """Give the human everything they need to make a decision."""
        return {
            "transaction_summary": transaction,
            "agent_recommendation": assessment.get("recommendation"),
            "agent_confidence": assessment.get("confidence"),
            "agent_reasoning": assessment.get("reasoning_chain"),
            "relevant_history": self._get_relevant_history(transaction),
            "applicable_rules": assessment.get("rules_checked"),
            "escalation_reason": assessment.get("escalation_reason"),
            "suggested_action": assessment.get("suggested_action"),
            "audit_trail": [r.record_id for r in trail[-5:]],  # Last 5 steps
        }


# Checkpoint rules configuration
CHECKPOINT_RULES = [
    {
        "condition": "amount_exceeds",
        "threshold": 100_000,
        "reason": "Transaction exceeds autonomous authority limit",
        "reviewer_role": "senior_finance",
        "sla_hours": 4
    },
    {
        "condition": "confidence_below",
        "threshold": 0.85,
        "reason": "Agent confidence below threshold",
        "reviewer_role": "domain_specialist",
        "sla_hours": 8
    },
    {
        "condition": "new_counterparty",
        "minimum_days": 90,
        "reason": "Counterparty has less than 90 days of history",
        "reviewer_role": "risk_analyst",
        "sla_hours": 24
    },
    {
        "condition": "regulatory_flag",
        "reason": "Transaction flagged for regulatory review",
        "reviewer_role": "compliance_officer",
        "sla_hours": 4
    },
]

The _assemble_context method is where the oversight becomes genuinely useful rather than merely procedural. A human reviewer who receives “please review this transaction” makes a worse decision than one who receives the transaction data, the agent’s recommendation, the reasoning chain, the applicable rules, and the suggested action. The context assembly is what makes human oversight efficient, the human starts with understanding, not investigation.

The sla_hours field is the detail that operational teams care about. A checkpoint without a response deadline creates a queue that grows without bound. The SLA ensures that escalated transactions are reviewed within a defined window, and transactions that aren’t reviewed within the SLA trigger secondary escalation.

Putting it together: the compliant agent loop

Here’s how the components compose into a complete agent execution loop:

class FinanceAgent:
    """A finance agent that's auditable, controlled, and supervised."""

    def __init__(self, tools, permissions, checkpoints, audit_store, llm):
        self.tools = tools
        self.permissions = permissions
        self.checkpoints = checkpoints
        self.audit = audit_store
        self.llm = llm

    async def process(self, transaction: dict) -> ProcessingResult:
        trail = []

        # Step 1: Deterministic compliance checks first
        compliance_result = await self._run_compliance_checks(transaction)
        trail.append(await self._audit(
            transaction, "compliance_check", 
            EvaluationType.DETERMINISTIC, compliance_result
        ))

        if not compliance_result.passed:
            return ProcessingResult(
                status="compliance_failed",
                details=compliance_result.failures,
                audit_trail=trail
            )

        # Step 2: AI-assisted assessment where judgment is needed
        assessment = await self._run_ai_assessment(transaction)
        trail.append(await self._audit(
            transaction, "ai_assessment",
            EvaluationType.AI_ASSISTED, assessment
        ))

        # Step 3: Check if human oversight is required
        checkpoint = await self.checkpoints.evaluate(
            transaction, assessment, trail
        )

        if checkpoint.requires_human:
            trail.append(await self._audit(
                transaction, "escalated",
                EvaluationType.AI_ASSISTED, 
                {"reason": checkpoint.reason}
            ))
            return ProcessingResult(
                status="pending_human_review",
                context=checkpoint.context,
                reviewer=checkpoint.reviewer_role,
                sla=checkpoint.sla_hours,
                audit_trail=trail
            )

        # Step 4: Execute within authority
        self.permissions.check_access("write", f"{transaction['type']}.status")

        if transaction.get("amount", 0) > self.permissions.permissions.get(
            "amount_ceiling", float("inf")
        ):
            trail.append(await self._audit(
                transaction, "amount_escalation",
                EvaluationType.DETERMINISTIC,
                {"ceiling": self.permissions.permissions["amount_ceiling"]}
            ))
            return ProcessingResult(
                status="escalated_amount_ceiling",
                audit_trail=trail
            )

        # Step 5: Execute and document
        action_result = await self._execute_action(transaction, assessment)
        trail.append(await self._audit(
            transaction, "action_executed",
            EvaluationType.DETERMINISTIC, action_result
        ))

        return ProcessingResult(
            status="completed",
            action=action_result,
            audit_trail=trail
        )

    async def _audit(self, transaction, step, eval_type, data) -> AuditRecord:
        """Every step gets an immutable audit record."""
        record = AuditRecord(
            record_id=generate_uuid(),
            timestamp=datetime.utcnow(),
            agent_id=self.agent_id,
            agent_version=self.version,
            transaction_id=transaction["id"],
            transaction_type=transaction["type"],
            input_data_hash=hash_data(transaction),
            input_data_snapshot=freeze(transaction),
            evaluation_type=eval_type,
            rules_applied=data.get("rules", []),
            model_version=self.llm.model_version if eval_type == 
                EvaluationType.AI_ASSISTED else None,
            prompt_hash=data.get("prompt_hash"),
            decision=data.get("decision", ActionType.PENDING_REVIEW),
            confidence=data.get("confidence"),
            reasoning_chain=data.get("reasoning", []),
            escalated_to=data.get("escalated_to"),
            human_reviewer=None,
            human_decision=None,
            human_override_reason=None,
            human_decision_timestamp=None,
            previous_record_hash=await self.audit.get_latest_hash(
                transaction["id"]
            )
        )
        await self.audit.append(record)
        return record

The execution order matters. Compliance checks run first, deterministic, binary, fast. If the transaction fails a regulatory requirement, there’s no point running the AI assessment. The AI assessment runs second, only for transactions that need interpretive evaluation. The checkpoint evaluation runs third, comparing the transaction characteristics and agent confidence against the oversight rules. The action execution runs last, only for transactions that passed compliance, received a confident assessment, and didn’t trigger a checkpoint.

Every step produces an audit record. The complete chain for a single transaction typically contains four to six records, each linked to the previous by hash. A regulator reviewing a decision can trace the complete chain: what data came in, what compliance checks passed, what the AI assessed and why, whether a human reviewed it, and what action was taken.

The testing that makes compliance teams comfortable

The code above gets you an architecture. The testing below gets you the approval to deploy it.

# Golden test set: known-outcome transactions for regression testing
class ComplianceRegressionSuite:
    """Run against every agent update, every prompt change, every model version."""

    def __init__(self, golden_set_path: str, drift_threshold: float = 0.02):
        self.golden_set = load_golden_set(golden_set_path)
        self.threshold = drift_threshold  # Max 2% decision drift

    async def run(self, agent: FinanceAgent) -> RegressionResult:
        results = []
        for case in self.golden_set:
            actual = await agent.process(case.transaction)
            results.append({
                "case_id": case.id,
                "expected_decision": case.expected_decision,
                "actual_decision": actual.status,
                "match": actual.status == case.expected_decision,
                "confidence": actual.audit_trail[-1].confidence
            })

        match_rate = sum(r["match"] for r in results) / len(results)
        drift = 1.0 - match_rate

        return RegressionResult(
            passed=drift <= self.threshold,
            match_rate=match_rate,
            drift=drift,
            failures=[r for r in results if not r["match"]],
            recommendation="BLOCK" if drift > self.threshold else "APPROVE"
        )

The golden test set contains 300 to 500 transactions with known correct outcomes, labelled by experienced finance professionals, not by the development team. Every agent update, every prompt change, and every model version upgrade runs against this set. If the decision outcomes drift by more than 2 percent from the baseline, the update is blocked.

This regression suite is what we point to when compliance asks “how do you know the agent still works correctly after a change?” The answer is: we run it against 500 transactions where we know the right answer, and we block any change that moves the outcomes by more than 2 percent.

For a deeper dive into the full development lifecycle, from workflow assessment through architecture selection, integration patterns, and deployment, our guide on how to build finance AI agents covers the engineering decisions we make before writing any agent code.

What most teams get wrong

Three patterns consistently cause compliance rejections or production failures.

Building the audit trail as an afterthought. If the audit architecture isn’t designed before the agent logic, the audit records won’t capture the right data at the right granularity. Retrofitting audit trails onto a working agent means re-engineering the decision flow to emit records at each step, which is more expensive than building it in from the start.

Testing with synthetic data only. Compliance teams want to see the agent’s performance on real transactions, transactions that humans have already processed with known correct outcomes. Synthetic test data proves the code works. Real test data proves the agent works on your actual business workflows.

Treating the human oversight as a failure mode rather than a design feature. The checkpoint architecture isn’t the system failing and asking for help. It’s the system correctly identifying which transactions need human judgment and routing them efficiently. The 20 to 30 percent of transactions that escalate aren’t a limitation, they’re the system working as designed.

The bottom line for finance developers

A finance AI agent that works but can’t prove it works is a demo, not a deployment. The audit trail, the access controls, and the human oversight checkpoints are what transform working code into a production system that compliance teams approve, regulators can examine, and operations teams trust.

Build the audit architecture first. Design the permissions before the agent logic. Define the oversight checkpoints with the compliance team, not for them. Test against real transactions with known outcomes. And treat every deployment as the beginning of a continuous proving process, not the end of a build project.

For teams ready to build finance agents that survive audit, the bespoke AI agent development practice at Dextra Labs designs and deploys agents with the compliance architecture described in this article as a foundational requirement, not an add-on. The audit trail, the access controls, and the oversight checkpoints are built into the agent from day one, because in finance, working without proving it is the same as not working.

Published by Dextra Labs - AI Consulting and Enterprise Agent Development