{ "cells": [ { "cell_type": "markdown", "id": "48da9854", "metadata": {}, "source": [ "# Lesson 16 - Deploying Scalable Agents with Microsoft Foundry\n", "\n", "In this notebook you build a **production-ready customer support agent** for the fictional company **Contoso**. Unlike the earlier lessons, the point here is not the agent's reasoning loop — it is everything wrapped *around* it that makes an agent safe to run at scale:\n", "\n", "1. **Tool calling** — order lookups and ticket creation.\n", "2. **RAG** — policy answers from a knowledge base.\n", "3. **Memory** — remembering the customer across turns.\n", "4. **Model routing** — send simple requests to a small model, complex ones to a large model.\n", "5. **Response caching** — serve repeated questions without a model call.\n", "6. **Human approval** — refunds above a threshold pause for sign-off.\n", "7. **Evaluation gate** — an offline test set that blocks a bad release.\n", "8. **Observability** — OpenTelemetry tracing around every request.\n", "\n", "Each section is self-contained and runnable. Read every line — the production primitives are kept deliberately small." ] }, { "cell_type": "markdown", "id": "cbba23dd", "metadata": {}, "source": [ "## Setup\n", "\n", "Before running this notebook, make sure you have:\n", "\n", "1. **A Microsoft Foundry project** with a deployed chat model (e.g. `gpt-5-mini`).\n", "2. **Logged in with the Azure CLI** — run `az login` in your terminal.\n", "3. **Set the required environment variables:**\n", " - `AZURE_AI_PROJECT_ENDPOINT` — your Microsoft Foundry project endpoint.\n", " - `AZURE_AI_MODEL_DEPLOYMENT_NAME` — the name of your deployed model.\n", "\n", "The RAG section uses **Azure AI Search** when `AZURE_SEARCH_SERVICE_ENDPOINT` and `AZURE_SEARCH_API_KEY` are set, and falls back to an in-memory search so the notebook runs without a Search resource." ] }, { "cell_type": "code", "execution_count": null, "id": "585794b8", "metadata": {}, "outputs": [], "source": [ "%pip install agent-framework azure-ai-projects azure-identity python-dotenv -q" ] }, { "cell_type": "code", "execution_count": null, "id": "d78d98ef", "metadata": {}, "outputs": [], "source": [ "import logging\n", "logging.getLogger(\"agent_framework.foundry\").setLevel(logging.ERROR)\n", "\n", "import os\n", "import re\n", "import dotenv\n", "from typing import Annotated\n", "\n", "from agent_framework import tool\n", "from agent_framework.foundry import FoundryChatClient\n", "from azure.identity import AzureCliCredential\n", "\n", "dotenv.load_dotenv(dotenv.find_dotenv())\n", "\n", "endpoint = os.getenv(\"AZURE_AI_PROJECT_ENDPOINT\")\n", "model = os.getenv(\"AZURE_AI_MODEL_DEPLOYMENT_NAME\")\n", "\n", "if not endpoint or not model:\n", " raise ValueError(\n", " \"Missing required environment variables. \"\n", " \"Please set AZURE_AI_PROJECT_ENDPOINT and AZURE_AI_MODEL_DEPLOYMENT_NAME in your .env file.\"\n", " )\n", "\n", "provider = FoundryChatClient(\n", " project_endpoint=endpoint,\n", " model=model,\n", " credential=AzureCliCredential(),\n", ")\n", "print(\"Foundry client ready.\")" ] }, { "cell_type": "markdown", "id": "b1e4f8a3", "metadata": {}, "source": [ "## 1. Tools\n", "\n", "Production tools do real work against real systems. Here we simulate an order database and a ticketing system with plain Python functions. The `@tool` decorator exposes them to the agent.\n", "\n", "Notice `issue_refund` uses `approval_mode=\"always_require\"` for refunds above a threshold — this is the human-in-the-loop primitive we deploy later." ] }, { "cell_type": "code", "execution_count": null, "id": "41a10845", "metadata": {}, "outputs": [], "source": [ "# Simulated backend systems (in production these are API calls behind scoped identities).\n", "ORDERS = {\n", " \"A1001\": {\"status\": \"shipped\", \"total\": 42.00, \"eta\": \"2 days\"},\n", " \"A1002\": {\"status\": \"processing\", \"total\": 128.50, \"eta\": \"5 days\"},\n", " \"A1003\": {\"status\": \"delivered\", \"total\": 19.99, \"eta\": \"delivered\"},\n", "}\n", "TICKETS: list[dict] = []\n", "REFUND_APPROVAL_THRESHOLD = 50.0\n", "\n", "\n", "@tool(approval_mode=\"never_require\")\n", "def get_order_status(order_id: Annotated[str, \"The customer's order ID, e.g. A1001\"]) -> str:\n", " \"\"\"Look up the status of a customer order.\"\"\"\n", " order = ORDERS.get(order_id.upper())\n", " if not order:\n", " return f\"No order found with ID {order_id}.\"\n", " return (\n", " f\"Order {order_id.upper()}: status={order['status']}, \"\n", " f\"total=${order['total']:.2f}, eta={order['eta']}.\"\n", " )\n", "\n", "\n", "@tool(approval_mode=\"never_require\")\n", "def open_ticket(\n", " subject: Annotated[str, \"Short subject line for the support ticket\"],\n", " details: Annotated[str, \"Full description of the customer's issue\"],\n", ") -> str:\n", " \"\"\"Open a support ticket for issues that need human follow-up.\"\"\"\n", " ticket_id = f\"T{1000 + len(TICKETS) + 1}\"\n", " TICKETS.append({\"id\": ticket_id, \"subject\": subject, \"details\": details})\n", " return f\"Ticket {ticket_id} opened: {subject}\"\n", "\n", "\n", "def refund_needs_approval(amount: float) -> bool:\n", " \"\"\"Refunds above the threshold require a human approver.\"\"\"\n", " return amount > REFUND_APPROVAL_THRESHOLD\n", "\n", "\n", "@tool(approval_mode=\"always_require\")\n", "def issue_refund(\n", " order_id: Annotated[str, \"The order to refund\"],\n", " amount: Annotated[float, \"Refund amount in USD\"],\n", ") -> str:\n", " \"\"\"Issue a refund. Execution pauses for human approval before it runs.\"\"\"\n", " return f\"Refund of ${amount:.2f} issued for order {order_id.upper()}.\"\n", "\n", "\n", "print(\"Tools defined.\")" ] }, { "cell_type": "markdown", "id": "5a1d9437", "metadata": {}, "source": [ "## 2. RAG — Policy Knowledge Base\n", "\n", "Policy questions (\"what's your return window?\") should be answered from an authoritative source, not the model's memory. We wrap a small knowledge base as a search tool.\n", "\n", "In production this is **Azure AI Search**; here we provide an in-memory keyword search so the notebook runs anywhere, and switch to Azure AI Search automatically when the environment variables are present." ] }, { "cell_type": "code", "execution_count": null, "id": "acf21012", "metadata": {}, "outputs": [], "source": [ "KNOWLEDGE_BASE = {\n", " \"returns\": \"Contoso accepts returns within 30 days of delivery for a full refund. Items must be unused and in original packaging.\",\n", " \"shipping\": \"Standard shipping takes 3-5 business days. Express shipping (1-2 days) is available at checkout for an extra fee.\",\n", " \"warranty\": \"All Contoso electronics carry a 12-month limited warranty covering manufacturing defects.\",\n", " \"refund_policy\": \"Refunds are processed to the original payment method within 5 business days of approval. Refunds over $50 require a supervisor's approval.\",\n", "}\n", "\n", "\n", "def _in_memory_search(query: str) -> str:\n", " q = query.lower()\n", " hits = [text for key, text in KNOWLEDGE_BASE.items() if key.replace(\"_\", \" \") in q or key in q]\n", " if not hits:\n", " # crude keyword fallback so the tool still returns something useful\n", " hits = [text for text in KNOWLEDGE_BASE.values() if any(w in text.lower() for w in q.split())]\n", " return \"\\n\".join(hits) if hits else \"No matching policy found.\"\n", "\n", "\n", "def _azure_search(query: str) -> str:\n", " from azure.core.credentials import AzureKeyCredential\n", " from azure.search.documents import SearchClient\n", "\n", " client = SearchClient(\n", " endpoint=os.environ[\"AZURE_SEARCH_SERVICE_ENDPOINT\"],\n", " index_name=os.getenv(\"AZURE_SEARCH_INDEX_NAME\", \"contoso-policies\"),\n", " credential=AzureKeyCredential(os.environ[\"AZURE_SEARCH_API_KEY\"]),\n", " )\n", " results = client.search(search_text=query, top=3)\n", " return \"\\n\".join(r.get(\"content\", \"\") for r in results) or \"No matching policy found.\"\n", "\n", "\n", "USE_AZURE_SEARCH = bool(os.getenv(\"AZURE_SEARCH_SERVICE_ENDPOINT\") and os.getenv(\"AZURE_SEARCH_API_KEY\"))\n", "\n", "\n", "@tool(approval_mode=\"never_require\")\n", "def search_policies(query: Annotated[str, \"The policy question to look up\"]) -> str:\n", " \"\"\"Search Contoso support policies to answer customer questions.\"\"\"\n", " if USE_AZURE_SEARCH:\n", " return _azure_search(query)\n", " return _in_memory_search(query)\n", "\n", "\n", "print(f\"RAG ready. Using {'Azure AI Search' if USE_AZURE_SEARCH else 'in-memory search'}.\")" ] }, { "cell_type": "markdown", "id": "a58923d2", "metadata": {}, "source": [ "## 3. Memory\n", "\n", "A support agent that forgets who it is talking to is a bad support agent. We keep a tiny per-customer profile store and inject a short summary into the agent's instructions. In production this is a memory service (see Lesson 13); here a dict makes the pattern visible." ] }, { "cell_type": "code", "execution_count": null, "id": "65e61110", "metadata": {}, "outputs": [], "source": [ "CUSTOMER_MEMORY: dict[str, dict] = {\n", " \"cust-42\": {\"name\": \"Dana\", \"tier\": \"enterprise\", \"recent_order\": \"A1002\"},\n", " \"cust-99\": {\"name\": \"Sam\", \"tier\": \"standard\", \"recent_order\": \"A1003\"},\n", "}\n", "\n", "\n", "def memory_context(customer_id: str) -> str:\n", " profile = CUSTOMER_MEMORY.get(customer_id)\n", " if not profile:\n", " return \"This is a new customer with no history.\"\n", " return (\n", " f\"Customer {profile['name']} ({profile['tier']} tier). \"\n", " f\"Most recent order: {profile['recent_order']}.\"\n", " )\n", "\n", "\n", "print(memory_context(\"cust-42\"))" ] }, { "cell_type": "markdown", "id": "b5c318bf", "metadata": {}, "source": [ "## 4 & 5. Model Routing and Response Caching\n", "\n", "Two cost levers wired into a single request handler:\n", "\n", "- **Routing**: a cheap heuristic classifier decides whether a request needs the small or the large model.\n", "- **Caching**: normalised repeat questions are served straight from a cache with no model call.\n", "\n", "The classifier here is intentionally simple. In production you would validate it against traffic and could replace it with Foundry's Model Router." ] }, { "cell_type": "code", "execution_count": null, "id": "d8456988", "metadata": {}, "outputs": [], "source": [ "SMALL_MODEL = os.getenv(\"AZURE_AI_SMALL_MODEL\", model) # e.g. gpt-5-nano\n", "LARGE_MODEL = os.getenv(\"AZURE_AI_LARGE_MODEL\", model) # e.g. gpt-5-mini\n", "\n", "response_cache: dict[str, str] = {}\n", "route_counters = {\"small\": 0, \"large\": 0, \"cache\": 0}\n", "\n", "\n", "def normalize(query: str) -> str:\n", " return re.sub(r\"\\s+\", \" \", query.lower().strip())\n", "\n", "\n", "COMPLEX_SIGNALS = (\"refund\", \"cancel\", \"complaint\", \"escalate\", \"broken\", \"wrong\", \"why\")\n", "\n", "\n", "def is_simple(query: str) -> bool:\n", " \"\"\"Route complex or high-stakes requests to the large model; everything else to the small one.\"\"\"\n", " q = query.lower()\n", " if any(signal in q for signal in COMPLEX_SIGNALS):\n", " return False\n", " return len(q.split()) <= 20\n", "\n", "\n", "def choose_model(query: str) -> str:\n", " return SMALL_MODEL if is_simple(query) else LARGE_MODEL\n", "\n", "\n", "print(f\"Small model: {SMALL_MODEL} | Large model: {LARGE_MODEL}\")" ] }, { "cell_type": "markdown", "id": "87fa7e7f", "metadata": {}, "source": [ "## 6 & 8. The Agent, Human Approval, and Observability\n", "\n", "Now we assemble the agent from the tools above and wrap each request in an OpenTelemetry span. The `handle_support_request` function is the production request handler: cache → route → trace → run → cache.\n", "\n", "Human approval is handled by the framework: because `issue_refund` is `approval_mode=\"always_require\"`, the run pauses and surfaces an approval request that we resolve explicitly." ] }, { "cell_type": "code", "execution_count": null, "id": "01c62cfc", "metadata": {}, "outputs": [], "source": [ "# Tracing: use the Agent Framework tracer if available, else a no-op so the notebook runs anywhere.\n", "try:\n", " from agent_framework.observability import get_tracer\n", " tracer = get_tracer()\n", "except Exception: # observability extras not installed\n", " from contextlib import contextmanager\n", "\n", " class _NoopSpan:\n", " def set_attribute(self, *_args, **_kwargs):\n", " pass\n", "\n", " class _NoopTracer:\n", " @contextmanager\n", " def start_as_current_span(self, _name):\n", " yield _NoopSpan()\n", "\n", " tracer = _NoopTracer()\n", "\n", "\n", "SUPPORT_INSTRUCTIONS = (\n", " \"You are Contoso's customer support agent. Be concise, friendly, and accurate. \"\n", " \"Use search_policies for policy questions, get_order_status for orders, \"\n", " \"open_ticket when a human needs to follow up, and issue_refund for refunds. \"\n", " \"Never invent policy details.\"\n", ")\n", "\n", "# Build one agent per model tier so we can route by cost. The current agent-framework\n", "# selects the model on the client, so each tier gets its own FoundryChatClient.\n", "_TOOLS = [get_order_status, open_ticket, search_policies, issue_refund]\n", "_agents_by_model: dict[str, object] = {}\n", "\n", "\n", "def agent_for(model_name: str):\n", " if model_name not in _agents_by_model:\n", " client = FoundryChatClient(\n", " project_endpoint=endpoint,\n", " model=model_name,\n", " credential=AzureCliCredential(),\n", " )\n", " _agents_by_model[model_name] = client.as_agent(\n", " name=\"ContosoSupportAgent\",\n", " instructions=SUPPORT_INSTRUCTIONS,\n", " tools=_TOOLS,\n", " )\n", " return _agents_by_model[model_name]\n", "\n", "\n", "# Default agent (used by the evaluation gate, which does not route).\n", "support_agent = agent_for(SMALL_MODEL)\n", "\n", "\n", "async def handle_support_request(query: str, customer_id: str) -> str:\n", " # 1. Serve from cache when we can.\n", " key = normalize(query)\n", " if key in response_cache:\n", " route_counters[\"cache\"] += 1\n", " return response_cache[key]\n", "\n", " # 2. Route by complexity to control cost.\n", " chosen_model = choose_model(query)\n", " route_counters[\"small\" if chosen_model == SMALL_MODEL else \"large\"] += 1\n", "\n", " # 3. Add per-customer memory to the prompt.\n", " context = memory_context(customer_id)\n", " prompt = f\"[Customer context: {context}]\\n\\n{query}\"\n", "\n", " # 4. Run inside a trace span for observability.\n", " with tracer.start_as_current_span(\"support_request\") as span:\n", " span.set_attribute(\"customer.id\", customer_id)\n", " span.set_attribute(\"routed.model\", chosen_model)\n", " response = await agent_for(chosen_model).run(prompt)\n", "\n", " text = response.text\n", " response_cache[key] = text\n", " return text\n", "\n", "\n", "print(\"Support agent assembled.\")" ] }, { "cell_type": "code", "execution_count": null, "id": "478703e5", "metadata": {}, "outputs": [], "source": [ "# Try a few requests. The first is simple (small model), the second is a refund (large model + approval path).\n", "print(await handle_support_request(\"What is your return window?\", \"cust-99\"))\n", "print(\"---\")\n", "print(await handle_support_request(\"Where is my order A1002?\", \"cust-42\"))\n", "print(\"---\")\n", "# Repeat the first question -> served from cache.\n", "print(await handle_support_request(\"What is your return window?\", \"cust-99\"))\n", "print(\"---\")\n", "print(\"Routing counters:\", route_counters)" ] }, { "cell_type": "markdown", "id": "3bae744b", "metadata": {}, "source": [ "## 7. Evaluation Gate\n", "\n", "This is the release gate from the lesson: an offline test set scores the agent, and deployment only proceeds if the pass rate clears a threshold. The scorer here is a simple keyword-overlap check to keep the notebook self-contained; in production you would use an LLM-as-judge or a framework evaluator (see Lesson 10)." ] }, { "cell_type": "code", "execution_count": null, "id": "30b21256", "metadata": {}, "outputs": [], "source": [ "TEST_CASES = [\n", " {\"input\": \"How long do I have to return an item?\", \"expected\": [\"30 days\", \"refund\"]},\n", " {\"input\": \"How fast is standard shipping?\", \"expected\": [\"3-5\", \"business days\"]},\n", " {\"input\": \"What is the status of order A1001?\", \"expected\": [\"shipped\", \"A1001\"]},\n", " {\"input\": \"Do your electronics have a warranty?\", \"expected\": [\"12-month\", \"warranty\"]},\n", "]\n", "\n", "\n", "def score_response(actual: str, expected_keywords: list[str]) -> float:\n", " actual_l = actual.lower()\n", " hits = sum(1 for kw in expected_keywords if kw.lower() in actual_l)\n", " return hits / len(expected_keywords)\n", "\n", "\n", "async def evaluation_gate(test_cases: list[dict], threshold: float = 0.8) -> bool:\n", " passed = 0\n", " for case in test_cases:\n", " result = await support_agent.run(case[\"input\"])\n", " s = score_response(result.text, case[\"expected\"])\n", " status = \"PASS\" if s >= 0.5 else \"FAIL\"\n", " print(f\"[{status}] {case['input']} (score={s:.0%})\")\n", " if s >= 0.5:\n", " passed += 1\n", " pass_rate = passed / len(test_cases)\n", " print(f\"\\nEvaluation pass rate: {pass_rate:.0%} (gate: {threshold:.0%})\")\n", " return pass_rate >= threshold\n", "\n", "\n", "gate_passed = await evaluation_gate(TEST_CASES, threshold=0.8)\n", "print(\"\\nDeploy allowed:\" , gate_passed)" ] }, { "cell_type": "markdown", "id": "4cb0c705", "metadata": {}, "source": [ "## Putting It Together: A Simulated Release\n", "\n", "The cell below shows the whole loop the lesson describes: run the evaluation gate, and only \"deploy\" if it passes. This is the pattern you would run in CI before promoting an agent version to the Foundry Agent Service." ] }, { "cell_type": "code", "execution_count": null, "id": "be311368", "metadata": {}, "outputs": [], "source": [ "async def release(test_cases: list[dict]) -> None:\n", " print(\"Running pre-deployment evaluation gate...\\n\")\n", " if await evaluation_gate(test_cases, threshold=0.8):\n", " print(\"\\n✅ Gate passed — promoting agent version to the Foundry Agent Service.\")\n", " else:\n", " print(\"\\n❌ Gate failed — release blocked. Fix the agent and re-run.\")\n", "\n", "\n", "await release(TEST_CASES)" ] }, { "cell_type": "markdown", "id": "6c1333c0", "metadata": {}, "source": [ "## Summary\n", "\n", "You assembled a production-ready customer support agent with every operational concern wired in:\n", "\n", "- **Tools, RAG, and memory** give the agent capability and context.\n", "- **Model routing and caching** keep latency and cost under control.\n", "- **Human approval** guards high-risk actions like large refunds.\n", "- **The evaluation gate** blocks bad releases before they ship.\n", "- **Tracing** makes every request observable.\n", "\n", "### Challenge\n", "\n", "Extend this agent to:\n", "\n", "1. **Support multiple models** — add a third \"reasoning\" tier and route escalations/complaints to it.\n", "2. **Add evaluation gates** — expand `TEST_CASES` to include refund-approval scenarios and confirm the gate catches regressions.\n", "3. **Add cost-aware routing** — track an estimated cost per request (small vs large vs cache) and print a cost report after a batch of mixed queries.\n", "\n", "In the next lesson you take the opposite journey and run an agent entirely on your own machine with Microsoft Foundry Local and Qwen." ] } ], "metadata": { "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 5 }