Module 5: Vector Database Landscape for AI Engineers

Capsule 08: Project - Decision Tree for Choosing a Vector DB

Next →

Capsule description

You'll close the module by designing and implementing a programmatic decision tree to select a vector database for RAG projects. It's not just a static diagram: you'll build an engine in Python that receives concrete requirements and generates a recommendation with justification, alternatives, and explicit trade-offs.

The output feeds directly into Module 6 (scoring matrix).

Estimated time: 35-45 minutes

🎯 Project objective

  • Implement a decision tree as Python code with 6 decision dimensions.
  • Produce a primary recommendation + an alternative with justification.
  • Validate against 5 real scenarios and generate a Mermaid visualization.

📋 Project specifications

Functional requirements

  1. Receive the project's requirements and evaluate 6 dimensions: volume, operations, budget, latency, compliance, and team.
  2. Produce a primary recommendation with a score + an alternative with conditions for when to prefer it.
  3. Justify why the other options were discarded.

Success Criteria

  • ✅ Covers 5 providers: ChromaDB, Pinecone, Weaviate, Qdrant, Milvus
  • ✅ 5 validated scenarios with a coherent result
  • ✅ Justification with technical and financial trade-offs
  • ✅ Mermaid visualization generated

🧠 Context before starting

This project synthesizes capsules 02-07: the provider landscape (profiles), managed vs self-hosted (key node), features for RAG (criteria), costs/trade-offs (budget), when to choose each option (logic), and anti-patterns (validation). If any concept isn't clear, review the corresponding capsule before continuing.

💻 Step-by-step implementation

Step 1: Define the requirements model

The first step is to structure the inputs the tree needs to make a decision. Each field represents a dimension that impacts the choice.

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class OperationalModel(Enum):
    MANAGED = "managed"
    SELF_HOSTED = "self_hosted"
    FLEXIBLE = "flexible"  # no strong preference


class ComplianceLevel(Enum):
    NONE = "none"
    BASIC = "basic"        # generic GDPR, non-sensitive data
    STRICT = "strict"      # data residency, audit, financial/health regulation


class TeamCapability(Enum):
    MINIMAL = "minimal"    # 1-2 devs, no DevOps
    MODERATE = "moderate"  # small team with some infra
    STRONG = "strong"      # team with DevOps or a dedicated platform


class ProjectStage(Enum):
    POC = "poc"
    VALIDATION = "validation"
    PRODUCTION = "production"
    SCALE = "scale"


@dataclass
class ProjectRequirements:
    """Project requirements for vector DB selection."""

    vector_count: int                             # current or projected vectors at 6 months
    operational_preference: OperationalModel      # managed vs self-hosted
    monthly_budget_usd: float                     # monthly budget for the vector DB
    target_latency_p95_ms: float                  # target p95 latency in ms
    compliance: ComplianceLevel                   # required compliance level
    team_capability: TeamCapability               # the team's operational capacity
    project_stage: ProjectStage                   # the product's current stage
    needs_hybrid_search: bool = False             # does it require hybrid search?
    needs_multi_tenancy: bool = False             # does it require multi-tenant isolation?
    project_name: str = "unnamed"                 # project name (for reports)

    def validate(self) -> list[str]:
        """Validate the consistency of the requirements."""
        warnings = []
        if self.vector_count < 0:
            warnings.append("vector_count cannot be negative")
        if self.monthly_budget_usd < 0:
            warnings.append("monthly_budget_usd cannot be negative")
        if self.target_latency_p95_ms <= 0:
            warnings.append("target_latency_p95_ms must be positive")
        if (self.compliance == ComplianceLevel.STRICT
                and self.operational_preference == OperationalModel.MANAGED):
            warnings.append("ALERT: strict compliance + managed can be contradictory")
        if self.vector_count > 5_000_000 and self.team_capability == TeamCapability.MINIMAL:
            warnings.append("ALERT: scale >5M with a minimal team is an operational risk")
        if self.project_stage == ProjectStage.POC and self.monthly_budget_usd > 500:
            warnings.append("INFO: high budget for a PoC — consider a free option")
        return warnings

Why this design? The Enums force valid inputs, validate() detects contradictory combinations before running the tree, and each field maps to a dimension covered in capsules 02-06.

Step 2: Define each vector DB's profile

Before building the tree, you need a model of each provider. This centralizes the information and makes it easy to update the tree when the market changes.

@dataclass
class VectorDBProfile:
    """Profile of a vector database provider."""

    name: str
    max_scale_comfort: int         # vectors where it operates comfortably
    supports_managed: bool
    supports_self_hosted: bool
    supports_hybrid_search: bool
    supports_multi_tenancy: bool
    compliance_ready: bool         # does it support strict compliance natively?
    min_monthly_cost_usd: float    # estimated minimum operational cost
    operational_complexity: int    # 1 (simple) to 5 (requires a dedicated team)
    typical_latency_p95_ms: float  # typical p95 latency in production
    best_for: list[str] = field(default_factory=list)
    risks: list[str] = field(default_factory=list)


