feat: add Dozer + LangChain + Vector RAG banking advisor sample (fixes #1690)

examples/llm-banking-advisor/.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+*.pyc
+.pytest_cache/
+.chroma/
+

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.

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}"
+        )