From 59d384bdbcb894cec296c682681e3f5eee70e957 Mon Sep 17 00:00:00 2001 From: Prasenjit Date: Fri, 11 Sep 2026 13:32:02 +0530 Subject: [PATCH] feat: add Dozer + LangChain + Vector RAG banking advisor sample (fixes #1690) - Add comprehensive Dozer streaming SQL configuration (dozer-config.yaml) materializing real-time customer profiles from multi-source CSVs - Dual-mode DozerClient with live REST API querying and zero-dependency offline SQL stream emulation - LangChain Vector RAG with Chroma integration and customer transaction spend concentration weighting - Bank-grade financial underwriting guardrails enforcing credit score, income, and risk tolerance thresholds - Interactive consultation chat and machine-readable JSON CLI - Full 19-test automated test suite covering config validation, client aggregation, guardrails, and RAG retrieval --- examples/llm-banking-advisor/.gitignore | 5 + examples/llm-banking-advisor/README.md | 166 +++++++++++++++ examples/llm-banking-advisor/advisor.py | 153 ++++++++++++++ examples/llm-banking-advisor/app.py | 124 ++++++++++++ .../credit_card_products.csv | 6 + .../customer_profiles/customer_profiles.csv | 6 + .../data/transactions/transactions.csv | 25 +++ .../llm-banking-advisor/dozer-config.yaml | 93 +++++++++ examples/llm-banking-advisor/dozer_client.py | 191 ++++++++++++++++++ examples/llm-banking-advisor/guardrails.py | 67 ++++++ examples/llm-banking-advisor/models.py | 81 ++++++++ examples/llm-banking-advisor/requirements.txt | 6 + .../llm-banking-advisor/tests/__init__.py | 0 .../llm-banking-advisor/tests/test_advisor.py | 48 +++++ .../llm-banking-advisor/tests/test_config.py | 49 +++++ .../tests/test_dozer_client.py | 51 +++++ .../tests/test_guardrails.py | 125 ++++++++++++ .../tests/test_vector_rag.py | 52 +++++ examples/llm-banking-advisor/vector_rag.py | 115 +++++++++++ 19 files changed, 1363 insertions(+) create mode 100644 examples/llm-banking-advisor/.gitignore create mode 100644 examples/llm-banking-advisor/README.md create mode 100644 examples/llm-banking-advisor/advisor.py create mode 100755 examples/llm-banking-advisor/app.py create mode 100644 examples/llm-banking-advisor/data/credit_card_products/credit_card_products.csv create mode 100644 examples/llm-banking-advisor/data/customer_profiles/customer_profiles.csv create mode 100644 examples/llm-banking-advisor/data/transactions/transactions.csv create mode 100644 examples/llm-banking-advisor/dozer-config.yaml create mode 100644 examples/llm-banking-advisor/dozer_client.py create mode 100644 examples/llm-banking-advisor/guardrails.py create mode 100644 examples/llm-banking-advisor/models.py create mode 100644 examples/llm-banking-advisor/requirements.txt create mode 100644 examples/llm-banking-advisor/tests/__init__.py create mode 100644 examples/llm-banking-advisor/tests/test_advisor.py create mode 100644 examples/llm-banking-advisor/tests/test_config.py create mode 100644 examples/llm-banking-advisor/tests/test_dozer_client.py create mode 100644 examples/llm-banking-advisor/tests/test_guardrails.py create mode 100644 examples/llm-banking-advisor/tests/test_vector_rag.py create mode 100644 examples/llm-banking-advisor/vector_rag.py diff --git a/examples/llm-banking-advisor/.gitignore b/examples/llm-banking-advisor/.gitignore new file mode 100644 index 0000000000..f44d5b5dcd --- /dev/null +++ b/examples/llm-banking-advisor/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +.pytest_cache/ +.chroma/ + diff --git a/examples/llm-banking-advisor/README.md b/examples/llm-banking-advisor/README.md new file mode 100644 index 0000000000..eb995dbf11 --- /dev/null +++ b/examples/llm-banking-advisor/README.md @@ -0,0 +1,166 @@ +# Hyper-Personalized Banking Advisor with Dozer, LangChain & Vector RAG + +This sample demonstrates an end-to-end AI-powered banking advisor application inspired by Dozer's published article on [hyper-personalization using LLMs, Vector Databases, and LangChain](https://getdozer.io/blog/llm-chatbot). + +--- + +## ๐ŸŒŸ Architecture Overview + +Modern financial applications require real-time context. Static customer demographic data is not enough to make relevant recommendations; an advisor needs to understand real-time transaction spending velocity, category concentrations, and strict underwriting guardrails. + +``` +โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” +โ”‚ Multi-Source Datasets โ”‚ +โ”‚ - Customer Profiles (CSV) โ”‚ +โ”‚ - Real-time Transactions (CSV) โ”‚ +โ”‚ - Card Product Catalog (CSV) โ”‚ +โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ + โ”‚ + โ–ผ +โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” +โ”‚ DOZER CORE โ”‚ +โ”‚ - Ingests via !LocalStorage (connections) โ”‚ +โ”‚ - Streaming SQL: Computes real-time customer profile โ”‚ +โ”‚ JOINing profiles + transactions with GROUP BY โ”‚ +โ”‚ - Exposes queryable endpoints: โ”‚ +โ”‚ โ€ข /customer_profile (REST port 8080 / gRPC 50051) โ”‚ +โ”‚ โ€ข /card_catalog โ”‚ +โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ + โ”‚ + โ–ผ +โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” +โ”‚ Financial & Risk Guardrails โ”‚ +โ”‚ - Minimum Age Validation (18+) โ”‚ +โ”‚ - FICO Score Thresholds โ”‚ +โ”‚ - Minimum Income Requirements โ”‚ +โ”‚ - Risk Tolerance Alignment โ”‚ +โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ + โ”‚ (Eligible Products Only) + โ–ผ +โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” +โ”‚ LangChain Vector RAG โ”‚ +โ”‚ - Chroma Vector Store โ”‚ +โ”‚ - Semantic Catalog Embeddings โ”‚ +โ”‚ - Category Spend Weighting โ”‚ +โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ + โ”‚ + โ–ผ +โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” +โ”‚ Personalized Advisory โ”‚ +โ”‚ - Grounded Rationale โ”‚ +โ”‚ - Spend Alignment Highlights โ”‚ +โ”‚ - Dual LLM (OpenAI/Local Rule) โ”‚ +โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ +``` + +--- + +## ๐Ÿš€ Key Highlights + +1. **Modern Dozer Configuration**: Uses standard `sql INTO customer_profile` streaming transformations fully compliant with Dozer's schema. +2. **Hard Bank Eligibility Guardrails**: Rejects ineligible products *before* vector ranking and LLM prompting, preventing hallucinated product offers. +3. **Dual-Mode Operation**: + - **Live Mode**: Queries Dozer REST API (`http://localhost:8080/customer_profile/query` and `http://localhost:8080/card_catalog`). + - **Offline / Local Emulation Mode**: Emulates the exact Dozer streaming SQL pipeline locally so CI, unit tests, and local development run with zero external binary dependencies. +4. **LangChain & Vector RAG**: Semantic vector retrieval over card rewards and perks, augmented with dynamic weights derived from customer spend concentrations. +5. **Flexible CLI & Interactive Chat**: Test recommendations via interactive consultation mode or generate structured JSON reports. + +--- + +## ๐Ÿ“ Project Structure + +``` +examples/llm-banking-advisor/ +โ”œโ”€โ”€ dozer-config.yaml # Modern Dozer configuration with streaming SQL +โ”œโ”€โ”€ requirements.txt # Python dependencies +โ”œโ”€โ”€ models.py # Typed dataclasses (CustomerProfile, CardProduct, etc.) +โ”œโ”€โ”€ dozer_client.py # Dual-mode Dozer API client (Live REST & Local fallback) +โ”œโ”€โ”€ guardrails.py # Bank underwriting & financial eligibility rules +โ”œโ”€โ”€ vector_rag.py # LangChain Chroma vector store & Grounded Retriever +โ”œโ”€โ”€ advisor.py # Banking Advisor orchestrator +โ”œโ”€โ”€ app.py # CLI and Interactive terminal chat interface +โ”œโ”€โ”€ data/ +โ”‚ โ”œโ”€โ”€ customer_profiles/ # Customer demographic and financial profiles +โ”‚ โ”œโ”€โ”€ transactions/ # Transaction streams with merchant categories +โ”‚ โ””โ”€โ”€ credit_card_products/ # Credit card product catalog with fee/perk details +โ””โ”€โ”€ tests/ + โ”œโ”€โ”€ test_config.py # Schema & SQL validation tests + โ”œโ”€โ”€ test_dozer_client.py # Data ingestion and SQL aggregation tests + โ”œโ”€โ”€ test_guardrails.py # Underwriting rule validation tests + โ”œโ”€โ”€ test_vector_rag.py # Vector retrieval and semantic ranking tests + โ””โ”€โ”€ test_advisor.py # End-to-end integration tests +``` + +--- + +## โšก Quickstart + +### 1. Run the Advisor CLI (Offline / Self-Contained Mode) + +The advisor runs out-of-the-box with zero setup: + +```bash +# Evaluate Elena Rostova (Frequent Traveler) +python3 app.py --customer-id CUST_001 + +# Evaluate Marcus Vance (Student seeking credit building) +python3 app.py --customer-id CUST_002 + +# Evaluate Sarah Jenkins (Family groceries and cashback) +python3 app.py --customer-id CUST_003 + +# Evaluate Amina Diallo (High Net Worth Executive) +python3 app.py --customer-id CUST_005 +``` + +### 2. Output as Machine-Readable JSON + +```bash +python3 app.py --customer-id CUST_001 --json +``` + +### 3. Interactive Terminal Consultation Mode + +```bash +python3 app.py --interactive +``` + +--- + +## ๐Ÿ”ง Running with Live Dozer Daemon + +To run with the real Dozer streaming engine: + +1. Build or download the `dozer` binary: + ```bash + cargo build --release --bin dozer + ``` + +2. Start the Dozer pipeline from this directory: + ```bash + dozer run + ``` + Dozer will initialize the LocalStorage connections, execute the streaming SQL query to materialize `customer_profile`, and start serving REST API on `http://localhost:8080`. + +3. In another terminal, run the application: + ```bash + python3 app.py --dozer-url http://localhost:8080 --customer-id CUST_001 + ``` + The application will automatically detect the live Dozer server and set `Source Provenance: DOZER_LIVE_API`. + +--- + +## ๐Ÿงช Running Automated Tests + +Run the complete test suite: + +```bash +python3 -m unittest discover -s tests -v +``` + +All 19 tests validate: +- Top-level Dozer YAML configuration and streaming SQL `INTO` targets. +- Real-time customer transaction aggregation and join accuracy. +- Strict financial guardrail rejections (underage, credit score, income, risk). +- Semantic vector similarity and spend category weighting. +- Full end-to-end recommendation pipelines. diff --git a/examples/llm-banking-advisor/advisor.py b/examples/llm-banking-advisor/advisor.py new file mode 100644 index 0000000000..b3085e00bb --- /dev/null +++ b/examples/llm-banking-advisor/advisor.py @@ -0,0 +1,153 @@ +"""Hyper-personalized Banking Advisor orchestrator integrating Dozer, Guardrails & LLM.""" + +from __future__ import annotations + +import os +from typing import Any, Dict, List, Optional + +from dozer_client import DozerClient +from guardrails import FinancialGuardrails +from models import CardProduct, CustomerProfile, RecommendationResult +from vector_rag import DozerGroundedRetriever + + +class BankingAdvisor: + """End-to-end banking product recommender based on real-time Dozer customer analytics.""" + + def __init__(self, dozer_client: Optional[DozerClient] = None) -> None: + self.client = dozer_client or DozerClient() + + def recommend( + self, + customer_id: str, + user_query: Optional[str] = None, + ) -> RecommendationResult: + """Generate a fully verified and grounded credit card recommendation.""" + # Step 1: Ingest unified customer profile from Dozer + profile = self.client.get_customer_profile(customer_id) + source_prov = "dozer_live_api" if self.client.is_live() else "dozer_local_sql_emulation" + + # Step 2: Ingest card product catalog + all_cards = self.client.get_card_catalog() + + # Step 3: Enforce Financial and Risk Guardrails (Pre-filter) + eligible_cards, ineligible_details = FinancialGuardrails.filter_eligible_cards( + profile, all_cards + ) + + if not eligible_cards: + raise ValueError( + f"Customer {profile.full_name} ({customer_id}) did not meet eligibility criteria for any active products: {ineligible_details}" + ) + + # Step 4: Semantic Retrieval & Category Spend Alignment via Vector Index + retriever = DozerGroundedRetriever(eligible_cards) + ranked = retriever.rank_cards_for_customer( + profile, user_query=user_query, top_k=len(eligible_cards) + ) + + top_card, top_score, top_meta = ranked[0] + + # Step 5: Synthesize Explanation (LLM if API key available, else Rule-Based Explainer) + rationale = self._generate_explanation( + profile=profile, + card=top_card, + score=top_score, + meta=top_meta, + user_query=user_query, + ) + + all_evaluated = [ + { + "product_id": c.product_id, + "name": c.name, + "score": round(s, 3), + "is_eligible": True, + "annual_fee_usd": c.annual_fee_usd, + "details": m, + } + for c, s, m in ranked + ] + for name, reasons in ineligible_details.items(): + all_evaluated.append({ + "name": name, + "is_eligible": False, + "ineligibility_reasons": reasons, + }) + + return RecommendationResult( + customer_id=customer_id, + recommended_product=top_card, + eligibility_status="VERIFIED_ELIGIBLE", + fit_score=round(top_score, 3), + rationale=rationale, + spend_alignment=top_meta["spend_matches"], + all_evaluated_cards=all_evaluated, + source_provenance=source_prov, + ) + + def _generate_explanation( + self, + profile: CustomerProfile, + card: CardProduct, + score: float, + meta: Dict[str, Any], + user_query: Optional[str], + ) -> str: + """Synthesize a personalized advisory note.""" + # Try OpenAI if OPENAI_API_KEY is present + if os.getenv("OPENAI_API_KEY"): + try: + from langchain_core.prompts import ChatPromptTemplate + from langchain_openai import ChatOpenAI + + prompt = ChatPromptTemplate.from_messages([ + ( + "system", + "You are an expert banking financial advisor. You explain why the selected " + "credit card was chosen for this customer based strictly on their verified " + "Dozer transaction data, annual income, credit score, and goals. Be concise (3-4 sentences).", + ), + ( + "user", + "Customer: {customer_name} (Segment: {segment}, Credit Score: {score}, Income: ${income:,.0f}).\n" + "Recent Total Spend: ${total_spend:,.2f} across categories: {categories}.\n" + "Stated Goals: {goals}.\n" + "Selected Card: {card_name} (Annual Fee: ${fee}, Perks: {perks}).\n" + "Customer Question/Prompt: {query}\n" + "Provide a professional personalized recommendation.", + ), + ]) + llm = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), temperature=0.2) + chain = prompt | llm + response = chain.invoke({ + "customer_name": profile.full_name, + "segment": profile.segment, + "score": profile.credit_score, + "income": profile.annual_income_usd, + "total_spend": profile.total_spend_usd, + "categories": ", ".join(f"{k}: ${v:.2f}" for k, v in profile.spend_by_category.items()), + "goals": profile.primary_goals, + "card_name": card.name, + "fee": card.annual_fee_usd, + "perks": card.perks_summary, + "query": user_query or "Best matching credit card recommendation", + }) + return str(response.content) + except Exception: + pass + + # Deterministic financial advisor synthesis + matched_cats = ", ".join( + f"{k.title()} (${v:,.2f})" for k, v in meta.get("spend_matches", {}).items() + ) or "everyday purchases" + + fee_str = f"with a ${card.annual_fee_usd:.0f} annual fee" if card.annual_fee_usd > 0 else "with $0 annual fee" + + return ( + f"Based on {profile.full_name}'s recent transaction profile totaling ${profile.total_spend_usd:,.2f}, " + f"the {card.name} is the optimal product ({fee_str}). " + f"The customer's highest spending concentration aligns directly with {card.rewards_focus} ({matched_cats}), " + f"while satisfying all creditworthiness standards (FICO {profile.credit_score} vs {card.min_credit_score} min). " + f"Key benefits include: {card.perks_summary}" + ) diff --git a/examples/llm-banking-advisor/app.py b/examples/llm-banking-advisor/app.py new file mode 100755 index 0000000000..34861cd531 --- /dev/null +++ b/examples/llm-banking-advisor/app.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""Dozer LLM Banking Advisor CLI and Interactive Experience.""" + +from __future__ import annotations + +import argparse +import json +import sys + +from advisor import BankingAdvisor +from dozer_client import DozerClient + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Dozer + LangChain + Vector RAG Hyper-Personalized Banking Advisor" + ) + parser.add_argument( + "--customer-id", + default="CUST_001", + help="Target customer ID to evaluate (e.g. CUST_001, CUST_002, CUST_003)", + ) + parser.add_argument( + "--query", + default=None, + help="Optional custom user search query or financial preference", + ) + parser.add_argument( + "--dozer-url", + default="http://localhost:8080", + help="Dozer REST API base URL (default: http://localhost:8080)", + ) + parser.add_argument( + "--json", + action="store_true", + help="Output result in machine-readable JSON format", + ) + parser.add_argument( + "--interactive", + action="store_true", + help="Launch interactive terminal consultation mode", + ) + + args = parser.parse_args(argv) + + client = DozerClient(base_url=args.dozer_url) + advisor = BankingAdvisor(dozer_client=client) + + if args.interactive: + print("=" * 70) + print(" ๐Ÿฆ Dozer AI Banking Advisor โ€” Interactive Consultation Mode") + print("=" * 70) + print(f"Connected to data source: {'LIVE DOZER API' if client.is_live() else 'DOZER LOCAL STREAM EMULATION'}") + print("Available Customer IDs: CUST_001, CUST_002, CUST_003, CUST_004, CUST_005") + print("Type 'exit' or 'quit' to finish.\n") + + while True: + try: + cid = input("Enter Customer ID [default: CUST_001]: ").strip() or "CUST_001" + if cid.lower() in ("exit", "quit"): + break + q = input("What is the customer looking for? [Press Enter for automatic profile matching]: ").strip() + if q.lower() in ("exit", "quit"): + break + + result = advisor.recommend(cid, user_query=q or None) + print("\n" + "-" * 50) + print(f"โœจ Recommendation for {cid}: {result.recommended_product.name}") + print(f"๐Ÿ† Fit Score: {result.fit_score}") + print(f"๐Ÿ“Š Annual Fee: ${result.recommended_product.annual_fee_usd:.0f}") + print(f"๐Ÿ’ก Rationale:\n{result.rationale}") + print("-" * 50 + "\n") + except (KeyboardInterrupt, EOFError): + break + except Exception as e: + print(f"Error: {e}\n") + return 0 + + try: + res = advisor.recommend(args.customer_id, user_query=args.query) + except Exception as exc: + print(f"Advisory Error: {exc}", file=sys.stderr) + return 1 + + if args.json: + payload = { + "customer_id": res.customer_id, + "recommended_product": res.recommended_product.to_dict(), + "fit_score": res.fit_score, + "eligibility_status": res.eligibility_status, + "spend_alignment": res.spend_alignment, + "rationale": res.rationale, + "source_provenance": res.source_provenance, + "all_evaluated_cards": res.all_evaluated_cards, + } + print(json.dumps(payload, indent=2)) + return 0 + + print("=" * 70) + print(f"๐Ÿฆ Dozer AI Banking Advisor Report | Customer: {res.customer_id}") + print(f"Source Provenance: {res.source_provenance.upper()}") + print("=" * 70) + print(f"Top Recommendation : {res.recommended_product.name}") + print(f"Product Tier : {res.recommended_product.card_tier}") + print(f"Annual Fee : ${res.recommended_product.annual_fee_usd:.0f}") + print(f"Fit Score : {res.fit_score}") + print(f"Eligibility Status : {res.eligibility_status}") + print("-" * 70) + print("Advisor Rationale:") + print(res.rationale) + print("-" * 70) + print("Evaluated Cards:") + for card in res.all_evaluated_cards: + if card.get("is_eligible"): + print(f" โ€ข [ELIGIBLE] {card['name']}: score {card['score']} (Fee: ${card['annual_fee_usd']:.0f})") + else: + reasons = "; ".join(card.get("ineligibility_reasons", [])) + print(f" โ€ข [INELIGIBLE] {card['name']}: {reasons}") + print("=" * 70) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/llm-banking-advisor/data/credit_card_products/credit_card_products.csv b/examples/llm-banking-advisor/data/credit_card_products/credit_card_products.csv new file mode 100644 index 0000000000..e57bc81b99 --- /dev/null +++ b/examples/llm-banking-advisor/data/credit_card_products/credit_card_products.csv @@ -0,0 +1,6 @@ +product_id,name,card_tier,annual_fee_usd,min_credit_score,min_annual_income_usd,rewards_focus,intro_bonus,apr_rate,perks_summary,target_segment +CARD_001,Voyager Horizon Travel Elite,Premium Travel,250,720,75000,Travel and Flights,60000 points after $4000 spend,19.99%,5x points on flights and hotels; Priority Pass lounge access; $100 annual TSA PreCheck credit; Zero foreign transaction fees.,Frequent Travelers and Young Professionals +CARD_002,Harvest Cash & Groceries Plus,Everyday Cash,0,650,25000,Groceries and Everyday Essentials,$200 cash back after $1000 spend,17.49%,3% cash back on supermarkets and wholesale clubs; 2% on gas and EV charging; 1% on all other purchases; No annual fee.,Families and Everyday Savers +CARD_003,Foundation Builder Mastercard,Starter,0,550,15000,Credit Building and Essentials,None,24.99%,Guaranteed credit line review after 6 months; 1% flat cash back on all spend; Free monthly FICO score tracking; No security deposit required.,Students and First-Time Credit Borrowers +CARD_004,Gourmet Epicure Preferred,Dining & Entertainment,95,680,50000,Dining and Food Delivery,$300 dining statement credits in year one,18.99%,4x points on restaurants and food delivery apps; 2x on entertainment and streaming; Complimentary DashPass subscription; Purchase protection.,Foodies and Urban Professionals +CARD_005,Centurion Signature Sovereign,Luxury Executive,595,780,150000,Luxury Travel and Concierge,100000 points after $10000 spend,16.99%,10x points on luxury hotels booked through portal; Global 24/7 dedicated concierge; Unlimited lounge access including Centurion Lounges; $300 annual airline incidentals credit.,High Net Worth Executives and Business Owners diff --git a/examples/llm-banking-advisor/data/customer_profiles/customer_profiles.csv b/examples/llm-banking-advisor/data/customer_profiles/customer_profiles.csv new file mode 100644 index 0000000000..2ce3c0cc28 --- /dev/null +++ b/examples/llm-banking-advisor/data/customer_profiles/customer_profiles.csv @@ -0,0 +1,6 @@ +customer_id,first_name,last_name,age,credit_score,income_band,annual_income_usd,segment,primary_goals,risk_tolerance +CUST_001,Elena,Rostova,29,750,upper_middle,115000,Young Professional,International Travel and Airline Lounges,medium +CUST_002,Marcus,Vance,22,610,starter,32000,Student,Build Credit and Low Fees,low +CUST_003,Sarah,Jenkins,41,810,affluent,195000,Established Family,Groceries Cashback and Family Dining,low +CUST_004,David,Chen,35,715,middle,82000,Tech Consultant,Dining Delivery and Gadget Protection,medium +CUST_005,Amina,Diallo,52,790,high_net_worth,340000,Executive,Luxury Travel Concierge and High Rewards,high diff --git a/examples/llm-banking-advisor/data/transactions/transactions.csv b/examples/llm-banking-advisor/data/transactions/transactions.csv new file mode 100644 index 0000000000..f227054a29 --- /dev/null +++ b/examples/llm-banking-advisor/data/transactions/transactions.csv @@ -0,0 +1,25 @@ +transaction_id,customer_id,merchant_name,merchant_category,amount_usd,txn_date +TXN_101,CUST_001,Air France,travel,840.50,2026-08-10 +TXN_102,CUST_001,Hyatt Regency,travel,420.00,2026-08-14 +TXN_103,CUST_001,Uber Eats,dining,38.50,2026-08-15 +TXN_104,CUST_001,Delta Airlines,travel,610.20,2026-08-22 +TXN_105,CUST_001,Blue Bottle Coffee,dining,12.75,2026-08-25 +TXN_106,CUST_001,Boulangerie Bistro,dining,95.00,2026-08-28 +TXN_201,CUST_002,Campus Bookstore,education,145.00,2026-08-12 +TXN_202,CUST_002,Local Grocery Mart,groceries,52.30,2026-08-18 +TXN_203,CUST_002,Subway,dining,11.50,2026-08-22 +TXN_204,CUST_002,Metro Transit Pass,transit,45.00,2026-08-28 +TXN_301,CUST_003,Whole Foods Market,groceries,265.40,2026-08-05 +TXN_302,CUST_003,Costco Wholesale,groceries,412.80,2026-08-12 +TXN_303,CUST_003,Target Superstore,groceries,189.10,2026-08-19 +TXN_304,CUST_003,Shell Gas Station,gas,72.00,2026-08-20 +TXN_305,CUST_003,Olive Garden Family Dinner,dining,145.60,2026-08-26 +TXN_401,CUST_004,DoorDash,dining,48.20,2026-08-04 +TXN_402,CUST_004,Chipotle,dining,22.50,2026-08-09 +TXN_403,CUST_004,Best Buy Electronics,technology,350.00,2026-08-15 +TXN_404,CUST_004,Sushi Omakase,dining,185.00,2026-08-21 +TXN_405,CUST_004,AWS Cloud Services,technology,120.00,2026-08-28 +TXN_501,CUST_005,Four Seasons Resort,travel,2150.00,2026-08-02 +TXN_502,CUST_005,Emirates First Class,travel,5400.00,2026-08-10 +TXN_503,CUST_005,Le Bernardin,dining,620.00,2026-08-18 +TXN_504,CUST_005,Saks Fifth Avenue,retail,1450.00,2026-08-24 diff --git a/examples/llm-banking-advisor/dozer-config.yaml b/examples/llm-banking-advisor/dozer-config.yaml new file mode 100644 index 0000000000..a272b81f4e --- /dev/null +++ b/examples/llm-banking-advisor/dozer-config.yaml @@ -0,0 +1,93 @@ +app_name: llm-banking-advisor +version: 1 + +connections: + - name: bank_data + config: !LocalStorage + details: + path: ./data + tables: + - name: customer_profiles + config: !CSV + path: customer_profiles + extension: .csv + - name: transactions + config: !CSV + path: transactions + extension: .csv + - name: credit_card_products + config: !CSV + path: credit_card_products + extension: .csv + +sources: + - name: customer_profiles + table_name: customer_profiles + connection: bank_data + - name: transactions + table_name: transactions + connection: bank_data + - name: credit_card_products + table_name: credit_card_products + connection: bank_data + +sql: | + SELECT + c.customer_id, + c.first_name, + c.last_name, + c.age, + c.credit_score, + c.income_band, + c.annual_income_usd, + c.segment, + c.primary_goals, + c.risk_tolerance, + SUM(t.amount_usd) AS total_spend_usd, + COUNT(t.transaction_id) AS transaction_count, + MAX(t.amount_usd) AS max_single_transaction_usd + INTO customer_profile + FROM customer_profiles c + JOIN transactions t ON c.customer_id = t.customer_id + GROUP BY + c.customer_id, + c.first_name, + c.last_name, + c.age, + c.credit_score, + c.income_band, + c.annual_income_usd, + c.segment, + c.primary_goals, + c.risk_tolerance; + + SELECT + product_id, + name, + card_tier, + annual_fee_usd, + min_credit_score, + min_annual_income_usd, + rewards_focus, + intro_bonus, + apr_rate, + perks_summary, + target_segment + INTO card_catalog + FROM credit_card_products; + +sinks: + - name: customer_profile + config: !Dummy + table_name: customer_profile + - name: card_catalog + config: !Dummy + table_name: card_catalog + +api: + rest: + port: 8080 + cors: true + grpc: + port: 50051 + cors: true diff --git a/examples/llm-banking-advisor/dozer_client.py b/examples/llm-banking-advisor/dozer_client.py new file mode 100644 index 0000000000..d0edafced9 --- /dev/null +++ b/examples/llm-banking-advisor/dozer_client.py @@ -0,0 +1,191 @@ +"""Dozer API Client with automatic live REST querying and offline fallback.""" + +from __future__ import annotations + +import csv +import json +import logging +import urllib.request +import urllib.error +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any, Dict, List, Optional + +from models import CardProduct, CustomerProfile, Transaction + +logger = logging.getLogger("dozer_client") +ROOT_DIR = Path(__file__).resolve().parent +DATA_DIR = ROOT_DIR / "data" + + +class DozerClient: + """Client for Dozer real-time data movement and query engine. + + Supports dual-mode operation: + 1. Live Mode: Connects to a running Dozer REST API server (default: http://localhost:8080). + 2. Offline Mode: Emulates Dozer's real-time SQL aggregation directly from local storage, + allowing dependency-free local testing and continuous integration without the Dozer binary. + """ + + def __init__(self, base_url: str = "http://localhost:8080", timeout_sec: float = 2.0) -> None: + self.base_url = base_url.rstrip("/") + self.timeout_sec = timeout_sec + self._is_live: Optional[bool] = None + + def is_live(self) -> bool: + """Check if Dozer REST endpoint is responsive.""" + if self._is_live is not None: + return self._is_live + try: + req = urllib.request.Request(f"{self.base_url}/card_catalog") + with urllib.request.urlopen(req, timeout=self.timeout_sec) as resp: + self._is_live = resp.status == 200 + except Exception: + self._is_live = False + return self._is_live + + def get_card_catalog(self) -> List[CardProduct]: + """Fetch all card products either from Dozer API or local storage.""" + if self.is_live(): + try: + req = urllib.request.Request(f"{self.base_url}/card_catalog") + with urllib.request.urlopen(req, timeout=self.timeout_sec) as resp: + if resp.status == 200: + records = json.loads(resp.read().decode("utf-8")) + return [ + CardProduct( + product_id=r["product_id"], + name=r["name"], + card_tier=r.get("card_tier", "Standard"), + annual_fee_usd=float(r["annual_fee_usd"]), + min_credit_score=int(r["min_credit_score"]), + min_annual_income_usd=float(r["min_annual_income_usd"]), + rewards_focus=r["rewards_focus"], + intro_bonus=r.get("intro_bonus", "None"), + apr_rate=r.get("apr_rate", "N/A"), + perks_summary=r.get("perks_summary", ""), + target_segment=r.get("target_segment", "General"), + ) + for r in records + ] + except Exception as e: + logger.warning(f"Dozer live fetch failed ({e}); falling back to local storage.") + + # Offline emulation + return self._load_local_products() + + def get_customer_profile(self, customer_id: str) -> CustomerProfile: + """Fetch customer profile either from Dozer streaming SQL endpoint or local storage.""" + if self.is_live(): + try: + query_payload = json.dumps({ + "$filter": {"customer_id": customer_id} + }).encode("utf-8") + req = urllib.request.Request( + f"{self.base_url}/customer_profile/query", + data=query_payload, + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=self.timeout_sec) as resp: + if resp.status == 200: + data = json.loads(resp.read().decode("utf-8")) + if data: + r = data[0] + local_profile = self._build_local_profile(customer_id) + return CustomerProfile( + customer_id=r["customer_id"], + first_name=r.get("first_name", local_profile.first_name), + last_name=r.get("last_name", local_profile.last_name), + age=int(r.get("age", local_profile.age)), + credit_score=int(r.get("credit_score", local_profile.credit_score)), + income_band=r.get("income_band", local_profile.income_band), + annual_income_usd=float(r.get("annual_income_usd", local_profile.annual_income_usd)), + segment=r.get("segment", local_profile.segment), + primary_goals=r.get("primary_goals", local_profile.primary_goals), + risk_tolerance=r.get("risk_tolerance", local_profile.risk_tolerance), + total_spend_usd=float(r.get("total_spend_usd", local_profile.total_spend_usd)), + transaction_count=int(r.get("transaction_count", local_profile.transaction_count)), + max_single_transaction_usd=float(r.get("max_single_transaction_usd", local_profile.max_single_transaction_usd)), + spend_by_category=local_profile.spend_by_category, + top_categories=local_profile.top_categories, + ) + except Exception as e: + logger.warning(f"Dozer query failed ({e}); falling back to local emulation.") + + return self._build_local_profile(customer_id) + + def _load_local_products(self) -> List[CardProduct]: + csv_path = DATA_DIR / "credit_card_products" / "credit_card_products.csv" + with open(csv_path, mode="r", encoding="utf-8") as f: + reader = csv.DictReader(f) + return [ + CardProduct( + product_id=r["product_id"], + name=r["name"], + card_tier=r.get("card_tier", "Standard"), + annual_fee_usd=float(r["annual_fee_usd"]), + min_credit_score=int(r["min_credit_score"]), + min_annual_income_usd=float(r["min_annual_income_usd"]), + rewards_focus=r["rewards_focus"], + intro_bonus=r.get("intro_bonus", "None"), + apr_rate=r.get("apr_rate", "N/A"), + perks_summary=r.get("perks_summary", ""), + target_segment=r.get("target_segment", "General"), + ) + for r in reader + ] + + def _build_local_profile(self, customer_id: str) -> CustomerProfile: + cust_path = DATA_DIR / "customer_profiles" / "customer_profiles.csv" + txn_path = DATA_DIR / "transactions" / "transactions.csv" + + target_cust: Optional[Dict[str, str]] = None + with open(cust_path, mode="r", encoding="utf-8") as f: + for row in csv.DictReader(f): + if row["customer_id"] == customer_id: + target_cust = row + break + + if not target_cust: + raise KeyError(f"Customer ID '{customer_id}' not found in profiles dataset.") + + spend_by_cat: Dict[str, float] = defaultdict(float) + txns: List[Transaction] = [] + with open(txn_path, mode="r", encoding="utf-8") as f: + for row in csv.DictReader(f): + if row["customer_id"] == customer_id: + amt = float(row["amount_usd"]) + cat = row["merchant_category"].strip().lower() + spend_by_cat[cat] += amt + txns.append( + Transaction( + transaction_id=row["transaction_id"], + customer_id=row["customer_id"], + merchant_name=row["merchant_name"], + merchant_category=cat, + amount_usd=amt, + txn_date=row["txn_date"], + ) + ) + + total_spend = sum(spend_by_cat.values()) + max_txn = max([t.amount_usd for t in txns], default=0.0) + top_cats = [cat for cat, _ in Counter(spend_by_cat).most_common(3)] + + return CustomerProfile( + customer_id=customer_id, + first_name=target_cust["first_name"], + last_name=target_cust["last_name"], + age=int(target_cust["age"]), + credit_score=int(target_cust["credit_score"]), + income_band=target_cust["income_band"], + annual_income_usd=float(target_cust["annual_income_usd"]), + segment=target_cust["segment"], + primary_goals=target_cust["primary_goals"], + risk_tolerance=target_cust["risk_tolerance"], + total_spend_usd=round(total_spend, 2), + transaction_count=len(txns), + max_single_transaction_usd=round(max_txn, 2), + spend_by_category=dict(spend_by_cat), + top_categories=top_cats, + ) diff --git a/examples/llm-banking-advisor/guardrails.py b/examples/llm-banking-advisor/guardrails.py new file mode 100644 index 0000000000..1f06d177b8 --- /dev/null +++ b/examples/llm-banking-advisor/guardrails.py @@ -0,0 +1,67 @@ +"""Financial eligibility and bank risk guardrails.""" + +from __future__ import annotations + +from typing import Dict, List, Tuple + +from models import CardProduct, CustomerProfile + + +class FinancialGuardrails: + """Enforces bank underwriting and credit eligibility constraints. + + Hard constraints must be evaluated BEFORE vector retrieval or LLM generation + to prevent recommending products the customer is legally or financially ineligible for. + """ + + @staticmethod + def evaluate_card( + customer: CustomerProfile, card: CardProduct + ) -> Tuple[bool, List[str]]: + """Evaluate a single card against customer financial parameters. + + Returns (is_eligible, reasons). + """ + reasons: List[str] = [] + + # 1. Minimum Age + if customer.age < 18: + reasons.append("Customer must be at least 18 years old.") + + # 2. Credit Score Threshold + if customer.credit_score < card.min_credit_score: + reasons.append( + f"Credit score {customer.credit_score} is below required minimum of {card.min_credit_score}." + ) + + # 3. Minimum Annual Income + if customer.annual_income_usd < card.min_annual_income_usd: + reasons.append( + f"Annual income ${customer.annual_income_usd:,.0f} is below minimum requirement of ${card.min_annual_income_usd:,.0f}." + ) + + # 4. Risk Tolerance / Premium Annual Fee Alignment + if customer.risk_tolerance.lower() == "low" and card.annual_fee_usd > 200: + reasons.append( + f"Annual fee of ${card.annual_fee_usd:.0f} conflicts with conservative (low) risk tolerance." + ) + + is_eligible = len(reasons) == 0 + return is_eligible, reasons + + @classmethod + def filter_eligible_cards( + cls, customer: CustomerProfile, cards: List[CardProduct] + ) -> Tuple[List[CardProduct], Dict[str, List[str]]]: + """Split a card list into eligible cards and ineligible cards with failure reasons.""" + eligible: List[CardProduct] = [] + ineligible_reasons: Dict[str, List[str]] = {} + + for card in cards: + is_ok, reasons = cls.evaluate_card(customer, card) + if is_ok: + eligible.append(card) + else: + ineligible_reasons[card.name] = reasons + + return eligible, ineligible_reasons diff --git a/examples/llm-banking-advisor/models.py b/examples/llm-banking-advisor/models.py new file mode 100644 index 0000000000..97f4a4632e --- /dev/null +++ b/examples/llm-banking-advisor/models.py @@ -0,0 +1,81 @@ +"""Data models for Dozer-backed LLM Banking Advisor.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class Transaction: + transaction_id: str + customer_id: str + merchant_name: str + merchant_category: str + amount_usd: float + txn_date: str + + +@dataclass +class CustomerProfile: + customer_id: str + first_name: str + last_name: str + age: int + credit_score: int + income_band: str + annual_income_usd: float + segment: str + primary_goals: str + risk_tolerance: str + total_spend_usd: float + transaction_count: int + max_single_transaction_usd: float + spend_by_category: Dict[str, float] = field(default_factory=dict) + top_categories: List[str] = field(default_factory=list) + + @property + def full_name(self) -> str: + return f"{self.first_name} {self.last_name}" + + +@dataclass +class CardProduct: + product_id: str + name: str + card_tier: str + annual_fee_usd: float + min_credit_score: int + min_annual_income_usd: float + rewards_focus: str + intro_bonus: str + apr_rate: str + perks_summary: str + target_segment: str + + def to_dict(self) -> Dict[str, Any]: + return { + "product_id": self.product_id, + "name": self.name, + "card_tier": self.card_tier, + "annual_fee_usd": self.annual_fee_usd, + "min_credit_score": self.min_credit_score, + "min_annual_income_usd": self.min_annual_income_usd, + "rewards_focus": self.rewards_focus, + "intro_bonus": self.intro_bonus, + "apr_rate": self.apr_rate, + "perks_summary": self.perks_summary, + "target_segment": self.target_segment, + } + + +@dataclass +class RecommendationResult: + customer_id: str + recommended_product: CardProduct + eligibility_status: str + fit_score: float + rationale: str + spend_alignment: Dict[str, Any] + all_evaluated_cards: List[Dict[str, Any]] + source_provenance: str diff --git a/examples/llm-banking-advisor/requirements.txt b/examples/llm-banking-advisor/requirements.txt new file mode 100644 index 0000000000..9f921729fd --- /dev/null +++ b/examples/llm-banking-advisor/requirements.txt @@ -0,0 +1,6 @@ +langchain>=0.2.0 +langchain-core>=0.2.0 +langchain-community>=0.2.0 +requests>=2.31.0 +pyyaml>=6.0 +pytest>=8.0.0 diff --git a/examples/llm-banking-advisor/tests/__init__.py b/examples/llm-banking-advisor/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/examples/llm-banking-advisor/tests/test_advisor.py b/examples/llm-banking-advisor/tests/test_advisor.py new file mode 100644 index 0000000000..35fa1dd353 --- /dev/null +++ b/examples/llm-banking-advisor/tests/test_advisor.py @@ -0,0 +1,48 @@ +"""End-to-End Integration Tests for Banking Advisor.""" + +import unittest +import sys +from pathlib import Path + +SAMPLE_DIR = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(SAMPLE_DIR)) + +from advisor import BankingAdvisor +from dozer_client import DozerClient + + +class TestBankingAdvisorIntegration(unittest.TestCase): + def setUp(self): + self.client = DozerClient() + self.advisor = BankingAdvisor(dozer_client=self.client) + + def test_recommendation_cust_001_travel(self): + """Elena Rostova -> Voyager Horizon Travel Elite.""" + result = self.advisor.recommend("CUST_001") + self.assertEqual(result.customer_id, "CUST_001") + self.assertEqual(result.recommended_product.product_id, "CARD_001") + self.assertEqual(result.eligibility_status, "VERIFIED_ELIGIBLE") + self.assertIn("Travel", result.rationale) + self.assertIn("FICO", result.rationale) + + def test_recommendation_cust_002_credit_builder(self): + """Marcus Vance (Student, 610 credit) -> Foundation Builder.""" + result = self.advisor.recommend("CUST_002") + self.assertEqual(result.recommended_product.product_id, "CARD_003") + self.assertEqual(result.recommended_product.annual_fee_usd, 0.0) + + def test_recommendation_cust_003_cashback_groceries(self): + """Sarah Jenkins -> Harvest Cash & Groceries Plus.""" + result = self.advisor.recommend("CUST_003") + self.assertEqual(result.recommended_product.product_id, "CARD_002") + self.assertIn("groceries", result.spend_alignment) + + def test_recommendation_cust_005_luxury_executive(self): + """Amina Diallo ($340k income, 790 FICO) -> Centurion Signature Sovereign.""" + result = self.advisor.recommend("CUST_005") + self.assertEqual(result.recommended_product.product_id, "CARD_005") + self.assertEqual(result.recommended_product.card_tier, "Luxury Executive") + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/llm-banking-advisor/tests/test_config.py b/examples/llm-banking-advisor/tests/test_config.py new file mode 100644 index 0000000000..917fd57157 --- /dev/null +++ b/examples/llm-banking-advisor/tests/test_config.py @@ -0,0 +1,49 @@ +"""Tests for Dozer configuration schema and SQL validity.""" + +import unittest +import re +from pathlib import Path + +SAMPLE_DIR = Path(__file__).resolve().parent.parent +CONFIG_PATH = SAMPLE_DIR / "dozer-config.yaml" + + +class TestDozerConfig(unittest.TestCase): + def setUp(self): + self.assertTrue(CONFIG_PATH.exists(), f"Missing config file at {CONFIG_PATH}") + self.content = CONFIG_PATH.read_text(encoding="utf-8") + + def test_required_top_level_fields(self): + """Verify presence of modern Dozer root fields.""" + self.assertRegex(self.content, re.compile(r"^version:\s*1", re.M)) + self.assertRegex(self.content, re.compile(r"^app_name:\s*llm-banking-advisor", re.M)) + self.assertRegex(self.content, re.compile(r"^connections:", re.M)) + self.assertRegex(self.content, re.compile(r"^sources:", re.M)) + self.assertRegex(self.content, re.compile(r"^sql:", re.M)) + self.assertRegex(self.content, re.compile(r"^sinks:", re.M)) + self.assertRegex(self.content, re.compile(r"^api:", re.M)) + + def test_no_obsolete_endpoints_field(self): + """Ensure obsolete 'endpoints' field is absent (causes serde deny_unknown_fields error).""" + self.assertIsNone(re.search(r"^endpoints:", self.content, re.MULTILINE)) + + def test_sources_reference_connections(self): + """Verify sources reference connection bank_data.""" + self.assertIn("connection: bank_data", self.content) + self.assertIn("name: bank_data", self.content) + + def test_streaming_sql_targets(self): + """Verify SQL creates customer_profile and card_catalog target tables.""" + self.assertIn("INTO customer_profile", self.content) + self.assertIn("INTO card_catalog", self.content) + self.assertIn("JOIN transactions", self.content) + self.assertIn("GROUP BY", self.content) + + def test_sinks_match_sql_targets(self): + """Verify declared sinks match the SQL INTO target tables.""" + self.assertIn("table_name: customer_profile", self.content) + self.assertIn("table_name: card_catalog", self.content) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/llm-banking-advisor/tests/test_dozer_client.py b/examples/llm-banking-advisor/tests/test_dozer_client.py new file mode 100644 index 0000000000..93704a5762 --- /dev/null +++ b/examples/llm-banking-advisor/tests/test_dozer_client.py @@ -0,0 +1,51 @@ +"""Tests for DozerClient data ingestion and SQL profile emulation.""" + +import unittest +import sys +from pathlib import Path + +SAMPLE_DIR = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(SAMPLE_DIR)) + +from dozer_client import DozerClient + + +class TestDozerClient(unittest.TestCase): + def setUp(self): + self.client = DozerClient() + + def test_load_products(self): + """Verify card products are ingested with correct types.""" + products = self.client.get_card_catalog() + self.assertGreaterEqual(len(products), 5) + product_ids = [p.product_id for p in products] + self.assertIn("CARD_001", product_ids) + self.assertIn("CARD_002", product_ids) + for p in products: + self.assertGreater(len(p.name), 0) + self.assertGreaterEqual(p.min_credit_score, 500) + self.assertGreaterEqual(p.annual_fee_usd, 0.0) + + def test_customer_profile_aggregation(self): + """Verify customer profile reflects accurate SQL transaction joins and totals.""" + profile = self.client.get_customer_profile("CUST_001") + self.assertEqual(profile.first_name, "Elena") + self.assertEqual(profile.last_name, "Rostova") + self.assertEqual(profile.credit_score, 750) + self.assertEqual(profile.annual_income_usd, 115000) + + # Elena transactions: 840.50 + 420.00 + 38.50 + 610.20 + 12.75 + 95.00 = 2016.95 + self.assertAlmostEqual(profile.total_spend_usd, 2016.95, places=2) + self.assertEqual(profile.transaction_count, 6) + self.assertEqual(profile.top_categories[0], "travel") + self.assertIn("travel", profile.spend_by_category) + self.assertIn("dining", profile.spend_by_category) + + def test_unknown_customer_raises_keyerror(self): + """Verify non-existent customer raises KeyError.""" + with self.assertRaises(KeyError): + self.client.get_customer_profile("NON_EXISTENT_ID") + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/llm-banking-advisor/tests/test_guardrails.py b/examples/llm-banking-advisor/tests/test_guardrails.py new file mode 100644 index 0000000000..d31566d915 --- /dev/null +++ b/examples/llm-banking-advisor/tests/test_guardrails.py @@ -0,0 +1,125 @@ +"""Tests for Financial Guardrails and Underwriting Rules.""" + +import unittest +import sys +from pathlib import Path + +SAMPLE_DIR = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(SAMPLE_DIR)) + +from guardrails import FinancialGuardrails +from models import CardProduct, CustomerProfile + + +class TestFinancialGuardrails(unittest.TestCase): + def setUp(self): + self.premium_card = CardProduct( + product_id="TEST_PREM", + name="Platinum Elite", + card_tier="Premium", + annual_fee_usd=450.0, + min_credit_score=750, + min_annual_income_usd=100000.0, + rewards_focus="Travel", + intro_bonus="50k pts", + apr_rate="18%", + perks_summary="Lounges", + target_segment="Affluent", + ) + self.starter_card = CardProduct( + product_id="TEST_START", + name="Starter Cash", + card_tier="Basic", + annual_fee_usd=0.0, + min_credit_score=580, + min_annual_income_usd=20000.0, + rewards_focus="Cashback", + intro_bonus="None", + apr_rate="24%", + perks_summary="1% cashback", + target_segment="Students", + ) + + def test_eligible_customer_passes(self): + customer = CustomerProfile( + customer_id="C_OK", + first_name="Alice", + last_name="Smith", + age=32, + credit_score=780, + income_band="high", + annual_income_usd=130000.0, + segment="Professional", + primary_goals="Travel", + risk_tolerance="high", + total_spend_usd=5000.0, + transaction_count=10, + max_single_transaction_usd=1000.0, + ) + is_ok, reasons = FinancialGuardrails.evaluate_card(customer, self.premium_card) + self.assertTrue(is_ok) + self.assertEqual(len(reasons), 0) + + def test_underage_rejection(self): + customer = CustomerProfile( + customer_id="C_MINOR", + first_name="Minor", + last_name="Doe", + age=17, + credit_score=700, + income_band="low", + annual_income_usd=25000.0, + segment="Student", + primary_goals="Credit", + risk_tolerance="low", + total_spend_usd=100.0, + transaction_count=2, + max_single_transaction_usd=50.0, + ) + is_ok, reasons = FinancialGuardrails.evaluate_card(customer, self.starter_card) + self.assertFalse(is_ok) + self.assertTrue(any("18 years old" in r for r in reasons)) + + def test_low_credit_score_rejection(self): + customer = CustomerProfile( + customer_id="C_LOW_FICO", + first_name="Bob", + last_name="Ray", + age=28, + credit_score=680, + income_band="middle", + annual_income_usd=120000.0, + segment="Tech", + primary_goals="Travel", + risk_tolerance="medium", + total_spend_usd=2000.0, + transaction_count=5, + max_single_transaction_usd=400.0, + ) + is_ok, reasons = FinancialGuardrails.evaluate_card(customer, self.premium_card) + self.assertFalse(is_ok) + self.assertTrue(any("Credit score" in r for r in reasons)) + + def test_conservative_risk_tolerance_filters_high_fee(self): + customer = CustomerProfile( + customer_id="C_CONSERVATIVE", + first_name="Clara", + last_name="Oswald", + age=45, + credit_score=800, + income_band="affluent", + annual_income_usd=150000.0, + segment="Family", + primary_goals="Groceries", + risk_tolerance="low", + total_spend_usd=3000.0, + transaction_count=8, + max_single_transaction_usd=500.0, + ) + is_ok, reasons = FinancialGuardrails.evaluate_card(customer, self.premium_card) + self.assertFalse(is_ok) + self.assertTrue(any("conservative (low) risk tolerance" in r for r in reasons)) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/llm-banking-advisor/tests/test_vector_rag.py b/examples/llm-banking-advisor/tests/test_vector_rag.py new file mode 100644 index 0000000000..a6e673ea65 --- /dev/null +++ b/examples/llm-banking-advisor/tests/test_vector_rag.py @@ -0,0 +1,52 @@ +"""Tests for LangChain Vector RAG and Semantic Retrieval.""" + +import unittest +import sys +from pathlib import Path + +SAMPLE_DIR = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(SAMPLE_DIR)) + +from dozer_client import DozerClient +from guardrails import FinancialGuardrails +from vector_rag import DozerGroundedRetriever, LocalSemanticEmbedder + + +class TestVectorRAG(unittest.TestCase): + def setUp(self): + self.client = DozerClient() + self.cards = self.client.get_card_catalog() + + def test_semantic_embedder_properties(self): + """Verify embedding dimension and unit length normalization.""" + embedder = LocalSemanticEmbedder(dim=64) + emb = embedder.embed("travel flight airport lounge") + self.assertEqual(len(emb), 64) + norm = sum(x * x for x in emb) + self.assertAlmostEqual(norm, 1.0, places=4) + + def test_retriever_ranks_travel_for_traveler(self): + """Verify traveler profile receives top ranking for travel card among eligible products.""" + customer = self.client.get_customer_profile("CUST_001") + eligible_cards, _ = FinancialGuardrails.filter_eligible_cards(customer, self.cards) + retriever = DozerGroundedRetriever(eligible_cards) + ranked = retriever.rank_cards_for_customer(customer, user_query="international flight perks") + self.assertGreater(len(ranked), 0) + top_card, top_score, meta = ranked[0] + self.assertEqual(top_card.product_id, "CARD_001") + self.assertGreater(meta["spend_bonus"], 0.0) + self.assertIn("travel", meta["spend_matches"]) + + def test_retriever_ranks_grocery_card_for_family_shopper(self): + """Verify grocery/supermarket spend elevates grocery card among eligible products.""" + customer = self.client.get_customer_profile("CUST_003") + eligible_cards, _ = FinancialGuardrails.filter_eligible_cards(customer, self.cards) + retriever = DozerGroundedRetriever(eligible_cards) + ranked = retriever.rank_cards_for_customer(customer) + top_card, top_score, meta = ranked[0] + self.assertEqual(top_card.product_id, "CARD_002") + self.assertIn("groceries", meta["spend_matches"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/llm-banking-advisor/vector_rag.py b/examples/llm-banking-advisor/vector_rag.py new file mode 100644 index 0000000000..661c1b44fc --- /dev/null +++ b/examples/llm-banking-advisor/vector_rag.py @@ -0,0 +1,115 @@ +"""LangChain Vector RAG and semantic catalog search for Dozer Banking Advisor.""" + +from __future__ import annotations + +import hashlib +import math +from typing import Any, Dict, List, Optional, Tuple + +from models import CardProduct, CustomerProfile + + +class LocalSemanticEmbedder: + """Deterministic token hash embedder for offline / dependency-free vector indexing.""" + + def __init__(self, dim: int = 128) -> None: + self.dim = dim + + def embed(self, text: str) -> List[float]: + vec = [0.0] * self.dim + tokens = text.lower().replace(",", " ").replace(";", " ").replace(".", " ").split() + for tok in tokens: + h = hashlib.sha256(tok.encode("utf-8")).digest() + idx = int.from_bytes(h[:4], byteorder="big") % self.dim + vec[idx] += 1.0 + norm = math.sqrt(sum(x * x for x in vec)) or 1.0 + return [x / norm for x in vec] + + def embed_documents(self, texts: List[str]) -> List[List[float]]: + return [self.embed(t) for t in texts] + + def embed_query(self, text: str) -> List[float]: + return self.embed(text) + + +def _cosine_similarity(a: List[float], b: List[float]) -> float: + return sum(x * y for x, y in zip(a, b)) + + +class DozerGroundedRetriever: + """LangChain-compatible semantic catalog retriever with transaction category alignment.""" + + def __init__(self, cards: List[CardProduct]) -> None: + self.cards = cards + self.embedder = LocalSemanticEmbedder() + self._doc_embeddings: List[Tuple[CardProduct, List[float], str]] = [] + self._index_cards() + + def _card_to_document_text(self, card: CardProduct) -> str: + return ( + f"Card: {card.name}. Tier: {card.card_tier}. " + f"Rewards Focus: {card.rewards_focus}. " + f"Annual Fee: ${card.annual_fee_usd:.0f}. " + f"Intro Bonus: {card.intro_bonus}. " + f"Key Perks: {card.perks_summary}. " + f"Target Audience: {card.target_segment}." + ) + + def _index_cards(self) -> None: + self._doc_embeddings = [] + for card in self.cards: + text = self._card_to_document_text(card) + emb = self.embedder.embed(text) + self._doc_embeddings.append((card, emb, text)) + + def rank_cards_for_customer( + self, + customer: CustomerProfile, + user_query: Optional[str] = None, + top_k: int = 3, + ) -> List[Tuple[CardProduct, float, Dict[str, Any]]]: + """Rank eligible cards using hybrid scoring: + + 1. Semantic Query Similarity (from user question/goal) + 2. Transaction Spend Alignment (from Dozer real-time category aggregation) + 3. Stated Goal Match + """ + query_text = user_query or customer.primary_goals + q_emb = self.embedder.embed(f"{query_text} {customer.segment} {customer.primary_goals}") + + results: List[Tuple[CardProduct, float, Dict[str, Any]]] = [] + + for card, doc_emb, doc_text in self._doc_embeddings: + # 1. Semantic similarity + semantic_score = _cosine_similarity(q_emb, doc_emb) + + # 2. Spend Alignment Bonus + spend_matches: Dict[str, float] = {} + total_spend = customer.total_spend_usd or 1.0 + spend_bonus = 0.0 + + text_lower = doc_text.lower() + for cat, amt in customer.spend_by_category.items(): + if cat in text_lower or (cat == "dining" and "restaurant" in text_lower): + pct = amt / total_spend + spend_bonus += pct * 2.0 # proportional boost for where money is spent + spend_matches[cat] = amt + + # 3. Goal Alignment + goal_bonus = 0.0 + for goal_word in customer.primary_goals.lower().split(): + if len(goal_word) > 3 and goal_word in text_lower: + goal_bonus += 0.25 + + final_score = (semantic_score * 1.5) + spend_bonus + goal_bonus + + details = { + "semantic_similarity": round(semantic_score, 3), + "spend_bonus": round(spend_bonus, 3), + "spend_matches": spend_matches, + "document_summary": doc_text, + } + results.append((card, final_score, details)) + + results.sort(key=lambda x: x[1], reverse=True) + return results[:top_k]