From defc95b9cbd195bd18dee519baadf6569ea3dce2 Mon Sep 17 00:00:00 2001 From: CallumJB125 Date: Sun, 4 Oct 2026 16:45:39 +0200 Subject: [PATCH] Add samples/llm-chatbot: Dozer + LLM + vector DB + Langchain sample Complete working sample for #1690, based on the hyper-personalization article (getdozer.io/blog/llm-chatbot): - docker-compose.yml: Postgres (seeded customer profiles, transactions, credit-card products) + Qdrant (vector store) + Dozer + chatbot - dozer-config.yaml: ingests the three datasets, builds a unified customer profile via SQL join + aggregation, serves REST/gRPC - chatbot/: Langchain app - Dozer REST client, Qdrant RAG over the product catalog, prompt assembly, FastAPI + CLI entrypoints - Works keyless (mock LLM + hashing embeddings) for CI; real LLM with OPENAI_API_KEY - tests/: end-to-end smoke tests over the full chain --- samples/llm-chatbot/.gitignore | 5 + samples/llm-chatbot/README.md | 146 +++++++++++++++ samples/llm-chatbot/chatbot/Dockerfile | 10 + samples/llm-chatbot/chatbot/chatbot.py | 187 +++++++++++++++++++ samples/llm-chatbot/chatbot/dozer_client.py | 98 ++++++++++ samples/llm-chatbot/chatbot/main.py | 96 ++++++++++ samples/llm-chatbot/chatbot/requirements.txt | 8 + samples/llm-chatbot/chatbot/vector_store.py | 118 ++++++++++++ samples/llm-chatbot/docker-compose.yml | 102 ++++++++++ samples/llm-chatbot/dozer-config.yaml | 102 ++++++++++ samples/llm-chatbot/pg_hba.conf | 8 + samples/llm-chatbot/seed/init.sql | 77 ++++++++ samples/llm-chatbot/tests/test_sample.py | 80 ++++++++ 13 files changed, 1037 insertions(+) create mode 100644 samples/llm-chatbot/.gitignore create mode 100644 samples/llm-chatbot/README.md create mode 100644 samples/llm-chatbot/chatbot/Dockerfile create mode 100644 samples/llm-chatbot/chatbot/chatbot.py create mode 100644 samples/llm-chatbot/chatbot/dozer_client.py create mode 100644 samples/llm-chatbot/chatbot/main.py create mode 100644 samples/llm-chatbot/chatbot/requirements.txt create mode 100644 samples/llm-chatbot/chatbot/vector_store.py create mode 100644 samples/llm-chatbot/docker-compose.yml create mode 100644 samples/llm-chatbot/dozer-config.yaml create mode 100644 samples/llm-chatbot/pg_hba.conf create mode 100644 samples/llm-chatbot/seed/init.sql create mode 100644 samples/llm-chatbot/tests/test_sample.py diff --git a/samples/llm-chatbot/.gitignore b/samples/llm-chatbot/.gitignore new file mode 100644 index 0000000000..8b131d33c0 --- /dev/null +++ b/samples/llm-chatbot/.gitignore @@ -0,0 +1,5 @@ +.dozer/ +.env +__pycache__/ +*.pyc +qdrant_storage/ diff --git a/samples/llm-chatbot/README.md b/samples/llm-chatbot/README.md new file mode 100644 index 0000000000..3d727a10d1 --- /dev/null +++ b/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 | diff --git a/samples/llm-chatbot/chatbot/Dockerfile b/samples/llm-chatbot/chatbot/Dockerfile new file mode 100644 index 0000000000..f077a3ccf8 --- /dev/null +++ b/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 diff --git a/samples/llm-chatbot/chatbot/chatbot.py b/samples/llm-chatbot/chatbot/chatbot.py new file mode 100644 index 0000000000..6504281c8a --- /dev/null +++ b/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] diff --git a/samples/llm-chatbot/chatbot/dozer_client.py b/samples/llm-chatbot/chatbot/dozer_client.py new file mode 100644 index 0000000000..2347c6d0db --- /dev/null +++ b/samples/llm-chatbot/chatbot/dozer_client.py @@ -0,0 +1,98 @@ +"""Unified customer profile client for the Dozer REST API. + +Dozer turns the `customer_profile` SQL transform (see ../dozer-config.yaml) +into a low-latency REST endpoint. Every chatbot turn starts by pulling the +caller's profile from here, so the LLM always reasons over fresh, joined +customer data instead of a stale snapshot. + +Endpoints served by Dozer (see ../dozer-config.yaml): + GET /profile -> list profiles (default limit 50) + GET /profile/{id} -> single profile by primary key + POST /profile/query -> {"$filter": {...}, "$limit": n, "$skip": n} + POST /profile/count +""" + +from __future__ import annotations + +import os +from typing import Any + +import requests + +DOZER_REST_URL = os.getenv("DOZER_REST_URL", "http://localhost:8080") +_TIMEOUT = 10 + + +class DozerClient: + """Thin client over Dozer's generated REST endpoints.""" + + def __init__(self, base_url: str | None = None) -> None: + self.base_url = (base_url or DOZER_REST_URL).rstrip("/") + + def health(self) -> bool: + try: + response = requests.get(f"{self.base_url}/health", timeout=_TIMEOUT) + payload = response.json() + except (requests.RequestException, ValueError): + return False + return response.ok and payload.get("status") == "SERVING" + + def list_endpoints(self) -> list[str]: + """GET / returns the paths of all endpoints defined in the config.""" + try: + response = requests.get(self.base_url, timeout=_TIMEOUT) + except requests.RequestException: + return [] + return [ + entry if isinstance(entry, str) else entry.get("path", "") + for entry in response.json() + ] + + # ------------------------------------------------------------------ + # Generic helpers over Dozer's endpoint routes. + # ------------------------------------------------------------------ + def _list(self, endpoint: str) -> list[dict[str, Any]]: + response = requests.get(f"{self.base_url}/{endpoint}", timeout=_TIMEOUT) + response.raise_for_status() + return response.json() + + def _get(self, endpoint: str, record_id: Any) -> dict[str, Any] | None: + response = requests.get(f"{self.base_url}/{endpoint}/{record_id}", timeout=_TIMEOUT) + if response.status_code == 404: + return None + response.raise_for_status() + return response.json() + + def _query(self, endpoint: str, body: dict[str, Any]) -> list[dict[str, Any]]: + response = requests.post( + f"{self.base_url}/{endpoint}/query", json=body, timeout=_TIMEOUT + ) + response.raise_for_status() + return response.json() + + # ------------------------------------------------------------------ + # Sample-specific accessors. + # ------------------------------------------------------------------ + def get_customer(self, customer_id: int) -> dict[str, Any] | None: + return self._get("customers", customer_id) + + def get_transactions(self, customer_id: int, limit: int = 5) -> list[dict[str, Any]]: + return self._query( + "transactions", + { + "$filter": {"customer_id": customer_id}, + "$order_by": {"txn_date": "desc"}, + "$limit": limit, + }, + ) + + def get_products(self) -> list[dict[str, Any]]: + return self._list("products") + + def get_profile(self, customer_id: int) -> dict[str, Any] | None: + """The unified profile produced by Dozer's SQL join + aggregation.""" + records = self._query("profile", {"$filter": {"customer_id": customer_id}, "$limit": 1}) + return records[0] if records else None + + def all_profiles(self) -> list[dict[str, Any]]: + return self._list("profile") diff --git a/samples/llm-chatbot/chatbot/main.py b/samples/llm-chatbot/chatbot/main.py new file mode 100644 index 0000000000..9d7cd670ae --- /dev/null +++ b/samples/llm-chatbot/chatbot/main.py @@ -0,0 +1,96 @@ +"""FastAPI wrapper + CLI around the chatbot. + +REST: POST /chat {"customer_id": 1001, "message": "..."} +CLI: python main.py --cli --customer-id 1001 +""" + +from __future__ import annotations + +import argparse +import os + +from fastapi import FastAPI +from pydantic import BaseModel + +from chatbot import answer, session_history +from dozer_client import DozerClient +from vector_store import ProductVectorStore + +app = FastAPI(title="Dozer LLM chatbot sample", version="1.0.0") +dozer = DozerClient() +products: ProductVectorStore | None = None + + +class ChatRequest(BaseModel): + customer_id: int + message: str + session_id: str = "default" + + +class ChatResponse(BaseModel): + customer_id: int + reply: str + products_used: list[str] = [] + + +@app.on_event("startup") +def startup() -> None: + global products + products = ProductVectorStore() + indexed = products.index_products(dozer.get_products()) + print(f"[chatbot] indexed {indexed} products from Dozer into Qdrant") + + +@app.post("/chat", response_model=ChatResponse) +def chat(request: ChatRequest) -> ChatResponse: + assert products is not None + history = session_history(request.session_id) + matches = products.search(request.message, k=2) + reply = answer(dozer, products, request.customer_id, request.message, history) + history.add_user_message(request.message) + history.add_ai_message(reply) + return ChatResponse( + customer_id=request.customer_id, + reply=reply, + products_used=[m["name"] for m in matches], + ) + + +@app.get("/health") +def health() -> dict: + return {"dozer": dozer.health(), "chatbot": "ok"} + + +def cli() -> None: + parser = argparse.ArgumentParser(description="Dozer + Langchain banking chatbot") + parser.add_argument("--cli", action="store_true", help="interactive terminal chat") + parser.add_argument("--customer-id", type=int, default=1001) + args = parser.parse_args() + + global products + products = ProductVectorStore() + indexed = products.index_products(dozer.get_products()) + print(f"[chatbot] indexed {indexed} products from Dozer into Qdrant") + + session = session_history(f"cli-{args.customer_id}") + print(f"Chatting as customer {args.customer_id}. Type 'quit' to exit.") + while True: + try: + question = input("you> ").strip() + except (EOFError, KeyboardInterrupt): + break + if not question or question.lower() in {"quit", "exit"}: + break + reply = answer(dozer, products, args.customer_id, question, session) + session.add_user_message(question) + session.add_ai_message(reply) + print(f"avery> {reply}\n") + + +if __name__ == "__main__": + if os.getenv("CHATBOT_CLI") or "--cli" in os.sys.argv: + cli() + else: + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/samples/llm-chatbot/chatbot/requirements.txt b/samples/llm-chatbot/chatbot/requirements.txt new file mode 100644 index 0000000000..dabfb905bf --- /dev/null +++ b/samples/llm-chatbot/chatbot/requirements.txt @@ -0,0 +1,8 @@ +langchain>=0.3,<0.4 +langchain-openai>=0.2,<0.4 +langchain-community>=0.3,<0.4 +qdrant-client>=1.9 +requests>=2.31 +fastapi>=0.110 +uvicorn>=0.29 +pydantic>=2.6 diff --git a/samples/llm-chatbot/chatbot/vector_store.py b/samples/llm-chatbot/chatbot/vector_store.py new file mode 100644 index 0000000000..beec230ec2 --- /dev/null +++ b/samples/llm-chatbot/chatbot/vector_store.py @@ -0,0 +1,118 @@ +"""Qdrant vector-store layer: embeds the credit-card catalog for RAG. + +The product catalog is fetched from Dozer's REST API, embedded and stored in +Qdrant. At query time the most semantically similar products are retrieved +and passed to the LLM as extra context (retrieval-augmented generation). + +Embeddings use OpenAI when OPENAI_API_KEY is set; otherwise a deterministic +hashing embedder keeps the sample fully runnable offline (CI, air-gapped +demo environments). +""" + +from __future__ import annotations + +import hashlib +import os +from typing import Any + +from langchain_core.documents import Document +from langchain_core.embeddings import Embeddings +from langchain_openai import OpenAIEmbeddings +from qdrant_client import QdrantClient +from qdrant_client.models import Distance, VectorParams + +QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333") +EMBED_DIM = 256 +COLLECTION = "credit_card_products" + + +class HashingEmbeddings(Embeddings): + """Deterministic, dependency-free embeddings for offline/CI runs. + + Bags-of-token-hashes projected onto a fixed-size vector: good enough to + cluster product descriptions by vocabulary overlap without any model + download, and stable across runs so tests are reproducible. + """ + + def __init__(self, dim: int = EMBED_DIM) -> None: + self.dim = dim + + def _embed(self, text: str) -> list[float]: + vec = [0.0] * self.dim + for token in text.lower().replace(",", " ").replace(".", " ").split(): + digest = hashlib.sha256(token.encode()).digest() + index = int.from_bytes(digest[:4], "big") % self.dim + sign = 1.0 if digest[4] % 2 else -1.0 + vec[index] += sign + norm = sum(v * v for v in vec) ** 0.5 or 1.0 + return [v / norm for v 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 get_embeddings() -> Embeddings: + if os.getenv("OPENAI_API_KEY"): + return OpenAIEmbeddings() + return HashingEmbeddings() + + +class ProductVectorStore: + """Thin Langchain-compatible wrapper around Qdrant for the catalog.""" + + def __init__(self) -> None: + self.client = QdrantClient(url=QDRANT_URL, timeout=10) + self.embeddings = get_embeddings() + self._ensure_collection() + + def _ensure_collection(self) -> None: + if not self.client.collection_exists(COLLECTION): + self.client.create_collection( + collection_name=COLLECTION, + vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.COSINE), + ) + + def index_products(self, products: list[dict[str, Any]]) -> int: + """Embed product descriptions from Dozer into Qdrant (idempotent).""" + if not products: + return 0 + self.client.delete_collection(COLLECTION) + self._ensure_collection() + docs = [ + Document( + page_content=( + f"{p['name']} ({p['category']} card). Annual fee: " + f"{p['annual_fee']}. APR: {p['apr']}. Credit limit: " + f"{p['credit_limit']}. Rewards: {p['rewards']}. " + f"{p['description']}" + ), + metadata={"product_id": int(p["product_id"]), "name": p["name"]}, + ) + for p in products + ] + vectors = self.embeddings.embed_documents([d.page_content for d in docs]) + self.client.upsert( + collection_name=COLLECTION, + points=[ + { + "id": d.metadata["product_id"], + "vector": vector, + "payload": {**d.metadata, "text": d.page_content}, + } + for d, vector in zip(docs, vectors) + ], + ) + return len(docs) + + def search(self, query: str, k: int = 2) -> list[dict[str, Any]]: + if self.client.count(collection_name=COLLECTION, exact=True).count == 0: + return [] + vector = self.embeddings.embed_query(query) + hits = self.client.query_points(collection_name=COLLECTION, query=vector, limit=k).points + return [ + {"text": h.payload.get("text", ""), "name": h.payload.get("name", ""), "score": h.score} + for h in hits + ] diff --git a/samples/llm-chatbot/docker-compose.yml b/samples/llm-chatbot/docker-compose.yml new file mode 100644 index 0000000000..bcf0342b67 --- /dev/null +++ b/samples/llm-chatbot/docker-compose.yml @@ -0,0 +1,102 @@ +version: '3.8' + +services: + # ------------------------------------------------------------------ + # Postgres: raw operational datasets (customer profiles, transactions + # and the credit-card product catalog) that Dozer ingests. + # ------------------------------------------------------------------ + postgres: + image: postgres:16 + container_name: bank-postgres + command: + - postgres + - -c + - wal_level=logical + - -c + - max_replication_slots=10 + - -c + - max_wal_senders=10 + - -c + - hba_file=/etc/dozer/pg_hba.conf + environment: + POSTGRES_DB: bank + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + POSTGRES_HOST_AUTH_METHOD: trust + volumes: + - ./seed/init.sql:/docker-entrypoint-initdb.d/init.sql:ro + - ./pg_hba.conf:/etc/dozer/pg_hba.conf:ro + ports: + - '5436:5432' + healthcheck: + test: ['CMD-SHELL', 'pg_isready -U postgres -d bank'] + interval: 5s + timeout: 5s + retries: 10 + + # ------------------------------------------------------------------ + # Qdrant: vector store for the credit-card product catalog. + # The chatbot embeds product descriptions into it and retrieves the + # most relevant products as context for the LLM (RAG). + # ------------------------------------------------------------------ + qdrant: + image: qdrant/qdrant:latest + container_name: bank-qdrant + ports: + - '6333:6333' + - '6334:6334' + volumes: + - qdrant_storage:/qdrant/storage + healthcheck: + test: ['CMD-SHELL', 'bash -c ":> /dev/tcp/127.0.0.1/6333" 2>/dev/null || exit 1'] + interval: 5s + timeout: 5s + retries: 10 + + # ------------------------------------------------------------------ + # Dozer: ingests customers + transactions + products from Postgres, + # joins them in real time (see dozer-config.yaml) and serves the + # unified customer profile over REST (:8080) and gRPC (:50051). + # ------------------------------------------------------------------ + dozer: + image: getdozer/dozer:latest + platform: linux/amd64 + container_name: bank-dozer + depends_on: + postgres: + condition: service_healthy + working_dir: /usr/dozer + volumes: + - .:/usr/dozer + ports: + - '8080:8080' + - '50051:50051' + command: ['dozer', 'run'] + + # ------------------------------------------------------------------ + # Chatbot: Langchain app that + # 1. indexes the product catalog (from Dozer REST) into Qdrant, + # 2. fetches the unified customer profile per request, + # 3. retrieves matching credit-card products from Qdrant, + # 4. calls the LLM (OpenAI) with all of it as context. + # Runs fully offline (mock LLM + fake embeddings) when OPENAI_API_KEY + # is not set, so it works in CI without secrets. + # ------------------------------------------------------------------ + chatbot: + build: ./chatbot + container_name: bank-chatbot + depends_on: + - dozer + - qdrant + environment: + DOZER_REST_URL: http://dozer:8080 + QDRANT_URL: http://qdrant:6333 + OPENAI_API_KEY: ${OPENAI_API_KEY:-} + OPENAI_BASE_URL: ${OPENAI_BASE_URL:-} + OPENAI_MODEL: ${OPENAI_MODEL:-gpt-4o-mini} + ports: + - '8000:8000' + command: ['uvicorn', 'main:app', '--host', '0.0.0.0', '--port', '8000'] + +volumes: + qdrant_storage: diff --git a/samples/llm-chatbot/dozer-config.yaml b/samples/llm-chatbot/dozer-config.yaml new file mode 100644 index 0000000000..9625ea4f86 --- /dev/null +++ b/samples/llm-chatbot/dozer-config.yaml @@ -0,0 +1,102 @@ +# Dozer configuration for the LLM chatbot sample. +# +# Dozer sources three Postgres tables (customers, transactions, products), +# joins them in real time into a unified customer profile and exposes +# everything as low-latency REST/gRPC endpoints on :8080 / :50051. +app_name: bank-chatbot +version: 1 + +connections: + - name: bank_conn + config: !Postgres + user: postgres + password: postgres + host: postgres + port: 5432 + database: bank + +sources: + - name: customers + table_name: customers + connection: bank_conn + + - name: transactions + table_name: transactions + connection: bank_conn + + - name: products + table_name: products + connection: bank_conn + +sql: | + -- Unified customer profile: demographics + credit standing joined with + -- aggregated spending behaviour. Dozer keeps this continuously up to + -- date as new transactions land in Postgres. + select + c.customer_id, + c.first_name, + c.last_name, + c.email, + c.city, + c.segment, + c.credit_score, + count(t.txn_id) as transaction_count, + sum(t.amount) as total_spend, + max(t.amount) as largest_transaction, + 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, + c.first_name, + c.last_name, + c.email, + c.city, + c.segment, + c.credit_score; + +endpoints: + - name: customers + path: /customers + table_name: customers + index: + primary_key: + - customer_id + + - name: transactions + path: /transactions + table_name: transactions + index: + primary_key: + - txn_id + secondary: + create: + - !SortedInverted + fields: + - customer_id + - amount + - !SortedInverted + fields: + - customer_id + - txn_date + + - name: products + path: /products + table_name: products + index: + primary_key: + - product_id + + - name: customer_profile + path: /profile + table_name: customer_profile + index: + primary_key: + - customer_id + - first_name + - last_name + - email + - city + - segment + - credit_score diff --git a/samples/llm-chatbot/pg_hba.conf b/samples/llm-chatbot/pg_hba.conf new file mode 100644 index 0000000000..e39366042f --- /dev/null +++ b/samples/llm-chatbot/pg_hba.conf @@ -0,0 +1,8 @@ +# Allow all connections (including logical replication) from any host. +# Used together with `postgres -c hba_file=/etc/dozer/pg_hba.conf` so the +# Dozer container can open a replication slot against Postgres. +local all all trust +host all all 0.0.0.0/0 trust +host all all ::0/0 trust +host replication all 0.0.0.0/0 trust +host replication all ::0/0 trust diff --git a/samples/llm-chatbot/seed/init.sql b/samples/llm-chatbot/seed/init.sql new file mode 100644 index 0000000000..c66a9719fc --- /dev/null +++ b/samples/llm-chatbot/seed/init.sql @@ -0,0 +1,77 @@ +-- Seed data for the Dozer + LLM chatbot sample. +-- Three operational datasets: customer profiles, transactions and a +-- credit-card product catalog. + +CREATE TABLE customers ( + customer_id BIGINT PRIMARY KEY, + first_name TEXT NOT NULL, + last_name TEXT NOT NULL, + email TEXT NOT NULL, + city TEXT NOT NULL, + segment TEXT NOT NULL, + credit_score INT NOT NULL +); + +CREATE TABLE transactions ( + txn_id BIGINT PRIMARY KEY, + customer_id BIGINT NOT NULL, + merchant TEXT NOT NULL, + category TEXT NOT NULL, + amount NUMERIC(10, 2) NOT NULL, + txn_date DATE NOT NULL +); + +CREATE TABLE products ( + product_id BIGINT PRIMARY KEY, + name TEXT NOT NULL, + category TEXT NOT NULL, + annual_fee NUMERIC(8, 2) NOT NULL, + apr NUMERIC(5, 2) NOT NULL, + credit_limit NUMERIC(8, 2) NOT NULL, + rewards TEXT NOT NULL, + description TEXT NOT NULL +); + +INSERT INTO customers (customer_id, first_name, last_name, email, city, segment, credit_score) VALUES + (1001, 'Alice', 'Nguyen', 'alice.nguyen@example.com', 'Singapore', 'premium', 782), + (1002, 'Bruno', 'Costa', 'bruno.costa@example.com', 'Sao Paulo', 'mass', 655), + (1003, 'Chloe', 'Dubois', 'chloe.dubois@example.com', 'Paris', 'premium', 811), + (1004, 'Daniel', 'Kim', 'daniel.kim@example.com', 'Seoul', 'mass', 690), + (1005, 'Elena', 'Rossi', 'elena.rossi@example.com', 'Milan', 'affluent', 745); + +INSERT INTO transactions (txn_id, customer_id, merchant, category, amount, txn_date) VALUES + (1, 1001, 'Wanderlust Airways', 'travel', 1250.00, '2026-09-28'), + (2, 1001, 'Grand Plaza Hotel', 'travel', 640.50, '2026-09-29'), + (3, 1001, 'Bistro Moderne', 'dining', 89.90, '2026-09-30'), + (4, 1002, 'Supermercado Bom', 'groceries', 152.30, '2026-09-27'), + (5, 1002, 'Cafe Central', 'dining', 18.75, '2026-09-28'), + (6, 1002, 'Metro Transit', 'transport', 42.00, '2026-09-30'), + (7, 1003, 'Mode Ensemble', 'shopping', 890.00, '2026-09-26'), + (8, 1003, 'Artisan Gallery', 'shopping', 1450.00, '2026-09-29'), + (9, 1003, 'Le Petit Bistrot', 'dining', 120.40, '2026-09-30'), + (10, 1004, 'TechNova Store', 'electronics', 799.00, '2026-09-25'), + (11, 1004, 'StreamFlix', 'entertainment', 15.99, '2026-09-28'), + (12, 1004, 'Hankuk BBQ', 'dining', 67.20, '2026-09-29'), + (13, 1005, 'Sempione Market', 'groceries', 210.15, '2026-09-26'), + (14, 1005, 'Milano Centrale Taxi','transport', 35.50, '2026-09-27'), + (15, 1005, 'Boutique Italia', 'shopping', 480.00, '2026-09-30'), + (16, 1005, 'Trattoria da Vito', 'dining', 95.80, '2026-10-01'), + (17, 1001, 'SkyLounge Bar', 'dining', 210.00, '2026-10-02'), + (18, 1003, 'Boutique Hotel Lune', 'travel', 980.00, '2026-10-02'); + +INSERT INTO products (product_id, name, category, annual_fee, apr, credit_limit, rewards, description) VALUES + (2001, 'Velocity Travel Rewards', 'travel', 195.00, 17.90, 15000.00, + '3x points on flights and hotels', + 'Premium travel card with 3x points on flights and hotels, airport lounge access and no foreign transaction fees. Ideal for frequent flyers who book international travel.'), + (2002, 'Everyday Cashback', 'cashback', 0.00, 15.90, 8000.00, + '5% cashback on groceries and fuel', + 'No annual fee cashback card earning 5% back on groceries and fuel and 1% on everything else. A straightforward everyday card for households and commuters.'), + (2003, 'Luxe Concierge', 'premium', 549.00, 16.90, 30000.00, + '2x points everywhere plus concierge', + 'Invitation-style premium card with a personal concierge, fine-dining and hotel elite status, 2x points on all spending and complimentary premium subscriptions for high spenders.'), + (2004, 'Student Starter', 'starter', 0.00, 19.90, 2000.00, + '1% cashback, credit builder', + 'Entry-level card for students and first cardholders with 1% cashback, no annual fee and free credit-score monitoring to help build a credit history responsibly.'), + (2005, 'Dining & Entertainment', 'dining', 95.00, 16.40, 10000.00, + '4x points on restaurants and shows', + 'Rewards card focused on lifestyle spending with 4x points at restaurants, bars, theatres and streaming services, plus periodic dining-exclusive offers and tastings.'); diff --git a/samples/llm-chatbot/tests/test_sample.py b/samples/llm-chatbot/tests/test_sample.py new file mode 100644 index 0000000000..5d1ffdaec5 --- /dev/null +++ b/samples/llm-chatbot/tests/test_sample.py @@ -0,0 +1,80 @@ +"""End-to-end smoke tests for the Dozer + LLM chatbot sample. + +Prerequisites (see README): + docker compose up -d postgres qdrant + dozer run # or the dozer container from the compose file + +Run: + pytest tests/test_sample.py -v + +The tests run keyless (mock LLM + hashing embeddings) and exercise the full +chain: Dozer REST -> unified profile -> Qdrant retrieval -> LLM prompt. +""" + +from __future__ import annotations + +import os + +from dozer_client import DozerClient + +DOZER_URL = os.getenv("DOZER_REST_URL", "http://localhost:8080") +QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333") + + +def make_client() -> DozerClient: + return DozerClient(DOZER_URL) + + +def test_dozer_health() -> None: + assert make_client().health() + + +def test_endpoints_served() -> None: + endpoints = make_client().list_endpoints() + for expected in ("/customers", "/transactions", "/products", "/profile"): + assert expected in endpoints, f"missing endpoint {expected} in {endpoints}" + + +def test_unified_profile_is_joined_and_aggregated() -> None: + """The profile must combine customer columns with transaction aggregates.""" + profile = make_client().get_profile(1001) + assert profile is not None, "customer 1001 missing from /profile" + assert profile["first_name"] == "Alice" + assert profile["segment"] == "premium" + # Aggregates over the seeded transactions for Alice. + assert profile["transaction_count"] == 4 + assert profile["total_spend"] == "2190.40" + assert profile["last_transaction_date"] == "2026-10-02" + + +def test_transactions_query_dsl() -> None: + txns = make_client().get_transactions(1003, limit=3) + assert len(txns) == 3 + dates = [t["txn_date"] for t in txns] + assert dates == sorted(dates, reverse=True), "expected $order_by txn_date desc" + assert all(t["customer_id"] == 1003 for t in txns) + + +def test_chatbot_end_to_end_keyless() -> None: + from chatbot import answer + from vector_store import ProductVectorStore + + dozer = make_client() + store = ProductVectorStore() + indexed = store.index_products(dozer.get_products()) + assert indexed == 5, "expected the 5 seeded products to be indexed" + + reply = answer(dozer, store, 1001, "I travel often - which card fits me best?") + assert "Alice" in reply, "reply must be personalized with the profile name" + assert "premium" in reply, "reply must reference the profile segment" + assert "Mock LLM" in reply, "keyless run must use the mock LLM" + + +def test_unknown_customer_is_handled() -> None: + from chatbot import answer + from vector_store import ProductVectorStore + + dozer = make_client() + store = ProductVectorStore() + reply = answer(dozer, store, 9999, "hello") + assert "not found" in reply