How to Implement Mortgage Approval Workflows with LangGraph and Persistent State in Oracle

Build deterministic mortgage approval workflows by combining LangGraph StateGraph orchestration with Oracle AI Agent Memory (OAMP) to persist every verification stage, enabling conditional routing between identity, credit, and income checks while maintaining complete audit trails that survive system crashes.

This guide demonstrates the production-ready pattern found in the oracle-devrel/oracle-ai-developer-hub repository for implementing mortgage approval workflows with LangGraph and persistent state in Oracle. You will learn to orchestrate multiple LLM-powered verification agents using a typed state graph while storing all intermediate decisions and policy data in Oracle AI Agent Memory, ensuring regulatory compliance and workflow resumability.

Architectural Overview

The reference implementation in notebooks/agent_memory/03_mortgage_workflow_langgraph.ipynb combines four core components to create a deterministic orchestration layer with durable state:

• LangGraph StateGraph: Declaratively models the mortgage pipeline as nodes (identity, credit, income verification) with typed state and conditional routing via add_conditional_edges.
• TypedDict State (MortgageState): A strongly-typed dictionary defined with total=False that travels through the graph, where each node returns partial updates merged automatically by LangGraph.
• Oracle AI Agent Memory (OracleAgentMemory): Provides durable, searchable storage for applicant data, audit trails, and policy facts using text-embedding-3-small embeddings and gpt-4o-mini for extraction.
• Conditional Routing Functions: Business rules like minimum credit scores (620) and identity verification gates are encoded in route_after_identity and route_after_credit, keeping node logic deterministic and testable.

This architecture separates deterministic orchestration (the graph topology) from non-deterministic reasoning (LLM agent interpretation of documents), while OAMP provides crash resumability and policy-as-data capabilities.

Prerequisites and Installation

Install the required packages including oracleagentmemory with LiteLM support, LangGraph, and LangChain components:

%pip install -q "oracleagentmemory[litellm]" langgraph langchain langchain-openai nest_asyncio

Configure environment variables for OpenAI and Oracle Database connectivity:

import os, nest_asyncio

os.environ.setdefault("OPENAI_API_KEY", "sk-…")
os.environ.setdefault("DB_USER", "VECTOR")
os.environ.setdefault("DB_PASSWORD", "VectorPwd_2025")
os.environ.setdefault("DB_CONNECT_STRING", "localhost:1521/FREEPDB1")
nest_asyncio.apply()

Initialize Oracle AI Agent Memory

Create a persistent memory client connected to Oracle Database with automatic schema creation. The table_name_prefix ensures isolated tables for the mortgage workflow:

import oracledb
from oracleagentmemory.core import OracleAgentMemory
from oracleagentmemory.core.llms import Llm

connection = oracledb.connect(
    user=os.environ["DB_USER"],
    password=os.environ["DB_PASSWORD"],
    dsn=os.environ["DB_CONNECT_STRING"],
)

extraction_llm = Llm("gpt-4o-mini", temperature=0.1)

memory_client = OracleAgentMemory(
    connection=connection,
    embedder="text-embedding-3-small",
    llm=extraction_llm,
    extract_memories=False,
    schema_policy="create_if_necessary",
    table_name_prefix="mortgage_",
)

Seed Policy Facts and Define State

Store underwriting thresholds as immutable fact records in OAMP, enabling policy updates without code changes:

store = memory_client._store

POLICY_FACTS = [
    "Minimum acceptable credit score for conventional loans is 620.",
    "Maximum acceptable debt-to-income ratio is 43%.",
    "Accepted income documentation: W-2 forms, 1040 returns, pay stubs from the last 30 days.",
    "Identity verification requires a government-issued photo ID and a utility bill or bank statement.",
    "Loans above $1.5M require manual senior-underwriter review regardless of automated decision.",
]

store.add(
    contents=POLICY_FACTS, 
    record_type="fact",
    user_ids="underwriter-alex", 
    agent_ids="mortgage-workflow-v1"
)

Define the typed state schema that flows through the graph, capturing all verification outcomes and the final decision:

from typing_extensions import TypedDict
from typing import Optional

class MortgageState(TypedDict, total=False):
    applicant_id: str
    applicant_name: str
    loan_amount: float
    property_value: float
    identity_verified: bool
    identity_notes: str
    credit_score: int
    credit_notes: str
    monthly_income: float
    monthly_debt: float
    income_notes: str
    dti_ratio: float
    decision: str          # "approve" | "deny" | "manual_review"

    decision_reason: str

Create a helper function to write structured audit memories after each stage:

def log_stage(applicant_id: str, stage: str, outcome: str, rationale: str) -> str:
    content = f"[{stage}] applicant={applicant_id} outcome={outcome}. {rationale}"
    return memory_client.add_memory(
        content,
        user_id="underwriter-alex",
        agent_id="mortgage-workflow-v1",
        metadata={"applicant_id": applicant_id, "stage": stage, "outcome": outcome},
    )

