Capstone 2 — Domain B: Product Catalog & Contract Q&A

mkdir catalog-contract-rag && cd catalog-contract-rag
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install "anthropic>=0.30.0" pytest
export ANTHROPIC_API_KEY=your-key-here             # Windows: set ANTHROPIC_API_KEY=your-key-here

{
  "documents": [
    {
      "doc_id": "CATALOG-2024-Q1",
      "type": "product_catalog",
      "sections": [
        {
          "sku": "SKU-4892",
          "name": "Industrial Valve Assembly - 4 inch",
          "category": "Flow Control",
          "list_price": 595.00,
          "specs": {
            "material": "316 Stainless Steel",
            "pressure_rating": "150 PSI",
            "temperature_range": "-20F to 400F",
            "connection": "Flanged ANSI 150",
            "lead_time_days": 14
          }
        },
        {
          "sku": "SKU-7721",
          "name": "Pressure Gauge - Digital, 0-300 PSI",
          "category": "Instrumentation",
          "list_price": 189.00,
          "specs": {
            "display": "4-digit LCD backlit",
            "accuracy": "+/- 0.25% full scale",
            "power": "2x AA battery, 18-month life",
            "connection": "1/4 NPT male",
            "lead_time_days": 7
          }
        },
        {
          "sku": "SKU-2210",
          "name": "Stainless Steel Fitting Kit - 4 inch",
          "category": "Fittings",
          "list_price": 125.00,
          "specs": {
            "material": "316 Stainless Steel",
            "includes": "2 elbows, 2 tees, 4 couplings, 8 nipples",
            "pressure_rating": "150 PSI",
            "lead_time_days": 5
          }
        }
      ]
    },
    {
      "doc_id": "CONTRACT-ACME-2024",
      "type": "customer_contract",
      "customer": "Acme Manufacturing",
      "effective_date": "2024-01-01",
      "expiration_date": "2024-12-31",
      "pricing_tiers": [
        {
          "sku": "SKU-4892",
          "tier_1": {"min_qty": 1, "max_qty": 49, "unit_price": 525.00},
          "tier_2": {"min_qty": 50, "max_qty": 199, "unit_price": 495.00},
          "tier_3": {"min_qty": 200, "max_qty": null, "unit_price": 465.00}
        },
        {
          "sku": "SKU-7721",
          "tier_1": {"min_qty": 1, "max_qty": 24, "unit_price": 165.00},
          "tier_2": {"min_qty": 25, "max_qty": null, "unit_price": 145.00}
        }
      ],
      "terms": {
        "payment": "Net 45",
        "shipping": "FOB Destination",
        "warranty": "24 months",
        "annual_minimum": 50000.00
      }
    },
    {
      "doc_id": "CONTRACT-BETA-2024",
      "type": "customer_contract",
      "customer": "Beta Industrial",
      "effective_date": "2024-03-01",
      "expiration_date": "2025-02-28",
      "pricing_tiers": [
        {
          "sku": "SKU-4892",
          "tier_1": {"min_qty": 1, "max_qty": 99, "unit_price": 510.00},
          "tier_2": {"min_qty": 100, "max_qty": null, "unit_price": 480.00}
        }
      ],
      "terms": {
        "payment": "Net 30",
        "shipping": "FOB Origin",
        "warranty": "12 months",
        "annual_minimum": 25000.00
      }
    },
    {
      "doc_id": "CONTRACT-GAMMA-2024",
      "type": "customer_contract",
      "customer": "Gamma Technologies",
      "effective_date": "2024-02-15",
      "expiration_date": "2025-02-14",
      "pricing_tiers": [
        {
          "sku": "SKU-7721",
          "tier_1": {"min_qty": 1, "max_qty": 19, "unit_price": 172.00},
          "tier_2": {"min_qty": 20, "max_qty": null, "unit_price": 155.00}
        }
      ],
      "terms": {
        "payment": "Net 45",
        "shipping": "FOB Destination",
        "warranty": "36 months",
        "annual_minimum": 35000.00
      }
    },
    {
      "doc_id": "CONTRACT-DELTA-2024",
      "type": "customer_contract",
      "customer": "Delta Logistics",
      "effective_date": "2024-04-01",
      "expiration_date": "2025-03-31",
      "pricing_tiers": [
        {
          "sku": "SKU-2210",
          "tier_1": {"min_qty": 1, "max_qty": 99, "unit_price": 108.00},
          "tier_2": {"min_qty": 100, "max_qty": null, "unit_price": 95.00}
        },
        {
          "sku": "SKU-4892",
          "tier_1": {"min_qty": 1, "max_qty": null, "unit_price": 540.00}
        }
      ],
      "terms": {
        "payment": "Net 30",
        "shipping": "FOB Origin",
        "warranty": "12 months",
        "annual_minimum": 20000.00
      }
    },
    {
      "doc_id": "CONTRACT-EPSILON-2024",
      "type": "customer_contract",
      "customer": "Epsilon Engineering",
      "effective_date": "2024-01-15",
      "expiration_date": "2026-01-14",
      "pricing_tiers": [
        {
          "sku": "SKU-4892",
          "tier_1": {"min_qty": 1, "max_qty": 74, "unit_price": 515.00},
          "tier_2": {"min_qty": 75, "max_qty": null, "unit_price": 475.00}
        },
        {
          "sku": "SKU-7721",
          "tier_1": {"min_qty": 1, "max_qty": 49, "unit_price": 160.00},
          "tier_2": {"min_qty": 50, "max_qty": null, "unit_price": 140.00}
        },
        {
          "sku": "SKU-2210",
          "tier_1": {"min_qty": 1, "max_qty": 149, "unit_price": 105.00},
          "tier_2": {"min_qty": 150, "max_qty": null, "unit_price": 90.00}
        }
      ],
      "terms": {
        "payment": "Net 60",
        "shipping": "FOB Destination",
        "warranty": "60 months",
        "annual_minimum": 200000.00
      }
    }
  ]
}

