{ "nbformat": 4, "nbformat_minor": 5, "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.13.0" }, "blog_metadata": { "topic": "Why Azure Cosmos DB Indexing Policies Still Decide AI App Performance", "slug": "why-azure-cosmos-db-indexing-policies-still-decide-ai-app-pe", "generated_by": "LinkedIn Post Generator + Azure OpenAI", "generated_at": "2026-09-22T21:57:27.645Z" } }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Why Azure Cosmos DB Indexing Policies Still Decide AI App Performance\n", "\n", "This notebook turns the blog post into a hands-on validation workflow for Azure Cosmos DB retrieval design. The focus is simple: indexing policy, query shape, and workload boundaries often decide latency, RU consumption, and retrieval correctness before the LLM ever helps.\n", "\n", "You will inspect retrieval flow, model an indexing policy, compare broad versus constrained query patterns, and review operational signals that connect app complaints to database behavior." ] }, { "cell_type": "code", "metadata": {}, "source": [ "%pip install azure-cosmos python-dotenv pandas" ], "execution_count": null, "outputs": [] }, { "cell_type": "code", "metadata": {}, "source": [ "import os\n", "import json\n", "from pprint import pprint\n", "from textwrap import dedent\n", "\n", "import pandas as pd\n", "\n", "try:\n", " from dotenv import load_dotenv\n", "except ImportError:\n", " load_dotenv = None\n", "\n", "try:\n", " from azure.cosmos import CosmosClient, PartitionKey\n", "except ImportError:\n", " CosmosClient = None\n", " PartitionKey = None\n", "\n", "if load_dotenv:\n", " load_dotenv()" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Retrieval path overview\n", "\n", "This cell converts the blog's retrieval flow into a simple Python structure so you can validate where indexing policy affects behavior. The key point is that indexing influences candidate generation, metadata filtering, and projection before the model call." ] }, { "cell_type": "code", "metadata": {}, "source": [ "retrieval_flow = {\n", " \"steps\": [\n", " \"AI app request\",\n", " \"Embed query\",\n", " \"Cosmos DB container\",\n", " \"Vector similarity candidate set\",\n", " \"Metadata filters: tenant, docType, status\",\n", " \"Projection + top K\",\n", " \"LLM grounding context\",\n", " ],\n", " \"indexing_influence\": {\n", " \"candidate_generation\": \"indexing policy\",\n", " \"metadata_filtering\": \"included/excluded paths\",\n", " \"projection_and_sort\": \"composite indexes or aligned scalar indexes\",\n", " },\n", "}\n", "\n", "print(\"Retrieval steps:\")\n", "for i, step in enumerate(retrieval_flow[\"steps\"], start=1):\n", " print(f\"{i}. {step}\")\n", "\n", "print(\"\\nWhere indexing matters:\")\n", "pprint(retrieval_flow[\"indexing_influence\"])" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Indexing policy tuned for AI retrieval\n", "\n", "This example makes the retrieval contract explicit. It keeps likely filter and projection fields queryable while excluding large payloads that are usually expensive to index for ordinary scalar access." ] }, { "cell_type": "code", "metadata": {}, "source": [ "indexing_policy = {\n", " \"indexingMode\": \"consistent\",\n", " \"automatic\": True,\n", " \"includedPaths\": [\n", " {\"path\": \"/tenantId/?\"},\n", " {\"path\": \"/docType/?\"},\n", " {\"path\": \"/status/?\"},\n", " {\"path\": \"/title/?\"},\n", " {\"path\": \"/sourceUrl/?\"},\n", " ],\n", " \"excludedPaths\": [\n", " {\"path\": \"/chunk/*\"},\n", " {\"path\": \"/rawText/*\"},\n", " {\"path\": \"/embedding/*\"},\n", " {\"path\": \"/*\"},\n", " ],\n", " \"vectorIndexes\": [\n", " {\"path\": \"/embedding\", \"type\": \"diskANN\"}\n", " ],\n", "}\n", "\n", "print(\"Included paths:\")\n", "pprint(indexing_policy[\"includedPaths\"])\n", "print(\"\\nExcluded paths:\")\n", "pprint(indexing_policy[\"excludedPaths\"])\n", "print(\"\\nVector indexes:\")\n", "pprint(indexing_policy[\"vectorIndexes\"])" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Validate indexing intent against a sample document\n", "\n", "This cell checks which fields in a representative document are intended for filtering or projection versus which fields are likely ingestion baggage. It is a lightweight way to review whether the policy matches the actual retrieval path." ] }, { "cell_type": "code", "metadata": {}, "source": [ "sample_doc = {\n", " \"id\": \"doc-001#chunk-01\",\n", " \"tenantId\": \"contoso\",\n", " \"docType\": \"policy\",\n", " \"status\": \"published\",\n", " \"title\": \"Travel Policy\",\n", " \"sourceUrl\": \"https://contoso.example/policies/travel\",\n", " \"chunk\": \"Employees must submit receipts within 30 days.\",\n", " \"rawText\": \"Employees must submit receipts within 30 days. Full policy text...\",\n", " \"embedding\": [0.12, -0.44, 0.91, 0.03],\n", " \"ocrDiagnostics\": {\"engine\": \"sample\", \"confidence\": 0.98},\n", "}\n", "\n", "included_scalar_fields = {p[\"path\"].strip(\"/?\") for p in indexing_policy[\"includedPaths\"]}\n", "all_fields = set(sample_doc.keys())\n", "non_indexed_candidates = sorted(all_fields - included_scalar_fields - {\"id\"})\n", "\n", "print(\"Fields intended for scalar indexing:\", sorted(included_scalar_fields))\n", "print(\"All document fields:\", sorted(all_fields))\n", "print(\"Potential non-indexed or specially handled fields:\", non_indexed_candidates)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Required environment variables for live Azure Cosmos DB validation\n", "\n", "If you want to run the next live examples against Azure Cosmos DB, set these environment variables first:\n", "\n", "- `COSMOS_ENDPOINT`\n", "- `COSMOS_KEY`\n", "- `COSMOS_DATABASE`\n", "- `COSMOS_CONTAINER`\n", "\n", "Optional variables used in examples:\n", "\n", "- `COSMOS_PARTITION_KEY_PATH`\n", "- `COSMOS_OFFER_THROUGHPUT`" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Production-style retrieval request with metadata constraints\n", "\n", "This example shows the intended production query shape: vector similarity plus business boundaries such as tenant, document type, and publication status. The cell is written to be safe in a notebook and will print the query and parameters even if no live connection is configured." ] }, { "cell_type": "code", "metadata": {}, "source": [ "endpoint = os.getenv(\"COSMOS_ENDPOINT\", \"https://example.documents.azure.com:443/\")\n", "key = os.getenv(\"COSMOS_KEY\", \"fake-key\")\n", "database_name = os.getenv(\"COSMOS_DATABASE\", \"ai\")\n", "container_name = os.getenv(\"COSMOS_CONTAINER\", \"knowledge\")\n", "\n", "query_embedding = [0.12, -0.44, 0.91, 0.03]\n", "tenant_id = \"contoso\"\n", "doc_type = \"policy\"\n", "\n", "query = dedent(\"\"\"\n", "SELECT TOP 5 c.id, c.title, c.chunk, c.sourceUrl,\n", " VectorDistance(c.embedding, @embedding) AS score\n", "FROM c\n", "WHERE c.tenantId = @tenantId\n", " AND c.docType = @docType\n", " AND c.status = \"published\"\n", "ORDER BY VectorDistance(c.embedding, @embedding)\n", "\"\"\")\n", "\n", "params = [\n", " {\"name\": \"@embedding\", \"value\": query_embedding},\n", " {\"name\": \"@tenantId\", \"value\": tenant_id},\n", " {\"name\": \"@docType\", \"value\": doc_type},\n", "]\n", "\n", "print(\"Query to execute:\\n\")\n", "print(query)\n", "print(\"Parameters:\")\n", "pprint(params)\n", "\n", "if CosmosClient and endpoint != \"https://example.documents.azure.com:443/\" and key != \"fake-key\":\n", " try:\n", " client = CosmosClient(endpoint, credential=key)\n", " container = client.get_database_client(database_name).get_container_client(container_name)\n", " results = list(container.query_items(query=query, parameters=params, enable_cross_partition_query=True))\n", " print(\"\\nResults:\")\n", " for item in results:\n", " print(item.get(\"id\"), item.get(\"title\"), round(item.get(\"score\", 0), 4))\n", " except Exception as e:\n", " print(f\"Live query failed: {e}\")\n", "else:\n", " print(\"\\nLive execution skipped. Set COSMOS_ENDPOINT and COSMOS_KEY to run against Azure Cosmos DB.\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Create a container aligned to retrieval filters\n", "\n", "This example shows how partitioning and indexing policy work together. It is useful for validating that the hot retrieval path is reflected in container design rather than inherited from a prototype." ] }, { "cell_type": "code", "metadata": {}, "source": [ "endpoint = os.getenv(\"COSMOS_ENDPOINT\", \"https://example.documents.azure.com:443/\")\n", "key = os.getenv(\"COSMOS_KEY\", \"fake-key\")\n", "database_name = os.getenv(\"COSMOS_DATABASE\", \"ai\")\n", "container_name = os.getenv(\"COSMOS_CONTAINER\", \"knowledge\")\n", "partition_key_path = os.getenv(\"COSMOS_PARTITION_KEY_PATH\", \"/tenantId\")\n", "offer_throughput = int(os.getenv(\"COSMOS_OFFER_THROUGHPUT\", \"1000\"))\n", "\n", "container_definition = {\n", " \"id\": container_name,\n", " \"partition_key\": partition_key_path,\n", " \"indexing_policy\": {\n", " \"indexingMode\": \"consistent\",\n", " \"automatic\": True,\n", " \"includedPaths\": [\n", " {\"path\": \"/tenantId/?\"},\n", " {\"path\": \"/docType/?\"},\n", " {\"path\": \"/status/?\"},\n", " ],\n", " \"excludedPaths\": [\n", " {\"path\": \"/chunk/*\"},\n", " {\"path\": \"/rawText/*\"},\n", " ],\n", " },\n", " \"offer_throughput\": offer_throughput,\n", "}\n", "\n", "print(\"Planned container definition:\")\n", "pprint(container_definition)\n", "\n", "if CosmosClient and PartitionKey and endpoint != \"https://example.documents.azure.com:443/\" and key != \"fake-key\":\n", " try:\n", " client = CosmosClient(endpoint, credential=key)\n", " db = client.get_database_client(database_name)\n", " container = db.create_container_if_not_exists(\n", " id=container_name,\n", " partition_key=PartitionKey(path=partition_key_path),\n", " indexing_policy=container_definition[\"indexing_policy\"],\n", " offer_throughput=offer_throughput,\n", " )\n", " print(\"\\nContainer ready:\", container.read()[\"id\"])\n", " except Exception as e:\n", " print(f\"Container creation failed: {e}\")\n", "else:\n", " print(\"\\nLive creation skipped. Set Cosmos environment variables to run this step.\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sequence of a retrieval-backed AI request\n", "\n", "This cell translates the sequence diagram into a tabular view. It helps teams review where retrieval latency and filtering correctness sit in the end-to-end request path." ] }, { "cell_type": "code", "metadata": {}, "source": [ "sequence = [\n", " (\"User\", \"AI App\", \"Ask question\"),\n", " (\"AI App\", \"Embedding Model\", \"Create query embedding\"),\n", " (\"AI App\", \"Cosmos DB\", \"Vector search + tenant/status filters\"),\n", " (\"Cosmos DB\", \"AI App\", \"Top K grounded chunks\"),\n", " (\"AI App\", \"LLM\", \"Prompt with retrieved context\"),\n", " (\"LLM\", \"AI App\", \"Answer\"),\n", " (\"AI App\", \"User\", \"Response with citations\"),\n", "]\n", "\n", "sequence_df = pd.DataFrame(sequence, columns=[\"From\", \"To\", \"Action\"])\n", "sequence_df" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Compare broad and constrained query shapes\n", "\n", "This example makes access patterns explicit. A broad vector query and a constrained query are not just different SQL strings; they represent different product behavior, cost profiles, and relevance boundaries." ] }, { "cell_type": "code", "metadata": {}, "source": [ "embedding = [0.12, -0.44, 0.91, 0.03]\n", "\n", "broad = dedent(\"\"\"\n", "SELECT TOP 5 c.id FROM c\n", "ORDER BY VectorDistance(c.embedding, @embedding)\n", "\"\"\")\n", "\n", "constrained = dedent(\"\"\"\n", "SELECT TOP 5 c.id FROM c\n", "WHERE c.tenantId = @tenantId AND c.status = \"published\"\n", "ORDER BY VectorDistance(c.embedding, @embedding)\n", "\"\"\")\n", "\n", "params = [\n", " {\"name\": \"@embedding\", \"value\": embedding},\n", " {\"name\": \"@tenantId\", \"value\": \"contoso\"},\n", "]\n", "\n", "print(\"Broad query shape:\", broad.strip().splitlines()[0])\n", "print(\"Constrained query shape:\", constrained.strip().splitlines()[1].strip())\n", "print(\"Parameters:\")\n", "pprint(params)\n", "\n", "comparison = pd.DataFrame([\n", " {\"query_type\": \"broad\", \"has_tenant_filter\": False, \"has_status_filter\": False, \"likely_scope\": \"whole corpus\"},\n", " {\"query_type\": \"constrained\", \"has_tenant_filter\": True, \"has_status_filter\": True, \"likely_scope\": \"business-bounded subset\"},\n", "])\n", "comparison" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Simulate why constrained retrieval can reduce work\n", "\n", "Because a notebook may not have a live Cosmos DB account attached, this cell uses a small synthetic corpus to illustrate the idea behind constrained retrieval. The goal is not to reproduce Cosmos internals, but to validate that metadata boundaries reduce the candidate set before final ranking." ] }, { "cell_type": "code", "metadata": {}, "source": [ "docs = [\n", " {\"id\": \"1\", \"tenantId\": \"contoso\", \"status\": \"published\", \"docType\": \"policy\", \"score\": 0.11},\n", " {\"id\": \"2\", \"tenantId\": \"contoso\", \"status\": \"draft\", \"docType\": \"policy\", \"score\": 0.09},\n", " {\"id\": \"3\", \"tenantId\": \"fabrikam\", \"status\": \"published\", \"docType\": \"policy\", \"score\": 0.08},\n", " {\"id\": \"4\", \"tenantId\": \"contoso\", \"status\": \"published\", \"docType\": \"guide\", \"score\": 0.13},\n", " {\"id\": \"5\", \"tenantId\": \"contoso\", \"status\": \"published\", \"docType\": \"policy\", \"score\": 0.05},\n", " {\"id\": \"6\", \"tenantId\": \"fabrikam\", \"status\": \"archived\", \"docType\": \"policy\", \"score\": 0.07},\n", "]\n", "\n", "broad_candidates = sorted(docs, key=lambda x: x[\"score\"])[:5]\n", "constrained_candidates = sorted(\n", " [d for d in docs if d[\"tenantId\"] == \"contoso\" and d[\"status\"] == \"published\"],\n", " key=lambda x: x[\"score\"],\n", ")[:5]\n", "\n", "print(\"Broad candidate count considered:\", len(docs))\n", "print(\"Constrained candidate count considered:\", len([d for d in docs if d[\"tenantId\"] == \"contoso\" and d[\"status\"] == \"published\"]))\n", "print(\"\\nBroad top results:\")\n", "pprint(broad_candidates)\n", "print(\"\\nConstrained top results:\")\n", "pprint(constrained_candidates)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Inspect throughput posture from Python\n", "\n", "The original post used PowerShell to inspect throughput. This Python version is notebook-friendly and prints guidance even when no live Azure connection is configured." ] }, { "cell_type": "code", "metadata": {}, "source": [ "throughput_inspection = {\n", " \"resourceGroup\": \"rg-ai-prod\",\n", " \"accountName\": \"cosmos-ai-prod\",\n", " \"databaseName\": \"ai\",\n", " \"containerName\": \"knowledge\",\n", " \"what_to_check\": [\"manual throughput\", \"autoscale max throughput\", \"container workload shape\"],\n", "}\n", "\n", "print(\"Throughput inspection checklist:\")\n", "pprint(throughput_inspection)\n", "print(\"\\nFor live inspection, use Azure Portal, Azure CLI, or SDK tooling tied to your subscription.\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Minimal ingestion example for retrieval-focused documents\n", "\n", "This example writes only the fields needed for retrieval and filtering. It reinforces the architectural point that not every ingestion artifact belongs on the hot retrieval path." ] }, { "cell_type": "code", "metadata": {}, "source": [ "doc = {\n", " \"id\": \"doc-001#chunk-01\",\n", " \"tenantId\": \"contoso\",\n", " \"docType\": \"policy\",\n", " \"status\": \"published\",\n", " \"title\": \"Travel Policy\",\n", " \"sourceUrl\": \"https://contoso.example/policies/travel\",\n", " \"chunk\": \"Employees must submit receipts within 30 days.\",\n", " \"embedding\": [0.12, -0.44, 0.91, 0.03],\n", "}\n", "\n", "print(\"Document prepared for ingestion:\")\n", "pprint(doc)\n", "\n", "endpoint = os.getenv(\"COSMOS_ENDPOINT\", \"https://example.documents.azure.com:443/\")\n", "key = os.getenv(\"COSMOS_KEY\", \"fake-key\")\n", "database_name = os.getenv(\"COSMOS_DATABASE\", \"ai\")\n", "container_name = os.getenv(\"COSMOS_CONTAINER\", \"knowledge\")\n", "\n", "if CosmosClient and endpoint != \"https://example.documents.azure.com:443/\" and key != \"fake-key\":\n", " try:\n", " client = CosmosClient(endpoint, credential=key)\n", " container = client.get_database_client(database_name).get_container_client(container_name)\n", " result = container.upsert_item(doc)\n", " print(\"\\nUpserted document id:\", result[\"id\"])\n", " except Exception as e:\n", " print(f\"Live upsert failed: {e}\")\n", "else:\n", " print(\"\\nLive upsert skipped. Set Cosmos environment variables to run this step.\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Operational observability for retrieval-backed AI apps\n", "\n", "The blog argues that AI observability should include the data layer, not just model latency. This cell creates a simple metrics frame you can adapt to correlate request pressure, RU consumption, and server-side latency." ] }, { "cell_type": "code", "metadata": {}, "source": [ "metrics_df = pd.DataFrame([\n", " {\"Metric\": \"TotalRequests\", \"Average\": 1200, \"Maximum\": 1800},\n", " {\"Metric\": \"NormalizedRUConsumption\", \"Average\": 62.5, \"Maximum\": 88.0},\n", " {\"Metric\": \"ServerSideLatency\", \"Average\": 14.2, \"Maximum\": 37.9},\n", "])\n", "\n", "metrics_df" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Correlate app symptoms with database signals\n", "\n", "This final validation cell demonstrates the mindset behind observability: connect user-visible degradation to retrieval-path changes. In practice, you would join these signals with endpoint, scenario, tenant, or deployment metadata." ] }, { "cell_type": "code", "metadata": {}, "source": [ "observations = pd.DataFrame([\n", " {\"scenario\": \"knowledge_search\", \"app_latency_ms\": 820, \"normalized_ru\": 88.0, \"server_side_latency_ms\": 37.9, \"note\": \"possible retrieval pressure spike\"},\n", " {\"scenario\": \"knowledge_search\", \"app_latency_ms\": 410, \"normalized_ru\": 62.5, \"server_side_latency_ms\": 14.2, \"note\": \"healthier baseline\"},\n", "])\n", "\n", "observations[\"suspect_retrieval_bottleneck\"] = (\n", " (observations[\"normalized_ru\"] > 80) |\n", " (observations[\"server_side_latency_ms\"] > 30)\n", ")\n", "\n", "observations" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Next Steps\n", "\n", "Indexing policy in Azure Cosmos DB is not a background setting for AI apps; it is part of the product contract for retrieval speed, cost, and correctness. If your app depends on vector search plus metadata boundaries, validate that your partitioning, indexing paths, and query shapes reflect the real workload rather than a prototype default.\n", "\n", "Next steps:\n", "\n", "1. Replace the sample endpoint and key with your Cosmos DB environment variables.\n", "2. Run the live query and ingestion cells against a non-production container.\n", "3. Compare broad and constrained query patterns using your real tenant, status, and document-type filters.\n", "4. Review your current indexing policy for fields that are never used in the latency-critical retrieval path.\n", "5. Add observability that correlates app latency with RU consumption and server-side database latency." ] } ] }