Construct the StateGraph

Define Verification Nodes

Each node invokes its respective agent, parses the LLM response, logs the audit trail, and returns a partial state update. The identity node example demonstrates the pattern:

def identity_node(state: MortgageState) -> dict:
    out = identity_agent.invoke({
        "messages": [{"role": "user",
                      "content": f"Verify applicant {state['applicant_id']}."}],
    })
    text = out["messages"][-1].content
    verified = "true" in text.lower().split("verified:", 1)[-1][:20]
    log_stage(state["applicant_id"], "identity",
              "pass" if verified else "fail", text)
    return {"identity_verified": verified, "identity_notes": text}

Configure Conditional Routing

Implement business logic gates that determine the next node based on current state values:

def route_after_identity(state: MortgageState) -> str:
    return "credit" if state.get("identity_verified") else "decide"

def route_after_credit(state: MortgageState) -> str:
    return "income" if (state.get("credit_score") or 0) >= 620 else "decide"

Assemble the Graph

Wire nodes and conditional edges to create the complete workflow topology from START to END:

from langgraph.graph import StateGraph, START, END

builder = StateGraph(MortgageState)
builder.add_node("identity", identity_node)
builder.add_node("credit", credit_node)
builder.add_node("income", income_node)
builder.add_node("dti", dti_node)
builder.add_node("decide", decide_node)

builder.add_edge(START, "identity")
builder.add_conditional_edges("identity", route_after_identity,
                              {"credit": "credit", "decide": "decide"})
builder.add_conditional_edges("credit", route_after_credit,
                              {"income": "income", "decide": "decide"})
builder.add_edge("income", "dti")
builder.add_edge("dti", "decide")
builder.add_edge("decide", END)

graph = builder.compile()

Execute and Audit the Workflow

Run Complete Applications

Process an applicant dictionary through the compiled graph to receive the final state containing the decision and complete audit history:

applicant = {
    "applicant_id": "A-001",
    "applicant_name": "J. Patel",
    "loan_amount": 480_000,
    "property_value": 600_000,
}

final_state = graph.invoke(applicant)
print(f"Decision: {final_state['decision'].upper()}")
print(f"Reason:   {final_state['decision_reason']}")

Stream Real-Time Node Updates

Use stream_mode="updates" to introspect graph execution for debugging or UI feedback:

for chunk in graph.stream(applicant, stream_mode="updates"):
    for node, update in chunk.items():
        print(f"[{node}] -> {list(update.keys())}")

Retrieve Complete Audit Trails

Query OAMP to reconstruct every decision point for a specific applicant using metadata filtering:

trail = store.list(
    "memory",
    user_id="underwriter-alex",
    agent_id="mortgage-workflow-v1",
    metadata_filter={"applicant_id": "A-001"},
    limit=50,
)

for r in trail:
    stage = (r.metadata or {}).get("stage", "?")
    outcome = (r.metadata or {}).get("outcome", "?")
    print(f"[{stage:<8}] [{outcome}] {r.content[:100]}")

Summary

• LangGraph StateGraph provides deterministic orchestration of mortgage verification stages with add_conditional_edges encoding business rules directly in the graph topology.
• Oracle AI Agent Memory persists every workflow state and audit record to Oracle Database, enabling crash resumability and compliance with immutable audit trails.
• TypedDict state ensures type safety across nodes while allowing partial updates that LangGraph merges automatically.
• Policy-as-data storage allows updating underwriting thresholds (like the 620 credit score minimum) without redeploying code.

Frequently Asked Questions

How does LangGraph StateGraph ensure deterministic execution?

The StateGraph compiles into a fixed topology where nodes communicate via an immutable MortgageState object. Conditional edges (route_after_identity, route_after_credit) return deterministic string literals mapped to node names, ensuring the same inputs always produce the same execution path. All non-deterministic reasoning is isolated inside agent nodes that do not affect the graph structure.

What is Oracle AI Agent Memory and why is it critical for mortgage workflows?

Oracle AI Agent Memory (OAMP) is a durable vector store backed by Oracle Database that persists applicant data, verification results, and policy facts. For regulated mortgage workflows, OAMP provides an immutable audit trail where every log_stage call creates a searchable record with metadata tags like applicant_id and stage, satisfying compliance requirements for decision explainability and data retention.

How do conditional edges handle business logic like minimum credit scores?

Conditional routing functions inspect the current state and return literal strings mapped to next nodes. In route_after_credit, the function checks if state.get("credit_score") >= 620 and returns either "income" to continue processing or "decide" to route directly to the final decision node for denial. This keeps business logic declarative and testable outside of node implementations.

Can the workflow resume if a node crashes during execution?

Yes. Because Oracle AI Agent Memory persists state after every successful node completion (via log_stage and implicit state storage), the MortgageState up to the last successful node is preserved in Oracle Database. A new run can query the memory store for the last known state of an applicant_id and restart from that checkpoint, providing fault tolerance for long-running verification processes.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}