mkdir -p data    # Windows: mkdir data
# Paste the JSON from the Mock Data section into data/catalog_contracts.json
python -c "import json; d=json.load(open('data/catalog_contracts.json')); print(f'{len(d[\"documents\"])} documents loaded')"

# Quick test: create a store, add a doc, search it
python -c "
from vector_store import MockVectorStore
store = MockVectorStore()
store.add(chunk_id='test-1', content='Industrial valve assembly 4 inch', metadata={'sku': 'SKU-4892'})
results = store.search('valve', top_k=1)
print(f'Found: {results[0][\"text\"][:40]}... score={results[0][\"score\"]}')
"

python -c "
from vector_store import MockVectorStore
from ingest import ingest_corpus
store = MockVectorStore()
count = ingest_corpus('data/catalog_contracts.json', store)
print(f'Ingested {count} chunks')
# Test: search for Acme contract pricing
results = store.search('SKU-4892 pricing', top_k=2, filters={'customer': 'Acme Manufacturing'})
for r in results:
    print(f'  [{r[\"metadata\"].get(\"doc_type\",\"\")}] {r[\"text\"][:60]}...')
"

export ANTHROPIC_API_KEY=your-key-here
python agent.py

pytest test_rag.py -v

"""vector_store.py — TF-IDF Mock Vector Store

WHAT: A self-contained vector store using TF-IDF similarity.
WHY:  Free, fast, no external API needed. Demonstrates core
      retrieval concepts (tokenize, embed, search) without
      adding cost or complexity.
GOTCHA: TF-IDF is a bag-of-words model — it ignores word order.
        For production, use neural embeddings (e.g., Voyage AI).
"""

import math
import re
from collections import Counter