VECTOR_DB_PROFILES = {
    "ChromaDB": VectorDBProfile(
        name="ChromaDB",
        max_scale_comfort=500_000,
        supports_managed=False, supports_self_hosted=True,
        supports_hybrid_search=False, supports_multi_tenancy=False,
        compliance_ready=False,
        min_monthly_cost_usd=0, operational_complexity=1, typical_latency_p95_ms=5.0,
        best_for=["PoC and prototypes", "learning", "local development"],
        risks=["not meant for production scale", "no managed", "limited features"],
    ),
    "Pinecone": VectorDBProfile(
        name="Pinecone",
        max_scale_comfort=50_000_000,
        supports_managed=True, supports_self_hosted=False,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=70, operational_complexity=1, typical_latency_p95_ms=10.0,
        best_for=["managed production without DevOps", "fast SLA", "time-to-market"],
        risks=["strong vendor lock-in", "cost scales with volume", "no self-hosted"],
    ),
    "Weaviate": VectorDBProfile(
        name="Weaviate",
        max_scale_comfort=100_000_000,
        supports_managed=True, supports_self_hosted=True,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=25, operational_complexity=3, typical_latency_p95_ms=8.0,
        best_for=["advanced hybrid search", "managed+self-hosted flexibility", "broad ecosystem"],
        risks=["higher learning curve", "self-hosted operational complexity"],
    ),
    "Qdrant": VectorDBProfile(
        name="Qdrant",
        max_scale_comfort=100_000_000,
        supports_managed=True, supports_self_hosted=True,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=25, operational_complexity=2, typical_latency_p95_ms=5.0,
        best_for=["performance and efficiency", "self-hosted control with good UX"],
        risks=["younger ecosystem", "self-hosted requires operational discipline"],
    ),
    "Milvus": VectorDBProfile(
        name="Milvus",
        max_scale_comfort=1_000_000_000,
        supports_managed=True, supports_self_hosted=True,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=100, operational_complexity=5, typical_latency_p95_ms=12.0,
        best_for=["massive enterprise scale (>10M)", "dedicated platform team"],
        risks=["overkill for small projects", "high operational complexity"],
    ),
}

Note: these values are pedagogical estimates. In a real project, validate against your own benchmarks and up-to-date pricing.

Step 3: Build the decision engine

Each node evaluates a dimension and scores candidates. The structure is deliberately readable: another engineer should be able to follow the logic without additional documentation.

@dataclass
class Recommendation:
    """Result of the decision tree."""

    primary: str                           # name of the recommended DB
    primary_score: float                   # normalized score 0-100
    primary_justification: list[str]       # reasons for the recommendation
    alternative: str                       # second option
    alternative_score: float
    alternative_justification: list[str]
    eliminated: dict[str, list[str]]       # {db_name: [elimination reasons]}
    warnings: list[str]                    # process alerts
    confidence: str                        # "high", "medium", "low"


