Automating Financial Statement Auditing and Risk Assessment Using RAG and LangChain
Scope: an audit assistant, not an auditor
This pipeline assists a human auditor by (1) computing standard financial ratios and flagging statistical outliers against prior periods and industry benchmarks, and (2) using retrieval-augmented generation to let an LLM answer specific questions about a company's financial statements grounded in the actual filed documents, citing the source page/section for every claim. It does not replace auditor judgment, materiality assessment, or professional skepticism — every flagged item routes to a human for review, and the system is designed to make false negatives (missed issues) less likely than a purely manual first pass, not to make final determinations.
1. Technical architecture
┌─────────────────────┐ ┌────────────────────┐ ┌─────────────────────────┐
│ 10-K/10-Q PDFs, │──▶│ Document loader & │──▶│ Chunking (section-aware) │
│ prior audit notes │ │ table extraction │ │ + embedding │
└─────────────────────┘ └────────────────────┘ └────────────┬────────────┘
▼
┌─────────────────────────┐
│ Vector store (Chroma) │
└────────────┬────────────┘
┌─────────────────────┐ │
│ Structured financials │ │
│ (extracted line items) │ │
└──────────┬──────────┘ │
▼ │
┌─────────────────────┐ │
│ Ratio & anomaly engine │ │
│ (YoY variance, Beneish M-│ │
│ Score style checks) │ │
└──────────┬──────────┘ │
│ flagged items │
└───────────────────────┐ ┌───────────────┘
▼ ▼
┌─────────────────────────────────┐
│ LangChain RAG chain: for each │
│ flagged item, retrieve supporting │
│ context + generate cited explanation │
└──────────────┬──────────────────┘
▼
┌─────────────────────────────────┐
│ Audit review queue (flag, ratio, │
│ citation, LLM explanation, status) │
└─────────────────────────────────┘
2. Complete code implementation
2.1 Document ingestion and chunking
# audit/ingest.py
import re
from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
SECTION_HEADERS = re.compile(r"^(ITEM\s+\d+[A-Z]?\.|Note\s+\d+\s*[—\-:])", re.IGNORECASE | re.MULTILINE)
def load_and_chunk(pdf_path: str, company: str, fiscal_year: int):
loader = PyPDFLoader(pdf_path)
pages = loader.load() # one Document per PDF page, with page_content + metadata
splitter = RecursiveCharacterTextSplitter(
chunk_size=1200,
chunk_overlap=200,
separators=["\n\n", "\n", ". ", " "],
)
chunks = splitter.split_documents(pages)
for chunk in chunks:
chunk.metadata.update({
"company": company,
"fiscal_year": fiscal_year,
"source_file": Path(pdf_path).name,
})
return chunks
def build_vector_store(chunks, persist_dir: str = "audit/chroma_db"):
if not chunks:
raise ValueError("No document chunks to index — check that the PDF loaded correctly")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = Chroma.from_documents(chunks, embeddings, persist_directory=persist_dir)
store.persist()
return store
2.2 Ratio and anomaly engine
# audit/anomaly_engine.py
from dataclasses import dataclass
@dataclass
class FinancialStatement:
fiscal_year: int
revenue: float
cost_of_goods_sold: float
net_income: float
total_assets: float
total_liabilities: float
receivables: float
operating_cash_flow: float
@dataclass
class Flag:
metric: str
current_value: float
prior_value: float
pct_change: float
severity: str # "info" | "warning" | "high"
rationale: str
def _pct_change(current: float, prior: float) -> float | None:
if prior == 0:
return None
return (current - prior) / abs(prior)
def compute_flags(current: FinancialStatement, prior: FinancialStatement,
variance_threshold: float = 0.25) -> list[Flag]:
flags: list[Flag] = []
gross_margin_current = (current.revenue - current.cost_of_goods_sold) / current.revenue if current.revenue else 0
gross_margin_prior = (prior.revenue - prior.cost_of_goods_sold) / prior.revenue if prior.revenue else 0
margin_delta = gross_margin_current - gross_margin_prior
if abs(margin_delta) > 0.10:
flags.append(Flag(
metric="gross_margin",
current_value=round(gross_margin_current, 4),
prior_value=round(gross_margin_prior, 4),
pct_change=round(margin_delta, 4),
severity="warning" if abs(margin_delta) < 0.15 else "high",
rationale="Gross margin shifted more than 10 percentage points year over year.",
))
receivables_growth = _pct_change(current.receivables, prior.receivables)
revenue_growth = _pct_change(current.revenue, prior.revenue)
if receivables_growth is not None and revenue_growth is not None:
gap = receivables_growth - revenue_growth
if gap > variance_threshold:
flags.append(Flag(
metric="receivables_vs_revenue_growth",
current_value=round(receivables_growth, 4),
prior_value=round(revenue_growth, 4),
pct_change=round(gap, 4),
severity="high" if gap > 0.40 else "warning",
rationale="Receivables grew significantly faster than revenue — a classic revenue-recognition red flag.",
))
ni_growth = _pct_change(current.net_income, prior.net_income)
ocf_growth = _pct_change(current.operating_cash_flow, prior.operating_cash_flow)
if ni_growth is not None and ocf_growth is not None:
divergence = ni_growth - ocf_growth
if divergence > variance_threshold:
flags.append(Flag(
metric="net_income_vs_operating_cash_flow",
current_value=round(ni_growth, 4),
prior_value=round(ocf_growth, 4),
pct_change=round(divergence, 4),
severity="high" if divergence > 0.40 else "warning",
rationale="Net income growth substantially outpaced operating cash flow growth — possible earnings quality concern.",
))
debt_to_assets = current.total_liabilities / current.total_assets if current.total_assets else 0
if debt_to_assets > 0.80:
flags.append(Flag(
metric="leverage_ratio",
current_value=round(debt_to_assets, 4),
prior_value=0.0,
pct_change=0.0,
severity="warning",
rationale="Total liabilities exceed 80% of total assets, indicating high leverage relative to typical thresholds.",
))
return flags
2.3 RAG-grounded explanation chain
# audit/rag_chain.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from openai import RateLimitError, APITimeoutError
PROMPT = ChatPromptTemplate.from_template(
"""You are assisting a financial statement audit. Using ONLY the context
below, explain the following flagged metric in 2-3 sentences and cite the
source page for every factual claim using the format [p.<page>]. If the
context does not contain enough information to explain the flag, say so
explicitly rather than speculating.
Flagged metric: {metric}
Rationale: {rationale}
Current value: {current_value}, Prior value: {prior_value}
Context:
{context}
"""
)
class AuditRagChain:
def __init__(self, persist_dir: str = "audit/chroma_db"):
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.store = Chroma(persist_directory=persist_dir, embedding_function=embeddings)
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
self.chain = PROMPT | self.llm | StrOutputParser()
def explain_flag(self, flag, company: str, fiscal_year: int, max_retries: int = 3) -> str:
query = f"{flag.metric} {flag.rationale} {company} {fiscal_year}"
docs = self.store.similarity_search(query, k=4, filter={"company": company})
if not docs:
return "No supporting context found in the indexed filings for this flag — requires manual review."
context = "\n\n".join(f"[p.{d.metadata.get('page', '?')}] {d.page_content}" for d in docs)
for attempt in range(max_retries):
try:
return self.chain.invoke({
"metric": flag.metric,
"rationale": flag.rationale,
"current_value": flag.current_value,
"prior_value": flag.prior_value,
"context": context,
})
except RateLimitError:
if attempt == max_retries - 1:
return "LLM explanation unavailable (rate limited) — flag requires manual review."
import time
time.sleep(2 ** attempt)
except APITimeoutError:
if attempt == max_retries - 1:
return "LLM explanation timed out — flag requires manual review."
continue
3. Step-by-step configuration guide
- Install dependencies:
pip install langchain langchain-openai langchain-community chromadb pypdf --break-system-packages - Set your API key:
export OPENAI_API_KEY="sk-...". - Extract structured financials — for most public filings, pull the standardized line items from a source like the SEC's XBRL "company facts" API rather than parsing tables out of the PDF by hand; use those values to populate
FinancialStatement. PDF ingestion (ingest.py) is for narrative sections (MD&A, footnotes) that need RAG-based Q&A, not for the numeric statements themselves. - Build the index per filing:
from audit.ingest import load_and_chunk, build_vector_store chunks = load_and_chunk("filings/acme_corp_10k_2025.pdf", company="Acme Corp", fiscal_year=2025) build_vector_store(chunks) - Run the anomaly engine against two consecutive periods'
FinancialStatementobjects to get alist[Flag]. - Generate grounded explanations:
from audit.rag_chain import AuditRagChain rag = AuditRagChain() for flag in flags: explanation = rag.explain_flag(flag, company="Acme Corp", fiscal_year=2025) print(flag.metric, flag.severity, "->", explanation) - Route every flag to the audit review queue (a simple table: metric, severity, citation, explanation, reviewer, status) rather than auto-resolving anything — this system's job ends at surfacing well-explained candidates for human review.
4. Error handling and edge cases
- Missing prior-period data —
_pct_changereturnsNone(not zero or an exception) when the prior value is 0 or unavailable, and every caller checks forNoneexplicitly before computing a delta, avoiding a divide-by-zero and avoiding silently treating "no data" as "no change." - Empty vector store retrieval —
explain_flagchecksif not docsand returns an explicit "no supporting context" message rather than calling the LLM with empty context, which would otherwise risk a fabricated-sounding explanation with no real grounding. - LLM rate limits and timeouts are retried with exponential backoff up to
max_retries, and on final failure the function returns a clear "requires manual review" string rather than raising and halting the whole batch — one slow/failed explanation should not block processing of the remaining flagged items. - Malformed or scanned (image-only) PDFs —
PyPDFLoaderreturns empty or near-emptypage_contentfor scanned pages with no text layer;build_vector_storeraises aValueErrorifchunksis empty, forcing an explicit decision (e.g., route to an OCR preprocessing step) instead of silently building an empty, useless index. - False positives in the anomaly engine — thresholds (
0.10margin delta,0.25variance gap,0.80leverage ratio) are intentionally conservative starting points, not calibrated to any specific industry; tune them against several years of the company's own historical variance before relying on the flag volume, and always route flags to a human rather than auto-escalating. - Cross-company retrieval bleed —
similarity_searchis called withfilter={"company": company}specifically to prevent one company's filing language from being retrieved (and cited) when explaining another company's flag, which would otherwise be a serious citation-accuracy bug in a multi-tenant audit tool. - XBRL tag inconsistency across filers — different companies sometimes use different XBRL tags for economically similar line items (e.g., varying revenue-recognition tag names); validate extracted
FinancialStatementvalues against the filing's face financials before running them through the anomaly engine, rather than trusting automated tag mapping blindly.