class MockVectorStore:
    def __init__(self):
        self.documents = []
        self.vectors = []
        self.idf = {}

    def _tokenize(self, text):
        """Split text into lowercase word tokens."""
        return re.findall(r'\b\w+\b', text.lower())

    def _compute_tf(self, tokens):
        """Compute term frequency: count / total tokens."""
        count = Counter(tokens)
        total = len(tokens)
        return {t: c / total for t, c in count.items()}

    def _build_idf(self):
        """Recompute IDF across all documents.
        IDF = log(total_docs / (1 + docs_containing_term))
        The +1 prevents division by zero."""
        doc_count = len(self.documents)
        term_docs = {}
        for doc in self.documents:
            tokens = set(self._tokenize(doc["text"]))
            for t in tokens:
                term_docs[t] = term_docs.get(t, 0) + 1
        self.idf = {t: math.log(doc_count / (1 + c)) for t, c in term_docs.items()}

    def _vectorize(self, text):
        """Convert text to a TF-IDF vector (sparse dict)."""
        tokens = self._tokenize(text)
        tf = self._compute_tf(tokens)
        return {t: tf_val * self.idf.get(t, 0) for t, tf_val in tf.items()}

    def add(self, chunk_id=None, content="", text=None, metadata=None):
        """Add a document to the store.

        Args:
            chunk_id: Optional unique identifier for the chunk.
            content: The document text to index (preferred kwarg name).
            text: Alias for content (used in simpler call signatures).
            metadata: Dict of metadata (doc_type, customer, sku, etc.)
                      Used for filtered retrieval.
        """
        doc_text = content or text or ""
        self.documents.append({"text": doc_text, "metadata": metadata or {},
                               "chunk_id": chunk_id})
        # Rebuild IDF and re-vectorize all docs (fine for small corpora)
        self._build_idf()
        self.vectors = [self._vectorize(d["text"]) for d in self.documents]

    def search(self, query, top_k=3, filters=None):
        """Search for documents similar to query.

        Args:
            query: Natural language search string.
            top_k: Max results to return (default 3).
            filters: Optional dict of metadata key-value pairs.
                     Only documents matching ALL filters are returned.

        Returns:
            List of dicts with text, metadata, and similarity score.
        """
        if not query.strip():
            return [{"error": "EMPTY_QUERY", "message": "Query cannot be empty"}]

        query_vec = self._vectorize(query)
        results = []

        for i, doc in enumerate(self.documents):
            # ── Metadata filtering ────────────────────────────
            if filters:
                if not all(doc["metadata"].get(k) == v for k, v in filters.items()):
                    continue

            # ── Cosine similarity ─────────────────────────────
            doc_vec = self.vectors[i]
            all_terms = set(list(query_vec.keys()) + list(doc_vec.keys()))
            dot = sum(query_vec.get(t, 0) * doc_vec.get(t, 0) for t in all_terms)
            mag_q = math.sqrt(sum(v ** 2 for v in query_vec.values())) or 1
            mag_d = math.sqrt(sum(v ** 2 for v in doc_vec.values())) or 1
            score = dot / (mag_q * mag_d)

            results.append({
                "text": doc["text"],
                "content": doc["text"],
                "doc_id": doc["metadata"].get("doc_id", ""),
                "chunk_id": doc.get("chunk_id", ""),
                "metadata": doc["metadata"],
                "score": round(score, 4),
            })

        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:top_k]

"""ingest.py — Dual-source ingestion for catalog + contracts.

WHAT: Chunks product catalog and customer contracts differently,
      preserving metadata for filtered retrieval.
WHY:  Contract pricing must override catalog pricing. Metadata
      filters (doc_type, customer, sku) enable this priority.
"""

import json
from vector_store import MockVectorStore