def evaluate_decision_tree(reqs: ProjectRequirements) -> Recommendation:
    """
    Evaluate requirements against vector DB profiles.
    Returns a structured recommendation with justification.
    """
    warnings = reqs.validate()
    scores: dict[str, float] = {}
    reasons: dict[str, list[str]] = {}
    eliminations: dict[str, list[str]] = {}

    for db_name, profile in VECTOR_DB_PROFILES.items():
        score = 0.0
        db_reasons = []
        db_eliminations = []

        # ── Node 1: Scale ──
        if reqs.vector_count <= profile.max_scale_comfort:
            scale_ratio = reqs.vector_count / profile.max_scale_comfort
            if scale_ratio < 0.3:
                score += 20
                db_reasons.append(f"comfortable scale ({reqs.vector_count:,} vectors)")
            elif scale_ratio < 0.7:
                score += 15
                db_reasons.append("scale within operational range")
            else:
                score += 8
                db_reasons.append("scale near the comfortable limit")
        else:
            score -= 30
            db_eliminations.append(
                f"scale ({reqs.vector_count:,}) exceeds comfortable zone "
                f"({profile.max_scale_comfort:,})"
            )

        # ── Node 2: Operational model ──
        if reqs.operational_preference == OperationalModel.MANAGED:
            if profile.supports_managed:
                score += 20
                db_reasons.append("supports the managed model")
            else:
                score -= 25
                db_eliminations.append("no managed option")

        elif reqs.operational_preference == OperationalModel.SELF_HOSTED:
            if profile.supports_self_hosted:
                score += 20
                db_reasons.append("supports self-hosted")
            else:
                score -= 25
                db_eliminations.append("no self-hosted option")

        else:  # FLEXIBLE
            if profile.supports_managed and profile.supports_self_hosted:
                score += 15
                db_reasons.append("flexible: supports both models")
            elif profile.supports_managed or profile.supports_self_hosted:
                score += 8
                db_reasons.append("supports at least one operational model")

        # ── Node 3: Budget ──
        if reqs.monthly_budget_usd >= profile.min_monthly_cost_usd:
            budget_headroom = reqs.monthly_budget_usd / max(profile.min_monthly_cost_usd, 1)
            if budget_headroom > 3:
                score += 15
                db_reasons.append("generous budget for this option")
            else:
                score += 10
                db_reasons.append("sufficient budget")
        else:
            score -= 20
            db_eliminations.append(
                f"budget (${reqs.monthly_budget_usd:.0f}) lower than "
                f"minimum cost (${profile.min_monthly_cost_usd:.0f})"
            )

        # ── Node 4: Latency ──
        if profile.typical_latency_p95_ms <= reqs.target_latency_p95_ms:
            latency_margin = reqs.target_latency_p95_ms - profile.typical_latency_p95_ms
            if latency_margin > 10:
                score += 15
                db_reasons.append("latency with ample margin")
            else:
                score += 10
                db_reasons.append("latency within target")
        else:
            score -= 15
            db_eliminations.append(
                f"typical latency ({profile.typical_latency_p95_ms}ms) "
                f"exceeds target ({reqs.target_latency_p95_ms}ms)"
            )

        # ── Node 5: Compliance ──
        if reqs.compliance == ComplianceLevel.STRICT:
            if profile.compliance_ready and profile.supports_self_hosted:
                score += 20
                db_reasons.append("supports strict compliance with self-hosted")
            elif profile.compliance_ready:
                score += 10
                db_reasons.append("compliance ready (verify data residency)")
            else:
                score -= 20
                db_eliminations.append("does not meet strict compliance requirements")
        elif reqs.compliance == ComplianceLevel.BASIC:
            if profile.compliance_ready:
                score += 10
                db_reasons.append("basic compliance covered")
            else:
                score += 3

        # ── Node 6: Team capability ──
        complexity_gap = profile.operational_complexity - _team_to_complexity(reqs.team_capability)
        if complexity_gap <= 0:
            score += 15
            db_reasons.append("operational complexity manageable by the team")
        elif complexity_gap == 1:
            score += 5
            db_reasons.append("operational complexity at the team's limit")
        else:
            score -= 15
            db_eliminations.append(
                f"operational complexity ({profile.operational_complexity}/5) "
                f"exceeds the team's capacity"
            )

        # ── Bonus: Special features ──
        if reqs.needs_hybrid_search:
            if profile.supports_hybrid_search:
                score += 10
                db_reasons.append("supports hybrid search")
            else:
                score -= 10
                db_eliminations.append("does not support the required hybrid search")

        if reqs.needs_multi_tenancy:
            if profile.supports_multi_tenancy:
                score += 10
                db_reasons.append("supports multi-tenancy")
            else:
                score -= 10
                db_eliminations.append("does not support the required multi-tenancy")

        # ── Bonus: Project stage ──
        if reqs.project_stage == ProjectStage.POC:
            if profile.operational_complexity <= 2:
                score += 10
                db_reasons.append("low friction for PoC")
        elif reqs.project_stage == ProjectStage.SCALE:
            if profile.max_scale_comfort > 10_000_000:
                score += 10
                db_reasons.append("ready for scale")

        scores[db_name] = score
        reasons[db_name] = db_reasons
        if db_eliminations:
            eliminations[db_name] = db_eliminations

    # ── Final ranking ──
    sorted_dbs = sorted(scores.items(), key=lambda x: x[1], reverse=True)

    primary_name = sorted_dbs[0][0]
    primary_score = sorted_dbs[0][1]
    alternative_name = sorted_dbs[1][0]
    alternative_score = sorted_dbs[1][1]

    score_gap = primary_score - alternative_score
    if score_gap > 30:
        confidence = "high"
    elif score_gap > 15:
        confidence = "medium"
    else:
        confidence = "low"

    eliminated_others = {
        name: eliminations.get(name, ["scored lower than the main options"])
        for name, _ in sorted_dbs[2:]
    }

    return Recommendation(
        primary=primary_name,
        primary_score=primary_score,
        primary_justification=reasons[primary_name],
        alternative=alternative_name,
        alternative_score=alternative_score,
        alternative_justification=reasons[alternative_name],
        eliminated=eliminated_others,
        warnings=warnings,
        confidence=confidence,
    )


