How to Build an AI Agent for Reconciliation

from dataclasses import dataclass
from datetime import date

@dataclass
class Txn:
    source: str          # "gateway" | "bank" | "ledger"
    ref: str             # order id, UTR, or settlement id
    amount: float        # signed: credit positive, debit negative
    on: date
    party: str = ""

def normalize(rows: list[dict], source: str, cols: dict) -> list[Txn]:
    # cols maps THIS source's column names onto the common shape
    return [Txn(source, str(r[cols["ref"]]), float(r[cols["amount"]]),
                parse_date(r[cols["date"]]), str(r.get(cols.get("party", ""), "")))
            for r in rows]
# Now gateway, bank, and ledger are all list[Txn], ready to match the same way.

def exact_match(ledger: list[Txn], feed: list[Txn], tol: float = 1.0):
    by_ref = {t.ref: t for t in ledger if t.ref}
    matched, leftovers = [], []
    for t in feed:
        m = by_ref.get(t.ref)
        if m and abs(m.amount - t.amount) <= tol:
            matched.append((m, t))          # same reference, same amount
        else:
            leftovers.append(t)             # send to the fuzzy pass
    return matched, leftovers
# Reference plus amount within tolerance is a match you can defend in an audit.

def fuzzy_match(open_ledger: list[Txn], leftovers: list[Txn],
                days: int = 3, tol: float = 1.0, gate: float = 0.6):
    pairs, review = [], []
    for t in leftovers:
        best, score = None, 0.0
        for m in open_ledger:
            if abs(m.amount - t.amount) > tol:        continue
            if abs((m.on - t.on).days) > days:        continue
            s = 1 - abs((m.on - t.on).days) / (days + 1)   # closer date scores higher
            if same_party(m.party, t.party): s += 0.3
            if s > score: best, score = m, round(s, 2)
        (pairs if best and score >= gate else review).append((t, best, score))
    return pairs, review            # review = low confidence, a human decides

def check_settlement(gross: float, net: float,
                     fee_rate: float = 0.02, gst: float = 0.18, tol: float = 1.0):
    fee = round(gross * fee_rate, 2)
    expected_net = round(gross - fee - fee * gst, 2)   # GST applies on the fee
    gap = round(net - expected_net, 2)
    if abs(gap) <= tol:
        return "ok", gap                  # the difference was only the fee
    return ("short_settlement" if gap < 0 else "over_credit"), gap
# Run on every payout: this is how a fee charged at the wrong rate gets noticed.

from google import genai
client = genai.Client()
LABELS = ["bank_charge", "short_payment", "timing_difference",
          "missing_settlement", "duplicate", "chargeback", "other"]

def classify(txn: Txn, gap: float) -> str:
    return client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Classify this reconciliation exception into one of {LABELS}. "
                  f"Reply with the label only. "
                  f"Narration: {txn.party} {txn.ref}. Gap: Rs {gap}."]).text.strip()

from google import genai
client = genai.Client()

def propose_entry(txn: Txn, label: str, gap: float) -> str:
    return client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Propose a one-line double-entry journal to clear a "
                  f"'{label}' of Rs {gap} for reference {txn.ref}. "
                  f"Format exactly: Dr <account> / Cr <account> Rs <amount>."]).text.strip()
# Output is a suggestion shown to a reviewer, not something that posts itself.

def settle(matched, review, auto_post: bool = True) -> dict:
    posted = 0
    for (m, t) in matched:                       # exact or above the gate
        if auto_post:
            ledger.mark_reconciled(m, t)         # within tolerance, written to audit
            posted += 1
    for (t, suggestion, score) in review:        # below the gate or an exception
        review_queue.add(t, suggestion=suggestion, confidence=score)
    return {"posted": posted, "to_review": len(review)}
# A clean run posts the obvious and hands a human only what truly needs a decision.

from google.adk.agents import SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

recon_pipeline = SequentialAgent(name="recon_pipeline",
    sub_agents=[matcher, classifier])          # each an LlmAgent with the tools above
runner = Runner(agent=recon_pipeline, app_name="recon",
                session_service=InMemorySessionService())

def guard(row: Txn, gap: float, ceiling: float = 5000) -> str | None:
    if ledger.seen(row.ref):        return "already_reconciled"   # idempotent
    if abs(gap) > ceiling:          return "needs_human"          # too big to auto-post
    return None
# Run nightly over the day's gateway, bank, and ledger feeds via Cloud Scheduler.