{
  "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": "Designing Better Data Models with an AI Coding Agent: Where Cosmos DB Teams Should Draw the Line",
      "slug": "designing-better-data-models-with-an-ai-coding-agent-where-c",
      "generated_by": "LinkedIn Post Generator + Azure OpenAI",
      "generated_at": "2026-07-20T14:36:01.521Z"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Designing Better Data Models with an AI Coding Agent: Where Cosmos DB Teams Should Draw the Line\n",
        "\n",
        "AI coding agents can draft Cosmos DB models quickly, but speed at generating schemas is not the same as designing for production traffic. This notebook turns the blog post into hands-on validation steps focused on workload patterns, partition-key tradeoffs, indexing visibility, and human review loops. The goal is to test model ideas as workload hypotheses rather than trusting elegant generated artifacts."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "%pip install pandas matplotlib"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import random\n",
        "import statistics\n",
        "import csv\n",
        "import json\n",
        "from collections import Counter, defaultdict\n",
        "\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Review loop for agent-assisted Cosmos DB modeling\n",
        "\n",
        "The blog argues that AI agents should propose models and configuration, but humans must approve partition keys, indexing, TTL, and promotion decisions after workload testing. This cell renders a simple text version of that review loop so the decision boundary is explicit."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "review_flow = {\n",
        "    \"AI agent proposes model and config\": \"Human reviews partition key, indexing, TTL\",\n",
        "    \"Human reviews partition key, indexing, TTL\": \"Deploy draft to test environment\",\n",
        "    \"Deploy draft to test environment\": \"Run workload harness with skewed tenants\",\n",
        "    \"Run workload harness with skewed tenants\": \"Collect RU, latency, hot partition signals\",\n",
        "    \"Collect RU, latency, hot partition signals\": \"Acceptable tradeoffs?\",\n",
        "    \"Acceptable tradeoffs?\": {\n",
        "        \"Yes\": \"Promote with documented rationale\",\n",
        "        \"No\": \"Revise model and rerun\"\n",
        "    }\n",
        "}\n",
        "\n",
        "print(json.dumps(review_flow, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Generate skewed tenant operations\n",
        "\n",
        "A core point in the post is that partitioning mistakes only become obvious under realistic skew. This example creates a synthetic workload where a subset of tenants receives a disproportionate share of writes, making it easier to spot hot-partition risk before production."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def build_ops(tenants=20, writes=200, reads=120, hot_ratio=0.7, seed=42):\n",
        "    random.seed(seed)\n",
        "    ids = [f\"tenant-{i:02d}\" for i in range(tenants)]\n",
        "    hot = ids[: max(1, tenants // 5)]\n",
        "    ops = []\n",
        "    for i in range(writes):\n",
        "        tenant = random.choice(hot if random.random() < hot_ratio else ids)\n",
        "        ops.append({\"kind\": \"write\", \"tenantId\": tenant, \"id\": f\"doc-{i:04d}\"})\n",
        "    for _ in range(reads):\n",
        "        tenant = random.choice(ids)\n",
        "        ops.append({\"kind\": \"read\", \"tenantId\": tenant, \"id\": f\"doc-{random.randint(0, writes-1):04d}\"})\n",
        "    random.shuffle(ops)\n",
        "    return ops\n",
        "\n",
        "ops = build_ops()\n",
        "print(\"sample:\", ops[:5])\n",
        "print(\"tenant skew:\", Counter(op[\"tenantId\"] for op in ops).most_common(5))\n",
        "\n",
        "ops_df = pd.DataFrame(ops)\n",
        "ops_df.head()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Compare candidate partition-key strategies\n",
        "\n",
        "The post draws a hard line around partition-key selection: agents can suggest options, but humans must approve them after reviewing workload economics. This example scores two candidate strategies under the same synthetic pattern and highlights the tradeoff between total RU and hot-partition concentration."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def score(strategy, seed=42):\n",
        "    random.seed(seed)\n",
        "    total_ru = 0.0\n",
        "    hot_hits = 0\n",
        "    for _ in range(200):\n",
        "        tenant = random.choice([\"tenant-00\"] * 7 + [f\"tenant-{i:02d}\" for i in range(1, 10)])\n",
        "        is_hot = tenant == \"tenant-00\"\n",
        "        if strategy == \"tenantId\":\n",
        "            total_ru += 7.5 if is_hot else 5.5\n",
        "            hot_hits += 1 if is_hot else 0\n",
        "        else:\n",
        "            total_ru += 6.2\n",
        "            hot_hits += 0\n",
        "    return {\"strategy\": strategy, \"total_ru\": round(total_ru, 1), \"hot_partition_events\": hot_hits}\n",
        "\n",
        "results = [score(name) for name in (\"tenantId\", \"tenantId+bucket\")]\n",
        "for row in results:\n",
        "    print(row)\n",
        "\n",
        "pd.DataFrame(results)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Simulate RU and latency observations\n",
        "\n",
        "Generated schemas do not prove workload coverage, RU efficiency, or latency behavior under skew. This example simulates observations for reads and writes so reviewers can compare candidate models using metrics that matter operationally."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def observe(ops, seed=42):\n",
        "    random.seed(seed)\n",
        "    rows = []\n",
        "    for op in ops:\n",
        "        hot_penalty = 1.8 if op[\"tenantId\"] in {\"tenant-00\", \"tenant-01\", \"tenant-02\", \"tenant-03\"} else 1.0\n",
        "        base_ru = 6.0 if op[\"kind\"] == \"write\" else 2.5\n",
        "        base_ms = 18 if op[\"kind\"] == \"write\" else 9\n",
        "        rows.append({\n",
        "            \"kind\": op[\"kind\"],\n",
        "            \"tenantId\": op[\"tenantId\"],\n",
        "            \"ru\": round(base_ru * hot_penalty * random.uniform(0.9, 1.2), 2),\n",
        "            \"latency_ms\": round(base_ms * hot_penalty * random.uniform(0.8, 1.4), 1),\n",
        "        })\n",
        "    return rows\n",
        "\n",
        "sample_ops = [{\"kind\": \"write\", \"tenantId\": \"tenant-00\"}, {\"kind\": \"read\", \"tenantId\": \"tenant-09\"}] * 20\n",
        "rows = observe(sample_ops)\n",
        "print(\"avg_ru=\", round(statistics.mean(r[\"ru\"] for r in rows), 2))\n",
        "print(\"p95_ms=\", sorted(r[\"latency_ms\"] for r in rows)[int(len(rows) * 0.95) - 1])\n",
        "\n",
        "obs_df = pd.DataFrame(rows)\n",
        "obs_df.head()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summarize hot-tenant signals\n",
        "\n",
        "Average metrics can hide bad tails. This example aggregates RU, operation count, and max latency by tenant so a human reviewer can quickly see whether a small number of tenants dominate cost or experience worse latency."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def summarize(rows):\n",
        "    by_tenant = defaultdict(lambda: {\"ru\": 0.0, \"count\": 0, \"max_ms\": 0.0})\n",
        "    for r in rows:\n",
        "        t = by_tenant[r[\"tenantId\"]]\n",
        "        t[\"ru\"] += r[\"ru\"]\n",
        "        t[\"count\"] += 1\n",
        "        t[\"max_ms\"] = max(t[\"max_ms\"], r[\"latency_ms\"])\n",
        "    ranked = sorted(by_tenant.items(), key=lambda kv: kv[1][\"ru\"], reverse=True)\n",
        "    for tenant, stats in ranked[:5]:\n",
        "        avg_ru = round(stats[\"ru\"] / stats[\"count\"], 2)\n",
        "        print(f\"{tenant}: total_ru={stats['ru']:.1f}, avg_ru={avg_ru}, max_ms={stats['max_ms']}\")\n",
        "    return ranked\n",
        "\n",
        "sample = [\n",
        "    {\"tenantId\": \"tenant-00\", \"ru\": 12.1, \"latency_ms\": 31.0},\n",
        "    {\"tenantId\": \"tenant-00\", \"ru\": 10.4, \"latency_ms\": 28.2},\n",
        "    {\"tenantId\": \"tenant-09\", \"ru\": 2.7, \"latency_ms\": 8.9},\n",
        "]\n",
        "ranked = summarize(sample)\n",
        "ranked[:3]"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Sequence of human review around the agent loop\n",
        "\n",
        "The blog emphasizes that the agent should sit inside a control loop rather than replacing architectural judgment. This cell captures the sequence as structured data: the developer requests a draft, the agent proposes assumptions, the workload harness tests them, and only then does refinement continue."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "sequence = [\n",
        "    (\"Developer\", \"AI Coding Agent\", \"Propose partition key and indexing draft\"),\n",
        "    (\"AI Coding Agent\", \"Developer\", \"Reviewable config with explicit assumptions\"),\n",
        "    (\"Developer\", \"Workload Harness\", \"Approve and run skewed workload\"),\n",
        "    (\"Workload Harness\", \"Cosmos DB Test Container\", \"Writes and representative reads\"),\n",
        "    (\"Cosmos DB Test Container\", \"Workload Harness\", \"RU charges and latency\"),\n",
        "    (\"Workload Harness\", \"Developer\", \"Hot partition and cost summary\"),\n",
        "    (\"Developer\", \"AI Coding Agent\", \"Refine model only after human review\"),\n",
        "]\n",
        "\n",
        "for src, dst, msg in sequence:\n",
        "    print(f\"{src} -> {dst}: {msg}\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Draft explicit Cosmos DB container configuration in Python\n",
        "\n",
        "The original post used PowerShell to make hidden defaults visible. Here the same idea is expressed in Python: force explicit choices for partition key, TTL, and indexing paths so reviewers can challenge them before deployment."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def draft_container_config(\n",
        "    database_name=\"appdb\",\n",
        "    container_name=\"orders\",\n",
        "    partition_key_path=\"/tenantId\",\n",
        "    default_ttl=-1,\n",
        "    included_paths=None,\n",
        "    excluded_paths=None,\n",
        "):\n",
        "    included_paths = included_paths or [\"/*\"]\n",
        "    excluded_paths = excluded_paths or [\"/largeBlob/*\"]\n",
        "    config = {\n",
        "        \"databaseName\": database_name,\n",
        "        \"containerName\": container_name,\n",
        "        \"partitionKey\": {\"paths\": [partition_key_path], \"kind\": \"Hash\"},\n",
        "        \"defaultTtl\": default_ttl,\n",
        "        \"indexingPolicy\": {\n",
        "            \"indexingMode\": \"consistent\",\n",
        "            \"includedPaths\": [{\"path\": p} for p in included_paths],\n",
        "            \"excludedPaths\": [{\"path\": p} for p in excluded_paths],\n",
        "        },\n",
        "    }\n",
        "    return config\n",
        "\n",
        "config = draft_container_config()\n",
        "print(json.dumps(config, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Validate that risky defaults are not hidden\n",
        "\n",
        "A generated draft should not quietly omit important decisions. This validation step rejects missing or weak partition-key choices and ensures indexing paths are explicit, creating a simple gate before human review."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def validate_agent_draft(partition_key_path=\"/tenantId\", included_paths=None, composite_indexes=None):\n",
        "    included_paths = included_paths if included_paths is not None else [\"/*\"]\n",
        "    composite_indexes = composite_indexes if composite_indexes is not None else []\n",
        "\n",
        "    if (not partition_key_path) or partition_key_path.strip() == \"\" or partition_key_path == \"/id\":\n",
        "        raise ValueError(\"Partition key must be explicit and should not default to /id without review.\")\n",
        "    if len(included_paths) == 0:\n",
        "        raise ValueError(\"Indexing paths must be explicit; empty included paths require human sign-off.\")\n",
        "\n",
        "    return {\n",
        "        \"PartitionKeyReviewed\": True,\n",
        "        \"IncludedPathCount\": len(included_paths),\n",
        "        \"CompositeIndexCount\": len(composite_indexes),\n",
        "        \"Status\": \"Ready for human review\",\n",
        "    }\n",
        "\n",
        "validation = validate_agent_draft()\n",
        "print(json.dumps(validation, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Record observations to CSV for external review\n",
        "\n",
        "The post recommends producing artifacts that humans can inspect outside the agent loop. This example writes workload observations to CSV so architects, product teams, or operations reviewers can analyze cost and latency in familiar tools."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "rows = [\n",
        "    {\"kind\": \"write\", \"tenantId\": \"tenant-00\", \"ru\": 11.8, \"latency_ms\": 29.4},\n",
        "    {\"kind\": \"read\", \"tenantId\": \"tenant-09\", \"ru\": 2.6, \"latency_ms\": 8.7},\n",
        "]\n",
        "\n",
        "with open(\"cosmos_workload_observations.csv\", \"w\", newline=\"\", encoding=\"utf-8\") as f:\n",
        "    writer = csv.DictWriter(f, fieldnames=[\"kind\", \"tenantId\", \"ru\", \"latency_ms\"])\n",
        "    writer.writeheader()\n",
        "    writer.writerows(rows)\n",
        "\n",
        "print(\"wrote cosmos_workload_observations.csv\")\n",
        "pd.read_csv(\"cosmos_workload_observations.csv\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## End-to-end workload hypothesis check\n",
        "\n",
        "This final hands-on step combines workload generation, observation, and tenant-level summarization. It demonstrates the notebook's main lesson: treat every proposed Cosmos DB model as a hypothesis that must survive skewed workload testing before promotion."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "ops = build_ops(tenants=20, writes=300, reads=180, hot_ratio=0.75, seed=7)\n",
        "observations = observe(ops, seed=7)\n",
        "\n",
        "obs_df = pd.DataFrame(observations)\n",
        "summary = (\n",
        "    obs_df.groupby(\"tenantId\")\n",
        "    .agg(total_ru=(\"ru\", \"sum\"), avg_ru=(\"ru\", \"mean\"), max_ms=(\"latency_ms\", \"max\"), ops=(\"tenantId\", \"count\"))\n",
        "    .sort_values(\"total_ru\", ascending=False)\n",
        ")\n",
        "\n",
        "print(\"Top tenants by RU:\")\n",
        "print(summary.head(10).round(2))\n",
        "\n",
        "ax = summary.head(10)[\"total_ru\"].plot(kind=\"bar\", figsize=(10, 4), title=\"Top 10 Tenants by Total RU\")\n",
        "ax.set_ylabel(\"Total RU\")\n",
        "ax.set_xlabel(\"Tenant\")\n",
        "plt.tight_layout()\n",
        "plt.show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Next Steps\n",
        "\n",
        "This notebook validated the blog's central claim: AI agents are useful for drafting Cosmos DB models, but humans must own workload economics and operational risk. The most important review boundaries are partition-key choice, indexing policy, consistency expectations, and evidence from skewed workload testing.\n",
        "\n",
        "Next steps:\n",
        "- Build an access-pattern catalog before accepting any generated schema.\n",
        "- Test 2-3 candidate document shapes under skewed tenants and bursty writes.\n",
        "- Require explicit assumptions, unsupported queries, and tradeoffs in every agent-generated draft.\n",
        "- Add pre-deployment checkpoints for container creation, index changes, regional rollout, and representative load tests.\n",
        "- Extend these simulations with your real tenant mix, latency targets, and retention rules."
      ]
    }
  ]
}