def _team_to_complexity(capability: TeamCapability) -> int:
    """Map the team's capability to the complexity level it can handle."""
    return {
        TeamCapability.MINIMAL: 1,
        TeamCapability.MODERATE: 3,
        TeamCapability.STRONG: 5,
    }[capability]

Design decisions: additive scoring (each node adds/subtracts points), soft elimination (penalties instead of hard-filters), and confidence based on the score difference between the two best options.

Step 4: Generate the recommendation report

The report must be readable without additional context.

def generate_report(reqs: ProjectRequirements, rec: Recommendation) -> str:
    """Generate a readable report of the recommendation."""
    lines = []
    lines.append("=" * 70)
    lines.append(f"  DECISION TREE REPORT: {reqs.project_name}")
    lines.append("=" * 70)

    lines.append("\n📋 INPUTS")
    lines.append("-" * 40)
    inputs = [
        ("Vectors", f"{reqs.vector_count:,}"), ("Op. model", reqs.operational_preference.value),
        ("Budget", f"${reqs.monthly_budget_usd:,.0f}/mo"), ("p95 latency", f"{reqs.target_latency_p95_ms:.0f}ms"),
        ("Compliance", reqs.compliance.value), ("Team", reqs.team_capability.value),
        ("Stage", reqs.project_stage.value),
        ("Hybrid search", "yes" if reqs.needs_hybrid_search else "no"),
        ("Multi-tenancy", "yes" if reqs.needs_multi_tenancy else "no"),
    ]
    for label, val in inputs:
        lines.append(f"  {label:<20} {val}")

    if rec.warnings:
        lines.append(f"\n⚠️  ALERTS ({len(rec.warnings)})")
        lines.append("-" * 40)
        for w in rec.warnings:
            lines.append(f"  ⚠ {w}")

    lines.append(f"\n✅ PRIMARY RECOMMENDATION")
    lines.append("-" * 40)
    lines.append(f"  → {rec.primary}  (score: {rec.primary_score:.0f}, confidence: {rec.confidence})")
    for r in rec.primary_justification:
        lines.append(f"    • {r}")

    lines.append(f"\n🔄 ALTERNATIVE")
    lines.append("-" * 40)
    lines.append(f"  → {rec.alternative}  (score: {rec.alternative_score:.0f})")
    for r in rec.alternative_justification:
        lines.append(f"    • {r}")
    lines.append(f"\n  Prefer {rec.alternative} over {rec.primary} if:")
    _print_switch_conditions(lines, rec.primary, rec.alternative)

    lines.append(f"\n❌ DISCARDED OPTIONS")
    lines.append("-" * 40)
    for db_name, elim_reasons in rec.eliminated.items():
        lines.append(f"  {db_name}:")
        for r in elim_reasons:
            lines.append(f"    ✗ {r}")

    lines.append(f"\n📌 NEXT STEPS")
    lines.append("-" * 40)
    if rec.confidence == "low":
        lines.append("  1. Very close options — a comparative PoC is recommended.")
        lines.append("  2. Define success metrics before the test.")
    elif rec.confidence == "medium":
        lines.append(f"  1. Validate {rec.primary} with a 1-2 week PoC.")
        lines.append(f"  2. Keep {rec.alternative} as a documented plan B.")
    else:
        lines.append(f"  1. Proceed with {rec.primary} for the current phase.")
        lines.append(f"  2. Re-evaluate in 3 months or when scale changes 3x.")

    lines.append("\n" + "=" * 70)
    return "\n".join(lines)


def _print_switch_conditions(lines: list[str], primary: str, alternative: str):
    """Generate conditions for preferring the alternative."""
    switch_map = {
        ("ChromaDB", "Pinecone"): ["you need a production SLA", "no capacity to maintain your own infra"],
        ("Pinecone", "Weaviate"): ["hybrid search is a strong requirement", "you need a future self-hosted option"],
        ("Pinecone", "Qdrant"): ["you seek a better performance/cost ratio", "you want operational flexibility"],
        ("Weaviate", "Qdrant"): ["you prioritize simplicity over advanced features", "p95 latency is critical"],
        ("Qdrant", "Weaviate"): ["you need more mature hybrid search", "you want a broader ecosystem"],
        ("Milvus", "Weaviate"): ["the scale is under 10M vectors", "you prefer lower operational complexity"],
    }
    conditions = switch_map.get((primary, alternative), [
        "scale or compliance requirements change",
        "the team acquires different operational capacity",
    ])
    for condition in conditions:
        lines.append(f"    • {condition}")

Step 5: Run validation scenarios

Define at least 5 scenarios with different combinations of inputs.