def ingest_corpus(corpus_path: str, store: MockVectorStore) -> int:
    with open(corpus_path) as f:
        corpus = json.load(f)

    count = 0
    for doc in corpus["documents"]:
        doc_id = doc["doc_id"]
        doc_type = doc["type"]

        if doc_type == "product_catalog":
            # ── WHAT: One chunk per SKU ────────────────────────
            # WHY: Keep specs + price together so retrieval
            #      returns complete product info in one chunk.
            for product in doc["sections"]:
                sku = product["sku"]
                specs = product.get("specs", {})
                content = (
                    f"Product: {product['name']} (SKU: {sku}). "
                    f"Category: {product['category']}. "
                    f"List Price: ${product['list_price']:.2f}. "
                    f"Specs: {', '.join(f'{k}: {v}' for k, v in specs.items())}."
                )
                store.add(
                    chunk_id=f"{doc_id}:{sku}",
                    content=content,
                    metadata={
                        "doc_id": doc_id, "doc_type": "product_catalog",
                        "sku": sku, "category": product["category"],
                        "list_price": product["list_price"],
                    },
                )
                count += 1

        elif doc_type == "customer_contract":
            customer = doc["customer"]
            # ── WHAT: One chunk per SKU pricing tier ───────────
            # WHY: Separating tiers enables precise retrieval
            #      when the query mentions a specific quantity.
            for tier_entry in doc.get("pricing_tiers", []):
                sku = tier_entry["sku"]
                tiers = []
                for key in ["tier_1", "tier_2", "tier_3"]:
                    t = tier_entry.get(key)
                    if t:
                        max_label = str(t["max_qty"]) if t["max_qty"] else "unlimited"
                        tiers.append(f"{t['min_qty']}-{max_label} units: ${t['unit_price']:.2f}")
                content = (
                    f"Contract pricing for {customer}: SKU {sku}. "
                    f"Tiers: {'; '.join(tiers)}. "
                    f"Contract: {doc_id}, effective {doc['effective_date']} "
                    f"to {doc['expiration_date']}."
                )
                store.add(
                    chunk_id=f"{doc_id}:{sku}:pricing",
                    content=content,
                    metadata={
                        "doc_id": doc_id, "doc_type": "customer_contract",
                        "customer": customer, "sku": sku,
                        "effective_date": doc["effective_date"],
                        "expiration_date": doc["expiration_date"],
                    },
                )
                count += 1

            # ── WHAT: One chunk for contract terms ─────────────
            terms = doc.get("terms", {})
            content = (
                f"Contract terms for {customer}: "
                f"Payment: {terms.get('payment', 'N/A')}. "
                f"Shipping: {terms.get('shipping', 'N/A')}. "
                f"Warranty: {terms.get('warranty', 'N/A')}. "
                f"Annual minimum: ${terms.get('annual_minimum', 0):,.2f}. "
                f"Contract: {doc_id}, {doc['effective_date']} to {doc['expiration_date']}."
            )
            store.add(
                chunk_id=f"{doc_id}:terms",
                content=content,
                metadata={
                    "doc_id": doc_id, "doc_type": "customer_contract",
                    "customer": customer, "section": "terms",
                },
            )
            count += 1

    return count

"""agent.py — Product Catalog & Contract RAG Agent (Capstone 2-B)

Usage:
  export ANTHROPIC_API_KEY=your-key-here
  python agent.py
"""

import json
import anthropic
from vector_store import MockVectorStore
from ingest import ingest_corpus

client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
store = MockVectorStore()

TOOLS = [
    {
        "name": "search_catalog_knowledge_base",
        "description": (
            "Search the product catalog and customer contract knowledge base. "
            "Returns ranked results with similarity scores and metadata. "
            "Use filters to narrow by doc_type (product_catalog or customer_contract), "
            "customer name, or SKU. Always search when the user asks about "
            "pricing, specs, availability, contract terms, or lead times."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Natural language search query",
                },
                "top_k": {
                    "type": "integer",
                    "description": "Number of results (default: 5)",
                },
                "filters": {
                    "type": "object",
                    "description": "Optional filters",
                    "properties": {
                        "doc_type": {
                            "type": "string",
                            "enum": ["product_catalog", "customer_contract"],
                            "description": "Filter by document type",
                        },
                        "customer": {
                            "type": "string",
                            "description": "Filter by customer name",
                        },
                        "sku": {
                            "type": "string",
                            "description": "Filter by SKU",
                        },
                    },
                },
            },
            "required": ["query"],
        },
    },
]

SYSTEM_PROMPT = """You are a B2B product and pricing assistant. You help \
sales reps answer questions about products, pricing, and contract terms \
by searching the product catalog and customer contract database.

CRITICAL PRICING RULE:
- When a customer has a contract, ALWAYS use contract pricing, not list price.
- Only show list price when: (a) the customer has no contract, or \
  (b) the specific SKU is not covered by the customer's contract.
- If the user asks about pricing, ALWAYS search for the customer's contract first.

Citation Rules:
- Cite every fact: [Source: DOC_ID] or [Source: DOC_ID, SKU/Section]
- Clearly label whether a price comes from a CONTRACT or CATALOG.
- When showing tiered pricing, list all tiers.

Other Rules:
- You provide information only — you cannot modify prices or contracts.
- If asked about a customer you don't have data for, say so clearly.
- Include specs, lead times, and availability when relevant."""


def handle_search(args: dict) -> str:
    query = args.get("query", "")
    top_k = args.get("top_k", 5)
    filters = args.get("filters")
    results = store.search(query, top_k=top_k, filters=filters)
    return json.dumps(results, indent=2)


