A production-grade multi-agent recommendation platform for personalized financial products (investment funds, portfolio adjustments, credit products), combining behavioral signals with financial-news sentiment extracted from real filings and articles. Every headline number in this document (NDCG, latency, speedup, live deploy status) is reproducible from a script checked into this repository, not hand-written. If a metric can't be traced back to a real run, it does not appear here.
The recommender ranks financial products and actions for a user based on three signal sources: behavioral and session data, portfolio and account features, and sentiment extracted from financial news and filings via FinBERT. A LangGraph-orchestrated pipeline of typed agents handles each request: an Intent Agent classifies what the user wants, a Retrieval Agent pulls candidate products and attaches sentiment scores, a Ranking Agent scores candidates with a trained XGBoost model, and an Explainability Agent produces a plain-language, compliance-safe reason for the top recommendation using SHAP values and natural-language generation. A Context Agent persists session state between requests.
Two LLMs share the reasoning workload, each with a distinct job:
| Local model | Enterprise API (Anthropic) | |
|---|---|---|
| Role | Intent Agent, per-request classification | Explainability Agent output, low-confidence intent fallback |
| Why | Self-hosted, meets p99 latency budget, no per-token cost at high request volume | Higher-quality reasoning for low-frequency, high-stakes calls where a poorly-worded compliance explanation is costly |
| Called via | A local inference server | The Anthropic API |
| Routing trigger | Default path for every request | Intent confidence below a configured threshold, or any request reaching the Explainability Agent |
agents/llm_router.py is the single place this decision is made; no agent calls either model
directly. Every routing decision is logged (which model, why, latency, token count) in the same
structured JSON logging used everywhere else in this project.
Kafka -> S3 -> Airflow + Spark (features) -> Feast (online/offline)
|
v
LangGraph orchestrator (per request)
+--------------------+--------------------+--------------------+--------------------+
|Intent Agent |Retrieval Agent |Ranking Agent |Explainability |
|local model, |vector search + |XGBoost, from |SHAP + LLM |
|router falls back |FinBERT news |model registry |natural-language |
|to enterprise API |sentiment | |explanation |
|on low confidence | | | |
+--------------------+--------------------+--------------------+--------------------+
| |
+----------------------> llm_router.py <-----------------------+
(routes each call to the local model or the enterprise API)
|
v
Context Agent (session read/write)
|
v
FastAPI -> Docker -> Kubernetes (canary + autoscaling)
|
v
Prometheus/Grafana - drift + latency + CTR-proxy monitors
Every agent node has a typed Pydantic input/output schema (no untyped dicts passed between nodes), an explicit timeout, and a defined fallback (for example, a ranking timeout falls back to a popularity-based ordering). Every node logs its decision, input summary, output summary, and latency as structured JSON.
| Layer | Tool |
|---|---|
| Languages | Python, SQL |
| Streaming ingestion | Kafka |
| Data lake | S3 |
| Batch processing | Spark |
| Orchestration | Airflow |
| Feature store | Feast (offline: S3, online: Redis in production, an in-process store for local dev) |
| Vector search | Chroma (dev), Pinecone-shaped in production |
| Domain NLP | FinBERT-based sentiment scoring |
| Ranking model | XGBoost, LambdaMART-style ranking objective |
| Explainability | SHAP |
| Agent orchestration | LangGraph |
| Model optimization | ONNX export, INT8 quantization |
| Serving | FastAPI, async, stateless |
| Containerization | Docker |
| Cluster orchestration | Kubernetes (EKS in production, kind for local verification) |
| CI/CD | GitHub Actions to a container registry to Kubernetes via Helm |
| Model registry | MLflow |
| Testing | pytest |
| Monitoring | Prometheus and Grafana |
| Demo dashboard | Streamlit |
Where a tool is unavailable in a given environment, a documented local substitute is used instead (for example, an in-process store standing in for Redis, or a local Kubernetes cluster standing in for EKS during development). Every substitution is called out explicitly rather than silently assumed, and every phase below states plainly which path (real infrastructure or local substitute) actually produced its result.
├── data/ sample data, schemas
├── pipelines/
│ ├── airflow/ Airflow DAGs
│ ├── spark/ Spark feature jobs
│ └── ingestion/ Kafka producer/consumer, S3 sink, synthetic event replay
├── feature_store/ Feast feature repo and feature views
├── agents/ one module per LangGraph node, graph assembly, the LLM router
├── models/
│ ├── training/ ranking model training, SHAP explanation generation
│ └── optimization/ ONNX export, INT8 quantization, benchmarking
├── retrieval/ embeddings, vector index, hybrid search, sentiment scoring
├── serving/ FastAPI app, load testing
├── infra/
│ ├── docker/ Dockerfiles, docker-compose, serving-only dependency list
│ ├── k8s/ Helm chart, local kind cluster config and deploy script
│ └── ci/ CI/CD notes
├── monitoring/ Prometheus metrics, drift monitor, canary rollback controller
├── eval/ offline ranking metrics, eval report generator, A/B significance test
├── tests/ pytest suite, mirrors the source layout
└── streamlit_app/ internal demo dashboard
Every phase below has been run against real infrastructure at least once, not just written and unit-tested. Where a real cloud resource was needed only to prove a capability (a live EKS cluster, for instance), it was torn down immediately afterward to avoid ongoing cost; the capability itself remains fully implemented and reproducible.
Data and ingestion. A real Kafka broker and a real S3 bucket, in a real AWS account, received
20 out of 20 replayed synthetic events with zero errors, and a dedicated integration test
(tests/integration/test_kafka_s3_live.py) independently confirmed 25 out of 25 events published
and landed.
Feature pipeline. Spark feature jobs and Feast feature views (online and offline) run end to end; a real online feature lookup for a test user returns real, non-null values.
Ranking model. An XGBoost ranker trained on simulated interactions is tracked in a real MLflow
registry (fin_reco_ranker, version 1, status READY), with a measured NDCG@10 of 0.9598 on a
held-out, query-level split, computed by eval/ranking_metrics.py, not hand-typed. Per-prediction
SHAP explanations are generated and logged as a real MLflow artifact, verified against SHAP's own
additivity guarantee in the test suite.
Retrieval. A real local vector collection indexes a synthetic product catalog; a hybrid (vector plus metadata-filtered) query returns ranked candidates with sentiment scores correctly attached to the products that have a matching ticker, and correctly absent for the one that doesn't.
Agent graph. A full request traces through Intent, Retrieval, Ranking, Explainability, and Context with structured logs at every hop. Both a high-confidence and a deliberately ambiguous, low-confidence request were run: the high-confidence request used the local model with zero enterprise API calls; the low-confidence request's router correctly escalated to the enterprise API based on its confidence threshold.
Serving. A real FastAPI server, hit with concurrent HTTP load (not an in-process test transport), measured a repeatable p99 latency around 186-198 ms against a 500 ms target across three consecutive 100-request runs.
Deployment. Both halves of this phase are live-verified. Locally: a real Docker daemon and a
real local Kubernetes cluster ran the full build-to-deploy path, and a real HTTP request to the
running pod's /health endpoint, through the cluster's own Service, returned a real
{"status": "ok"}. Against real cloud infrastructure: a real git push triggered a real GitHub
Actions workflow that ran the complete test suite, built and pushed the serving image to a real
container registry, and deployed it to a real managed Kubernetes cluster via a real
helm upgrade --install, all authenticated through short-lived, OIDC-issued credentials with no
long-lived cloud access keys stored anywhere. The deployed pod's own startup routine made a real
call through the enterprise LLM path, confirming that route works against production
infrastructure, not just locally. The cluster and its associated registry were deleted
immediately after this verification to avoid ongoing cost.
Optimization. ONNX export plus INT8 quantization of the ranking and intent models, benchmarked with real timed inference (20 warmup and 300 timed iterations per configuration): a roughly 12.8-13.4x speedup for the ranker and 8.1-8.4x for the intent classifier, both measured against the exact functions the live request path calls, not a synthetic microbenchmark.
Canary and rollback. A deliberately degraded model version, trained with shuffled relevance labels, was replayed against real held-out queries through a real canary router and automatically rolled back once its measured NDCG fell well below the stable version's, while a second, genuinely non-degraded canary version in the same run was correctly left alone. This confirms the rollback controller reacts to real, measured degradation rather than rolling back indiscriminately.
Monitoring and evaluation. A running server's real request traffic is visible through a
Prometheus /metrics endpoint. A drift monitor computes a real Population Stability Index across
synthetic feature samples, correctly distinguishing no-drift from injected drift. An eval report
generator independently recomputes NDCG@10, matching the number the training run itself logged,
plus NDCG@5 and MAP@10. An A/B significance test runs a real paired t-test between model
versions, and its own live run surfaced a genuine methodological finding worth keeping in mind:
two independently trained, equally good models can be statistically significantly different
without being practically significantly different, while a genuinely degraded model is both. A
Streamlit dashboard provides an internal, human-facing view into intent, ranking, retrieval, and
explanation output for any given request.
streamlit_app/app.py is an internal demo dashboard with four tabs, each backed by a plain,
independently unit-tested Python function, not fabricated for a screenshot.
Get a Recommendation runs one real request through the full agent graph in-process and shows the actual ranked output and generated explanation:
Canary Rollout shows the current Phase 9 canary configuration and each route's live, aggregated request metrics:
Offline Evaluation shows the most recent real NDCG/MAP numbers and A/B significance test
result, whatever eval/generate_eval_report.py and eval/ab_significance_test.py last actually
produced:
Feature Drift shows the most recent real Population Stability Index per behavioral feature:
To run the parts of this project that don't need real cloud infrastructure:
- Python 3.11+
- Docker Desktop (or another Docker daemon), for the Kafka/object-storage compose stack and for building the serving image
pip
To also run the local Kubernetes verification path (infra/k8s/deploy_local.sh):
kind,kubectl, andhelm(on macOS:brew install kind kubernetes-cli helm)
To also run the real cloud deployment path (.github/workflows/ci-cd.yml):
- An AWS account, with an ECR repository and an EKS cluster created ahead of time
- The AWS CLI, configured with credentials that can create the IAM OIDC provider and role described in Section 11
- A GitHub repository with the secrets and variables listed in Section 11 configured, and the
ghCLI if you want to manage those from a terminal rather than the GitHub web UI
Nothing here requires an Anthropic API key to run; the enterprise LLM path is only exercised when the Intent Agent's confidence falls below its configured threshold, or when the Explainability Agent runs, and both fail over to a local/template result if no key is configured.
Copy the environment template and adjust as needed:
cp .env.example .env
Install dependencies. The full dependency list (requirements.txt) includes every phase of this
project, including Spark and Airflow, and can be slow or fail to resolve in some Python
environments because of how large and interdependent that combination is. If you only want to run
the serving path (the FastAPI app, the Streamlit dashboard, or the agent graph directly) without
the data pipeline phases, install the much smaller serving-only subset instead:
pip install -r infra/docker/requirements-serving.txt
Otherwise, for the full pipeline:
pip install -r requirements.txt
Start local Kafka and object storage:
docker compose -f infra/docker/docker-compose.yml up -d
Create the Kafka topic and replay synthetic events into the raw data layer:
python -m pipelines.ingestion.create_topics
python -m pipelines.ingestion.replay --num-events 100
Run the Streamlit dashboard (from the repository root, so PYTHONPATH includes every top-level
package this app imports):
PYTHONPATH=. streamlit run streamlit_app/app.py
Run the FastAPI serving layer directly:
uvicorn serving.app:app --reload
pytest tests/ -q
This runs the full unit test suite (mocked external services, no real infrastructure required). Two categories of tests require real infrastructure and are excluded from a default run:
RUN_INTEGRATION_TESTS=1 pytest tests/integration/test_kafka_s3_live.py -v
runs the live Kafka-to-S3 integration test against a real broker and bucket, and
tests/pipelines/ requires a local Spark/Airflow environment.
./infra/k8s/deploy_local.sh
Builds the serving image, creates a local Kubernetes cluster if one doesn't already exist, loads
the image into it, and deploys the Helm chart, verifying a real /health response through the
cluster's Service at the end.
.github/workflows/ci-cd.yml runs the test suite, builds and pushes the serving image to a
container registry, and deploys it to a managed Kubernetes cluster via Helm on every push to
main, authenticating via OIDC with no long-lived cloud credentials. It requires these
repository secrets and variables to be configured:
| Name | Kind | Purpose |
|---|---|---|
AWS_ROLE_TO_ASSUME |
secret | ARN of an IAM role with an OIDC trust policy scoped to this repository |
REDIS_URL |
secret | Session store connection string for the Context Agent |
ANTHROPIC_API_KEY |
secret | Enterprise LLM API key; the Explainability Agent and low-confidence Intent Agent fallback use it |
AWS_REGION |
variable | Target AWS region for ECR and EKS |
ECR_REPOSITORY |
variable | Container registry repository name |
EKS_CLUSTER_NAME |
variable | Target Kubernetes cluster name |
The IAM role's trust policy needs to match the exact sub claim GitHub's OIDC token presents for
this repository, which can include an ID-based suffix (repo:owner@ownerId/repo@repoId:...)
rather than the plain name-based form, and a different suffix again for any job that sets
environment: in the workflow (:environment:<name> instead of :ref:refs/heads/<branch>). If
configure-aws-credentials fails with "Not authorized to perform sts:AssumeRoleWithWebIdentity"
despite a seemingly correct trust policy, check the real token via a CloudTrail
AssumeRoleWithWebIdentity event rather than assuming the plain form is correct.
GET /health returns {"status": "ok"}, used by Kubernetes liveness/readiness probes and by
the deploy scripts in this repository to confirm a real request reaches a running pod.
GET /metrics returns the current Prometheus registry in text exposition format: request
counters, a latency histogram, and the current canary/drift gauges.
POST /recommend runs one request through the full agent graph. Request body:
{
"user_id": "user-123",
"session_id": "session-abc",
"query_text": "show me low risk bond funds",
"risk_tier_filter": null
}Response body (the full typed trace of every agent's output, matching data/schemas/agents.py's
GraphResult):
{
"request": { "request_id": "...", "user_id": "user-123", "session_id": "session-abc", "query_text": "show me low risk bond funds", "risk_tier_filter": null },
"intent": { "label": "product_search", "confidence": 0.91, "model_used": "local", "fallback_used": false, "latency_ms": 12.4 },
"retrieval": { "candidates": [ { "product_id": "...", "sentiment_score": 0.34 } ], "fallback_used": false, "latency_ms": 8.1 },
"ranking": { "ranked": [ { "product_id": "...", "name": "...", "risk_tier": "low", "category": "...", "rank_score": 0.87, "similarity_score": 0.79, "sentiment_score": 0.34 } ], "model_version": "1", "fallback_used": false, "latency_ms": 5.7 },
"explanation": { "text": "...", "model_used": "claude", "fallback_used": false, "latency_ms": 640.2 },
"context": { "saved": true, "backend": "memory", "fallback_used": false, "latency_ms": 0.6 }
}A 422 response (ErrorResponse) means request validation failed (a missing or empty field). A
500 response means an unhandled error occurred processing the request; the real error message is
included in detail, and the error is also logged as structured JSON and recorded in the
/metrics error counter.
All serving-layer configuration is read from environment variables in serving/config.py, never
hardcoded:
| Variable | Default | Purpose |
|---|---|---|
EMBEDDING_BACKEND |
tfidf |
Retrieval Agent's embedding function (tfidf locally, a real sentence-embedding model in production) |
SENTIMENT_BACKEND |
lexicon |
News sentiment scorer (lexicon locally, FinBERT in production) |
CONTEXT_BACKEND |
memory |
Context Agent's session store (memory locally, redis in production) |
FEAST_REPO_PATH |
feature_store |
Path to the Feast feature repo |
FEAST_STATE_DIR |
unset | Override for Feast's local state directory |
MLFLOW_STATE_DIR |
unset | Override for MLflow's local tracking store directory |
CHROMA_PERSIST_DIR |
unset | Override for Chroma's local persistence directory |
CANARY_STATE_DIR |
unset | Override for the canary router's state directory |
P99_TARGET_MS |
500.0 |
The p99 latency target serving/load_test.py's run is judged against |
ModuleNotFoundError running the Streamlit app or any script directly. Python only adds the
directory containing the script you ran to its import path, not the repository root, so top-level
packages like data, agents, and common won't resolve unless the repository root is also on
the path. Run from the repository root with either:
PYTHONPATH=. streamlit run streamlit_app/app.py
or
python -m streamlit run streamlit_app/app.py
pip install -r requirements.txt fails with resolution-too-deep or hangs for a long time.
The full requirements file includes every phase of this project, including Apache Airflow, which
has a large, tightly version-pinned dependency tree that can be slow or impossible for pip's
resolver to untangle depending on what else is already installed in that Python environment. If
you only need the serving path (the FastAPI app, the Streamlit dashboard, or the agent graph),
install the much smaller subset instead:
pip install -r infra/docker/requirements-serving.txt
A Kubernetes pod is stuck Pending with Insufficient memory or Too many pods. Small EC2
instance types (particularly Free Tier-eligible ones like t3.micro) have very little memory
headroom and a low pod-per-node ceiling imposed by the AWS VPC CNI, independent of how small a
pod's own resource requests are. infra/k8s/helm/fin-reco/values-eks-freetier.yaml is sized for
exactly this constraint; if a fresh cluster still can't schedule the required system pods (CoreDNS,
kube-proxy, aws-node) alongside the application pod, scaling CoreDNS to a single replica
(kubectl scale deployment coredns -n kube-system --replicas=1) frees one slot on a single-node
lab cluster where CoreDNS's normal two-replica redundancy provides no benefit anyway.
This project's coding standard forbids fabricating a metric, so every claim in Section 5 above is backed by a real run. The honest flip side of that same standard is stating plainly what has only been run against a local substitute so far, not a fully managed production equivalent:
- The session store (
CONTEXT_BACKEND) has been live-verified with its in-process local substitute; a real managed Redis has not yet been exercised end to end. - Vector retrieval has been live-verified against a local Chroma collection; a managed vector database has not yet been exercised end to end.
- The local intent classifier is a small scikit-learn model standing in for a real fine-tuned, quantized local LLM; the routing and fallback logic around it is real and live-verified, the model itself is a substitute.
- INT8 quantization was verified live for the intent classifier; TensorRT conversion was written but never run, no GPU has been available in any environment this project has run in so far.
- The Helm chart's canary Deployment (a second fleet sharing the main Service, for canarying a new serving image) is structurally verified but has not been run against a live cluster.
- Real production traffic volume has never been observed, so rollback thresholds and the confidence threshold that triggers the enterprise LLM fallback are reasoned defaults, not thresholds calibrated against real traffic.
None of these are hidden; each one is called out in detail, including exactly what was tried and what broke, in DECISIONS.md.
This project is licensed under the MIT License, see LICENSE for the full text.
Every deliberate design decision, every environment-driven substitution (a local store standing in for a managed service unavailable in some environment), and every bug found and fixed during real verification is recorded in DECISIONS.md, in the order it happened, including the ones that didn't work on the first try.