def run_validation_scenarios() -> list[dict]:
    """Run 5 validation scenarios and return the results."""

    scenarios = [
        {"name": "MVP Startup - Quick PoC",
         "description": "Early startup, small team, no budget",
         "requirements": ProjectRequirements(
             project_name="MVP Startup", vector_count=50_000,
             operational_preference=OperationalModel.FLEXIBLE, monthly_budget_usd=0,
             target_latency_p95_ms=50, compliance=ComplianceLevel.NONE,
             team_capability=TeamCapability.MINIMAL, project_stage=ProjectStage.POC),
         "expected_primary": "ChromaDB"},

        {"name": "Growing SaaS",
         "description": "Validated product, needs SLA, moderate team",
         "requirements": ProjectRequirements(
             project_name="SaaS Growth", vector_count=2_000_000,
             operational_preference=OperationalModel.MANAGED, monthly_budget_usd=500,
             target_latency_p95_ms=15, compliance=ComplianceLevel.BASIC,
             team_capability=TeamCapability.MODERATE, project_stage=ProjectStage.PRODUCTION,
             needs_multi_tenancy=True),
         "expected_primary": "Pinecone"},

        {"name": "Enterprise - Strict compliance",
         "description": "Financial regulation, data residency, strong team",
         "requirements": ProjectRequirements(
             project_name="Enterprise FinTech", vector_count=10_000_000,
             operational_preference=OperationalModel.SELF_HOSTED, monthly_budget_usd=2000,
             target_latency_p95_ms=20, compliance=ComplianceLevel.STRICT,
             team_capability=TeamCapability.STRONG, project_stage=ProjectStage.SCALE,
             needs_hybrid_search=True, needs_multi_tenancy=True),
         "expected_primary": "Weaviate"},

        {"name": "Technical team - Performance first",
         "description": "Team with DevOps, prioritizes latency and control",
         "requirements": ProjectRequirements(
             project_name="Performance Team", vector_count=5_000_000,
             operational_preference=OperationalModel.FLEXIBLE, monthly_budget_usd=800,
             target_latency_p95_ms=7, compliance=ComplianceLevel.BASIC,
             team_capability=TeamCapability.STRONG, project_stage=ProjectStage.PRODUCTION,
             needs_hybrid_search=True),
         "expected_primary": "Qdrant"},

        {"name": "Global enterprise mega-scale",
         "description": "Global platform, >100M vectors, platform team",
         "requirements": ProjectRequirements(
             project_name="Global Platform", vector_count=200_000_000,
             operational_preference=OperationalModel.SELF_HOSTED, monthly_budget_usd=5000,
             target_latency_p95_ms=25, compliance=ComplianceLevel.STRICT,
             team_capability=TeamCapability.STRONG, project_stage=ProjectStage.SCALE,
             needs_hybrid_search=True, needs_multi_tenancy=True),
         "expected_primary": "Milvus"},
    ]

    results = []
    for scenario in scenarios:
        reqs = scenario["requirements"]
        rec = evaluate_decision_tree(reqs)
        match = rec.primary == scenario["expected_primary"]

        results.append({
            "name": scenario["name"],
            "description": scenario["description"],
            "expected": scenario["expected_primary"],
            "actual": rec.primary,
            "score": rec.primary_score,
            "confidence": rec.confidence,
            "match": match,
        })

    return results

Step 6: Generate a Mermaid diagram

def generate_mermaid_diagram() -> str:
    """Generate a Mermaid diagram of the decision tree."""
    diagram = """```mermaid
flowchart TD
    START([🚀 Start: Vector DB Selection]) --> N1

    N1{📊 Volume > 10M vectors?}
    N1 -- Yes --> N1_HIGH{🏢 Platform team?}
    N1 -- No --> N2

    N1_HIGH -- Yes --> MILVUS[🟣 Milvus]
    N1_HIGH -- No --> N1_WARN[⚠️ Strengthen team or<br/>reduce scope]

    N2{☁️ Operational preference?}
    N2 -- Managed --> N3_M
    N2 -- Self-hosted --> N3_S
    N2 -- Flexible --> N3_F

    N3_M{💰 Budget > $70/mo?}
    N3_M -- Yes --> N4_M{🔒 Strict compliance?}
    N3_M -- No --> CHROMADB_POC[🟢 ChromaDB<br/>PoC / development]

    N4_M -- Yes --> PINECONE_CHECK[Verify Pinecone<br/>data residency]
    N4_M -- No --> N5_M{🔍 Hybrid search?}

    N5_M -- Yes --> WEAVIATE_M[🔵 Weaviate Cloud]
    N5_M -- No --> PINECONE[🟡 Pinecone]

    N3_S{🔒 Strict compliance?}
    N3_S -- Yes --> N4_S{⚡ Latency < 10ms?}
    N3_S -- No --> N4_S2{🔍 Hybrid search?}
    N4_S -- Yes --> QDRANT_SH[🔴 Qdrant self-hosted]
    N4_S -- No --> WEAVIATE_SH[🔵 Weaviate self-hosted]
    N4_S2 -- Yes --> WEAVIATE_SH
    N4_S2 -- No --> QDRANT_SH

    N3_F{📊 Volume > 1M?}
    N3_F -- Yes --> N4_F{⚡ Latency < 10ms?}
    N3_F -- No --> N4_F2{💰 Low budget?}
    N4_F -- Yes --> QDRANT_F[🔴 Qdrant]
    N4_F -- No --> WEAVIATE_F[🔵 Weaviate]
    N4_F2 -- Yes --> CHROMADB[🟢 ChromaDB]
    N4_F2 -- No --> QDRANT_F

    style MILVUS fill:#9b59b6,color:#fff
    style PINECONE fill:#f1c40f,color:#000
    style CHROMADB fill:#2ecc71,color:#fff
    style CHROMADB_POC fill:#2ecc71,color:#fff
```"""
    return diagram