def chat(user_message: str, history: list) -> str:
    history.append({"role": "user", "content": user_message})
    while True:
        response = client.messages.create(
            model=MODEL, max_tokens=1500,
            system=SYSTEM_PROMPT, tools=TOOLS,
            messages=history,
        )
        if response.stop_reason == "tool_use":
            history.append({"role": "assistant", "content": response.content})
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    try:
                        result = handle_search(block.input)
                    except Exception as e:
                        result = json.dumps({"error": "INDEX_UNAVAILABLE", "message": str(e)})
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
            history.append({"role": "user", "content": tool_results})
            continue

        history.append({"role": "assistant", "content": response.content})
        return "\n".join(b.text for b in response.content if hasattr(b, "text"))


def main():
    print("Loading product catalog and contracts...")
    count = ingest_corpus("data/catalog_contracts.json", store)
    print(f"Ingested {count} chunks.")
    print("=" * 60)
    print("  Product Catalog & Contract Q&A — Capstone 2-B")
    print("  Ask about pricing, specs, contracts, lead times.")
    print("  Type 'quit' to exit.")
    print("=" * 60)

    history = []
    while True:
        user_input = input("\nYou: ").strip()
        if not user_input: continue
        if user_input.lower() in ("quit", "exit", "q"): break
        try:
            print(f"\nAgent: {chat(user_input, history)}")
        except anthropic.APIError as e:
            print(f"\n[API Error] {e.message}")

if __name__ == "__main__":
    main()

"""test_rag.py — Tests for the Product Catalog & Contract RAG."""

import pytest
from vector_store import MockVectorStore
from ingest import ingest_corpus

@pytest.fixture
def loaded_store():
    store = MockVectorStore()
    ingest_corpus("data/catalog_contracts.json", store)
    return store

class TestRetrieval:
    def test_contract_found_for_acme(self, loaded_store):
        results = loaded_store.search("SKU-4892 pricing Acme",
                                       filters={"customer": "Acme Manufacturing"})
        assert len(results) > 0
        assert any("CONTRACT-ACME" in r["doc_id"] for r in results)

    def test_catalog_fallback_for_unknown_customer(self, loaded_store):
        results = loaded_store.search("SKU-4892 pricing",
                                       filters={"doc_type": "customer_contract",
                                                "customer": "Gamma Corp"})
        assert len(results) == 0  # No contract → empty

    def test_catalog_specs_returned(self, loaded_store):
        results = loaded_store.search("valve assembly specs",
                                       filters={"doc_type": "product_catalog"})
        assert len(results) > 0
        assert "316" in results[0]["content"]  # Stainless steel

    def test_contract_terms_returned(self, loaded_store):
        results = loaded_store.search("Acme payment terms",
                                       filters={"customer": "Acme Manufacturing"})
        assert any("Net 45" in r["content"] for r in results)

    def test_empty_query_error(self, loaded_store):
        results = loaded_store.search("")
        assert results[0].get("error") == "EMPTY_QUERY"

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

# verify.py — End-to-end smoke test for Capstone 2-B
from vector_store import MockVectorStore
from ingest import ingest_corpus

store = MockVectorStore()
count = ingest_corpus("data/catalog_contracts.json", store)

tests = [
    ("Contract pricing for Acme",
     {"query": "SKU-4892 pricing", "filters": {"customer": "Acme Manufacturing"}},
     "CONTRACT-ACME"),
    ("Catalog fallback for unknown customer",
     {"query": "SKU-4892 pricing", "filters": {"customer": "Unknown Corp", "doc_type": "customer_contract"}},
     None),  # Expect empty
    ("Catalog specs returned",
     {"query": "valve assembly specs", "filters": {"doc_type": "product_catalog"}},
     "CATALOG"),
]

print(f"Ingested {count} chunks.")
print("=== End-to-End Verification ===\n")
passed = 0
for name, search_args, expected in tests:
    results = store.search(search_args["query"], top_k=3,
                           filters=search_args.get("filters"))
    if expected is None:
        ok = len(results) == 0
    else:
        ok = any(expected in r.get("doc_id", "") for r in results)
    status = "PASS" if ok else "FAIL"
    if ok: passed += 1
    print(f"[{status}] {name}")

print(f"\nResult: {passed}/{len(tests)} tests passed.")
if passed == len(tests):
    print("All tests passed — your RAG pipeline is working correctly!")

python verify.py