Add samples/llm-chatbot: Dozer + LLM + Vector DB + Langchain sample (#1690)

samples/llm-chatbot/.gitignore

@@ -0,0 +1,5 @@
+.dozer/
+.env
+__pycache__/
+*.pyc
+qdrant_storage/

samples/llm-chatbot/README.md

@@ -0,0 +1,146 @@
+# Dozer + LLM + Vector DB + Langchain: hyper-personalized banking chatbot
+
+A complete working sample accompanying the article
+[*Hyper-personalization of LLM Chatbot using Dozer*](https://getdozer.io/blog/llm-chatbot)
+(issue #1690).
+
+A hypothetical bank wants an LLM chatbot that knows each customer. Dozer
+sources the bank's operational datasets — **customer profiles**,
+**transactions** and a **credit-card product catalog** — from Postgres,
+joins them in real time into a **unified customer profile**, and serves it
+over a low-latency REST/gRPC API. A Langchain chatbot then:
+
+1. pulls the caller's unified profile from **Dozer** on every turn,
+2. retrieves the most relevant credit-card products from a **Qdrant**
+   vector store (RAG over the product catalog),
+3. feeds profile + transactions + retrieved products to the **LLM** as
+   context (OpenAI, or a built-in mock when no API key is set).
+
+```
+                    ┌────────────────────────────────────────────┐
+                    │                  Dozer                     │
+ Postgres ─────────►│  customers + transactions + products       │
+ (raw datasets)     │  ── SQL join & aggregate ──►               │────► REST :8080
+                    │  customer_profile (unified, real-time)     │────► gRPC :50051
+                    └────────────────────────────────────────────┘
+                                                        │
+                              ┌─────────────────────────┴──────────┐
+                              │           Langchain chatbot        │
+   Qdrant ◄── index products ─│  profile + RAG products → LLM      │──► "Avery" assistant
+   (vector store) ◄─ search ─ │  (OpenAI or keyless mock LLM)      │
+                              └────────────────────────────────────┘
+```
+
+## Quick start
+
+```bash
+export OPENAI_API_KEY=sk-...   # optional - omit to run with the mock LLM
+docker compose up --build
+```
+
+Then chat:
+
+```bash
+curl -s localhost:8000/chat \
+  -H 'Content-Type: application/json' \
+  -d '{"customer_id": 1001, "message": "I travel often - which card fits me best?"}'
+```
+
+Interactive terminal mode:
+
+```bash
+docker compose exec chatbot python main.py --cli --customer-id 1001
+```
+
+Inspect the data APIs Dozer generates:
+
+```bash
+curl -s localhost:8080/                     # all endpoint paths
+curl -s localhost:8080/health               # {"status":"SERVING"}
+curl -s localhost:8080/transactions/1       # single transaction by primary key
+curl -s localhost:8080/transactions         # list
+curl -s localhost:8080/profile/query \
+  -H 'Content-Type: application/json' \
+  -d '{"$filter": {"customer_id": 1001}, "$limit": 1}'   # unified profile
+curl -s localhost:8080/transactions/query \
+  -H 'Content-Type: application/json' \
+  -d '{"$filter": {"customer_id": 1003}, "$order_by": {"amount": "desc"}, "$limit": 5}'
+```
+
+Without `OPENAI_API_KEY` everything still runs end-to-end: the chatbot
+answers with a deterministic mock LLM and hashing embeddings, so the sample
+is CI-friendly and needs zero secrets.
+
+## What each piece does
+
+| File | Purpose |
+| --- | --- |
+| `docker-compose.yml` | Postgres + Qdrant + Dozer + chatbot, one command |
+| `dozer-config.yaml` | Dozer pipeline: Postgres sources, SQL join, API endpoints |
+| `seed/init.sql` | Customer profiles, transactions, credit-card catalog |
+| `pg_hba.conf` | Replication-friendly auth for Dozer's Postgres CDC |
+| `chatbot/dozer_client.py` | REST client for Dozer's generated endpoints |
+| `chatbot/vector_store.py` | Qdrant index + Langchain embeddings (OpenAI or hashing) |
+| `chatbot/chatbot.py` | Prompt assembly: unified profile + RAG + history → LLM |
+| `chatbot/main.py` | FastAPI app and CLI entrypoint |
+
+## The Dozer pipeline
+
+`dozer-config.yaml` ingests three tables and builds the unified profile with
+plain SQL — Dozer materializes and keeps it up to date as transactions land:
+
+```sql
+select c.customer_id, c.first_name, ..., count(t.txn_id) as transaction_count,
+       sum(t.amount) as total_spend, max(t.txn_date) as last_transaction_date
+into customer_profile
+from customers c
+inner join transactions t on t.customer_id = c.customer_id
+group by c.customer_id, ...;
+```
+
+Each `endpoints` entry becomes a REST + gRPC API on :8080/:50051 with
+`/{id}` fetch, list, `/query` and `/count` routes generated automatically.
+
+## Running without Docker
+
+Requirements: a local `dozer` binary, Python 3.10+, a Postgres with the seed
+loaded (`wal_level=logical`), and Qdrant.
+
+```bash
+# 1. Load seed data and start dependencies
+psql postgresql://postgres:postgres@localhost:5436/bank -f seed/init.sql
+
+# 2. Point Dozer at localhost (edit host in dozer-config.yaml) and run
+dozer run
+
+# 3. Chatbot
+cd chatbot
+pip install -r requirements.txt
+export DOZER_REST_URL=http://localhost:8080
+export QDRANT_URL=http://localhost:6333
+uvicorn main:app --port 8000      # or: python main.py --cli --customer-id 1002
+```
+
+## Verify
+
+With the stack up (compose or native), run the end-to-end smoke tests:
+
+```bash
+pip install pytest
+DOZER_REST_URL=http://localhost:8080 QDRANT_URL=http://localhost:6333 \
+  PYTHONPATH=chatbot pytest tests/test_sample.py -v
+```
+
+The tests exercise the full keyless chain: Dozer health, the unified profile
+(join + aggregation), the query DSL (`$filter` + `$order_by`), and the
+chatbot (Qdrant indexing → retrieval → personalized mock-LLM reply).
+
+## Configuration
+
+| Env var | Default | Description |
+| --- | --- | --- |
+| `OPENAI_API_KEY` | _(empty)_ | Enables real LLM + OpenAI embeddings; mock fallback when unset |
+| `OPENAI_MODEL` | `gpt-4o-mini` | Chat model |
+| `OPENAI_BASE_URL` | _(empty)_ | Point at any OpenAI-compatible server (vLLM, Ollama, ...) |
+| `DOZER_REST_URL` | `http://localhost:8080` | Dozer REST API |
+| `QDRANT_URL` | `http://localhost:6333` | Qdrant vector store |

samples/llm-chatbot/chatbot/Dockerfile

@@ -0,0 +1,10 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8000

samples/llm-chatbot/chatbot/chatbot.py

@@ -0,0 +1,187 @@
+"""Hyper-personalized banking chatbot.
+
+Per turn it:
+  1. loads the caller's unified customer profile from Dozer (REST),
+  2. retrieves the matching credit-card products from Qdrant (RAG),
+  3. prompts the LLM with profile + retrieved products + chat history.
+
+Works with OpenAI when OPENAI_API_KEY is set. Without a key it falls back
+to a deterministic mock LLM that renders the same context template, so the
+whole sample runs end-to-end with zero secrets (CI-friendly).
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.runnables.history import RunnableWithMessageHistory
+from langchain_core.chat_history import InMemoryChatMessageHistory
+
+from dozer_client import DozerClient
+from vector_store import ProductVectorStore
+
+SYSTEM_TEMPLATE = """\
+You are Avery, the virtual banking assistant. You hyper-personalize every \
+answer using the customer's unified profile and only recommend products \
+from the retrieved catalog.
+
+# Unified customer profile (served by Dozer)
+{profile}
+
+# Recent transactions (served by Dozer)
+{transactions}
+
+Products: (retrieved via vector search over the catalog)
+{products}
+
+Rules:
+- Ground every statement in the profile, the transactions or the catalog.
+- Recommend at most two products and explain why they fit this customer.
+- If the customer asks about something the context does not cover, say so \
+plainly instead of inventing details.
+"""
+
+
+def _format_block(label: str, payload: Any) -> str:
+    if not payload:
+        return f"{label}: (none)"
+    if isinstance(payload, str):
+        return f"{label}: {payload}"
+    if isinstance(payload, dict):
+        lines = [f"  {k}: {v}" for k, v in payload.items()]
+        return f"{label}:\n" + "\n".join(lines)
+    lines = []
+    for item in payload:
+        if isinstance(item, dict):
+            lines.append("  - " + ", ".join(f"{k}: {v}" for k, v in item.items()))
+        else:
+            lines.append(f"  - {item}")
+    return f"{label}:\n" + "\n".join(lines)
+
+
+def build_context(dozer: DozerClient, products: ProductVectorStore, customer_id: int, question: str):
+    """Pull the unified profile from Dozer and RAG-match products."""
+    profile = dozer.get_profile(customer_id)
+    if profile is None:
+        customer = dozer.get_customer(customer_id)
+        if customer is None:
+            return None, None, None
+        profile = customer
+    transactions = dozer.get_transactions(customer_id)
+    matches = products.search(question, k=2)
+    return profile, transactions, matches
+
+
+def get_llm():
+    """OpenAI chat model when a key is present, mock otherwise."""
+    if os.getenv("OPENAI_API_KEY"):
+        from langchain_openai import ChatOpenAI
+
+        return ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), temperature=0.3)
+    return MockLLM()
+
+
+class MockLLM:
+    """Deterministic stand-in for CI / keyless demos.
+
+    Renders the same prompt the real model would see, so the plumbing
+    (Dozer profile -> vector retrieval -> prompt) is fully exercised
+    without network access or secrets. The reply is assembled from the
+    retrieved products and the profile segment.
+    """
+
+    def bind(self, **_kwargs):  # parity with chat models
+        return self
+
+    def with_config(self, **_kwargs):
+        return self
+
+    def invoke(self, messages, config=None, **_kwargs):
+        content = "\n".join(
+            m.content if isinstance(m.content, str) else str(m.content) for m in messages
+        )
+        lines = content.splitlines()
+
+        name, segment = "customer", "unknown"
+        for line in lines:
+            stripped = line.strip()
+            if stripped.startswith("first_name:") and name == "customer":
+                name = stripped.split(":", 1)[1].strip() or "customer"
+            if stripped.startswith("segment:"):
+                segment = stripped.split(":", 1)[1].strip() or "unknown"
+
+        product_names: list[str] = []
+        for idx, line in enumerate(lines):
+            if line.strip().startswith("Products:") or line.strip() == "Products":
+                for product_line in lines[idx + 1 :]:
+                    if not product_line.strip().startswith("- "):
+                        break
+                    product_names.append(product_line.strip()[2:].split(":")[0].strip())
+                    if len(product_names) == 2:
+                        break
+                break
+
+        if product_names:
+            rec_block = "\n".join(f"- {p}" for p in product_names)
+        else:
+            rec_block = "- (no matching products indexed yet)"
+
+        text = (
+            f"Hi {name}! As a {segment}-segment customer, based on your unified "
+            f"profile and recent activity, these picks from our catalog fit you best:\n"
+            f"{rec_block}\n\n"
+            "(Mock LLM response - set OPENAI_API_KEY for real answers.)"
+        )
+        return MockMessage(text)
+
+
+class MockMessage:
+    def __init__(self, content: str) -> None:
+        self.content = content
+
+
+_PROMPT = ChatPromptTemplate.from_messages(
+    [
+        ("system", SYSTEM_TEMPLATE),
+        MessagesPlaceholder("history"),
+        ("human", "{question}"),
+    ]
+)
+
+
+def answer(
+    dozer: DozerClient,
+    products: ProductVectorStore,
+    customer_id: int,
+    question: str,
+    history: InMemoryChatMessageHistory | None = None,
+) -> str:
+    profile, transactions, matches = build_context(dozer, products, customer_id, question)
+    if profile is None:
+        return f"Customer {customer_id} not found in Dozer. Try customer_id 1001-1005."
+
+    llm = get_llm()
+    prompt_value = _PROMPT.invoke(
+        {
+            "profile": _format_block("Profile", profile),
+            "transactions": _format_block("Recent transactions", transactions),
+            "products": (
+                "\n".join(f"- {m['name']}: {m['text']}" for m in matches) or "(none)"
+            ),
+            "history": list(history.messages) if history else [],
+            "question": question,
+        }
+    )
+    response = llm.invoke(prompt_value.to_messages())
+    return response.content
+
+
+_SESSIONS: dict[str, InMemoryChatMessageHistory] = {}
+
+
+def session_history(session_id: str) -> InMemoryChatMessageHistory:
+    if session_id not in _SESSIONS:
+        _SESSIONS[session_id] = InMemoryChatMessageHistory()
+    return _SESSIONS[session_id]