Step 7: Complete main script

Now put it all together in an executable flow.

def main():
    """Run the complete decision tree with validation."""

    print("=" * 70)
    print("  VECTOR DB DECISION TREE - Module 5 Final Project")
    print("=" * 70)

    # ── Step A: Run an individual scenario ──
    print("\n" + "─" * 70)
    print("  PART 1: Individual scenario")
    print("─" * 70)

    my_project = ProjectRequirements(
        project_name="My RAG Project",
        vector_count=800_000,
        operational_preference=OperationalModel.MANAGED,
        monthly_budget_usd=300,
        target_latency_p95_ms=20,
        compliance=ComplianceLevel.BASIC,
        team_capability=TeamCapability.MODERATE,
        project_stage=ProjectStage.PRODUCTION,
        needs_hybrid_search=False,
        needs_multi_tenancy=True,
    )

    rec = evaluate_decision_tree(my_project)
    report = generate_report(my_project, rec)
    print(report)

    # ── Step B: Scenario validation ──
    print("\n" + "─" * 70)
    print("  PART 2: Scenario validation")
    print("─" * 70)

    results = run_validation_scenarios()

    print(f"\n{'Scenario':<35} {'Expected':<12} {'Actual':<12} {'Score':>6} {'Conf':>8} {'OK':>4}")
    print("-" * 80)
    for r in results:
        status = "✅" if r["match"] else "⚠️"
        print(
            f"{r['name']:<35} {r['expected']:<12} {r['actual']:<12} "
            f"{r['score']:>6.0f} {r['confidence']:>8} {status:>4}"
        )

    passed = sum(1 for r in results if r["match"])
    total = len(results)
    print(f"\nResult: {passed}/{total} scenarios validated")

    if passed < total:
        print("\n⚠️  Scenarios with a discrepancy (check whether they make sense):")
        for r in results:
            if not r["match"]:
                print(f"  • {r['name']}: expected {r['expected']}, got {r['actual']}")

    # ── Step C: Mermaid diagram ──
    print("\n" + "─" * 70)
    print("  PART 3: Mermaid diagram")
    print("─" * 70)
    print("\nCopy into a .md file or into https://mermaid.live/\n")
    print(generate_mermaid_diagram())

    # ── Final summary ──
    print("\n" + "=" * 70)
    print(f"  PROJECT COMPLETED")
    print(f"  ✅ {len(VECTOR_DB_PROFILES)} providers | {passed}/{total} scenarios | Mermaid ready")
    print(f"  → Next: Module 6 (decision matrix with formal scoring)")
    print("=" * 70)


if __name__ == "__main__":
    main()

📊 Expected output

When you run the script, you'll see a report with this structure:

✅ PRIMARY RECOMMENDATION
  → Weaviate  (score: 105, confidence: low)
  Justification:
    • comfortable scale (800,000 vectors)
    • supports the managed model
    • generous budget for this option
    • latency with ample margin
    • supports multi-tenancy

🔄 ALTERNATIVE
  → Qdrant  (score: 105)

  Note: Weaviate and Qdrant tie at 105, which is why confidence is "low".
  The tree is telling you to run a comparative PoC (see Troubleshooting).

❌ DISCARDED OPTIONS
  ChromaDB: ✗ no managed option
  Milvus:   ✗ operational complexity (5/5) exceeds the team's capacity

SCENARIO VALIDATION
Scenario                            Expected     Actual       OK
MVP Startup - Quick PoC             ChromaDB     ChromaDB     ✅
Growing SaaS                        Pinecone     Pinecone     ✅
Enterprise - Strict compliance      Weaviate     Weaviate     ✅
Technical team - Performance first  Qdrant       Qdrant       ✅
Global enterprise mega-scale        Milvus       Milvus       ✅

Result: 5/5 scenarios validated

🔧 Project troubleshooting

"My tree always recommends the same DB"

Review the weights of each node. It's likely that a single criterion (like scale or operational model) is dominating the score. Adjust the points so that no individual node represents more than 25% of the maximum possible score. Also verify that your test scenarios have genuinely different inputs — if they all have team_capability=MINIMAL, they'll naturally converge.

"Two options have an almost identical score"

This is information, not an error. When the confidence is "low", the tree is telling you that you need more data to decide. Options:

  1. Add a differentiating criterion (e.g., community support, documentation).
  2. Run a comparative PoC with the two finalists.
  3. Use Module 6 (scoring matrix) for a more granular analysis.

"I don't know what values to put in the inputs"

Start with what you know and mark what you're estimating. Typical values: PoC (10k-100k vectors, $0, flexible latency), early production (100k-2M, $100-500, p95 <30ms), scale (>2M, >$500, p95 <15ms).

"I want to add a provider that isn't in the list"

Create a new VectorDBProfile and add it to VECTOR_DB_PROFILES. The engine will automatically evaluate it against all existing nodes.

"The recommendation doesn't match what I would have chosen"

Good sign. Review which criterion the tree values that you don't, or vice versa. Disagreements reveal implicit assumptions worth making explicit. Adjust weights to your context and document why.

🏋️ Post-project exercises

Exercise 1: Add a vendor lock-in dimension

Extend the tree with a portability criterion that penalizes options with high lock-in.

Hints:

  • Add a portability_concern: bool field to ProjectRequirements.
  • Add a lock_in_risk: int field (1-5) to VectorDBProfile.
  • Create a new node in evaluate_decision_tree.
See solution 
# In ProjectRequirements, add:
portability_concern: bool = False

# In VectorDBProfile, add:
lock_in_risk: int = 1  # 1=low, 5=high

# Update profiles:
# ChromaDB:  lock_in_risk=1 (open source, standard)
# Pinecone:  lock_in_risk=5 (proprietary API, no self-hosted)
# Weaviate:  lock_in_risk=2 (open source, standard API)
# Qdrant:    lock_in_risk=2 (open source, standard gRPC)
# Milvus:    lock_in_risk=2 (open source, standard API)

# In evaluate_decision_tree, add a node:

# ── Node 7: Vendor lock-in ──
if reqs.portability_concern:
    if profile.lock_in_risk <= 2:
        score += 10
        db_reasons.append("low vendor lock-in risk")
    elif profile.lock_in_risk >= 4:
        score -= 15
        db_eliminations.append(
            f"high lock-in risk ({profile.lock_in_risk}/5) "
            f"with a portability requirement"
        )

Exercise 2: Generate a side-by-side comparison

Create a function that takes two different ProjectRequirements and shows how the recommendation changes between scenarios.

Hints:

  • The function takes two ProjectRequirements objects.
  • Run evaluate_decision_tree for each one.
  • Print a comparison table with highlighted diffs.
See solution 
def compare_scenarios(reqs_a: ProjectRequirements, reqs_b: ProjectRequirements) -> str:
    """Compare recommendations between two scenarios."""
    rec_a = evaluate_decision_tree(reqs_a)
    rec_b = evaluate_decision_tree(reqs_b)

    lines = []
    lines.append("=" * 70)
    lines.append("  SCENARIO COMPARISON")
    lines.append("=" * 70)

    lines.append(f"\n{'Dimension':<25} {'Scenario A':<22} {'Scenario B':<22}")
    lines.append("-" * 70)
    lines.append(f"{'Project':<25} {reqs_a.project_name:<22} {reqs_b.project_name:<22}")
    lines.append(f"{'Vectors':<25} {reqs_a.vector_count:<22,} {reqs_b.vector_count:<22,}")
    lines.append(f"{'Budget':<25} {'$'+str(int(reqs_a.monthly_budget_usd)):<22} {'$'+str(int(reqs_b.monthly_budget_usd)):<22}")
    lines.append(f"{'Op. model':<25} {reqs_a.operational_preference.value:<22} {reqs_b.operational_preference.value:<22}")

    lines.append(f"\n{'RESULT':<25}")
    lines.append("-" * 70)

    marker_a = " ←" if rec_a.primary != rec_b.primary else ""
    marker_b = " ←" if rec_a.primary != rec_b.primary else ""
    lines.append(f"{'Recommendation':<25} {rec_a.primary + marker_a:<22} {rec_b.primary + marker_b:<22}")
    lines.append(f"{'Score':<25} {rec_a.primary_score:<22.0f} {rec_b.primary_score:<22.0f}")
    lines.append(f"{'Confidence':<25} {rec_a.confidence:<22} {rec_b.confidence:<22}")
    lines.append(f"{'Alternative':<25} {rec_a.alternative:<22} {rec_b.alternative:<22}")

    if rec_a.primary != rec_b.primary:
        lines.append(f"\n💡 The recommendation changed from {rec_a.primary} to {rec_b.primary}.")
        lines.append("   Review which input caused the change to understand the sensitivity.")

    return "\n".join(lines)

# Usage: compare_scenarios(reqs_mvp, reqs_production)

Exercise 3: Sensitivity — which input changes the decision most?

Create a function that takes a base ProjectRequirements and varies one input at a time to identify which are the most sensitive (the ones that change the ranking most).

Hints:

  • Use dataclasses.replace() to clone requirements with one changed field.
  • Iterate over predefined variations of each field.
  • Compare whether primary changes or whether confidence changes.
See solution 
from dataclasses import replace


def sensitivity_analysis(base_reqs: ProjectRequirements) -> str:
    """Analyze which inputs are most sensitive to a change of recommendation."""
    base_rec = evaluate_decision_tree(base_reqs)
    lines = []
    lines.append("=" * 70)
    lines.append(f"  SENSITIVITY ANALYSIS: {base_reqs.project_name}")
    lines.append(f"  Base recommendation: {base_rec.primary} (score: {base_rec.primary_score:.0f})")
    lines.append("=" * 70)

    variations = [
        ("vector_count", [10_000, 500_000, 2_000_000, 10_000_000, 100_000_000]),
        ("monthly_budget_usd", [0, 50, 200, 500, 2000]),
        ("target_latency_p95_ms", [5, 10, 20, 50, 100]),
        ("operational_preference", list(OperationalModel)),
        ("compliance", list(ComplianceLevel)),
        ("team_capability", list(TeamCapability)),
    ]

    for field_name, values in variations:
        lines.append(f"\n📊 Varying: {field_name}")
        lines.append(f"   {'Value':<25} {'Recommendation':<15} {'Score':>6} {'Changed?':>8}")
        lines.append("   " + "-" * 58)

        for val in values:
            try:
                varied_reqs = replace(base_reqs, **{field_name: val})
                varied_rec = evaluate_decision_tree(varied_reqs)
                changed = "⚠️ YES" if varied_rec.primary != base_rec.primary else "  no"
                display_val = f"{val:,}" if isinstance(val, (int, float)) else val.value
                lines.append(
                    f"   {str(display_val):<25} {varied_rec.primary:<15} "
                    f"{varied_rec.primary_score:>6.0f} {changed:>8}"
                )
            except (TypeError, ValueError):
                continue

    return "\n".join(lines)

# Usage: print(sensitivity_analysis(my_base_requirements))

The output will show a table for each varied dimension. The fields marked "YES" in the "Changed?" column are the most sensitive inputs — the ones you must estimate with the most care.

✅ Completeness checklist

Verify all these points before considering the project done:

Code structure:

  • ProjectRequirements has the 6 mandatory dimensions.
  • VectorDBProfile covers the 5 providers with coherent data.
  • evaluate_decision_tree() produces a numeric score and a complete Recommendation.

Decision quality:

  • The tree doesn't depend on a single criterion to decide.
  • The justification explains why yes (primary) and why no (discarded).
  • The alternative includes conditions for when to prefer it.

Validation:

  • At least 5 scenarios run (PoC, production, enterprise).
  • Discrepancies between expected and actual are explained.

Output:

  • Report readable without additional context, Mermaid renders correctly.
  • Numeric scores compatible with the Module 6 matrix.

Summary

  • You built a programmatic decision engine that transforms requirements into a justified recommendation.
  • Each node of the tree evaluates a concrete dimension: scale, operations, budget, latency, compliance, and team capability.
  • The system produces a primary recommendation + alternative, explaining why the other options were discarded.
  • You validated with 5 different scenarios covering everything from PoC to global enterprise.
  • The Mermaid diagram lets you communicate the logic to non-technical stakeholders.
  • The result's confidence ("high", "medium", "low") indicates when you need more analysis before deciding.
  • This project closes Module 5 and feeds directly into Module 6 where you'll formalize the decision with a scoring matrix.
  • You have extension exercises (lock-in, comparison, sensitivity) to deepen the analysis.

📚 Additional resources

Estimated time: 35-45 minutes
Next module: ../../module-06-decision-matrix/en/01-module-introduction-6.md

← PreviousCapsule 07: Anti-patterns When Choosing a Vector DBNext →Module 6: Decision Matrix for AI Engineers