Official docs: Monitor your generative AI application
Evaluation catches problems before deployment; monitoring catches them after. This lesson covers how to instrument your AI application with Azure Monitor, Application Insights, and OpenTelemetry to track quality, performance, cost, and safety in production.
┌────────────────────────────────────────────────────────────────────┐
│ YOUR AI APPLICATION │
│ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ User │──▶│ RAG Pipeline │──▶│ Azure OpenAI │ │
│ │ Request │ │ (retrieve + │ │ (generate) │ │
│ └──────────┘ │ rerank) │ └──────┬───────┘ │
│ └──────────────┘ │ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ OpenTelemetry SDK (traces + metrics) │ │
│ └──────────────────────┬──────────────────────────┘ │
└─────────────────────────┼──────────────────────────────────────────┘
│
▼
┌───────────────────────┐
│ Azure Monitor │
│ Exporter │
└───────────┬───────────┘
│
┌─────────────┼─────────────┐
▼ ▼ ▼
┌──────────────┐ ┌───────────┐ ┌───────────┐
│ Application │ │ Log │ │ Azure │
│ Insights │ │ Analytics │ │ Alerts │
│ (traces, │ │ (KQL │ │ (PagerDuty│
│ metrics) │ │ queries) │ │ Teams) │
└──────────────┘ └───────────┘ └───────────┘
│ │
▼ ▼
┌──────────────────────────┐
│ Dashboards & Workbooks │
└──────────────────────────┘
# Create Application Insights resource
az monitor app-insights component create \
--app "ai-app-insights" \
--location "eastus2" \
--resource-group "rg-ai-monitoring" \
--kind "web" \
--application-type "web"
# Get the connection string
az monitor app-insights component show \
--app "ai-app-insights" \
--resource-group "rg-ai-monitoring" \
--query "connectionString" -o tsv
pip install azure-monitor-opentelemetry opentelemetry-api opentelemetry-sdk
"""Set up Application Insights with OpenTelemetry for an AI application."""
import os
from azure.monitor.opentelemetry import configure_azure_monitor
# Configure Azure Monitor exporter — call this once at app startup
configure_azure_monitor(
connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"],
enable_live_metrics=True,
)
# Now all OpenTelemetry traces, metrics, and logs are sent to App Insights
print("Azure Monitor configured — telemetry will flow to Application Insights.")
Tip: Set
APPLICATIONINSIGHTS_CONNECTION_STRINGas an environment variable. Theconfigure_azure_monitor()function auto-detects it if you don’t pass it explicitly.
Add distributed tracing to see the full journey of each request through your RAG pipeline.
"""Instrument a RAG pipeline with OpenTelemetry traces."""
from opentelemetry import trace
from opentelemetry.trace import StatusCode
import time
tracer = trace.get_tracer("ai-app.rag-pipeline")
def handle_query(query: str) -> dict:
"""Process a user query through the full RAG pipeline with tracing."""
with tracer.start_as_current_span("rag_pipeline") as pipeline_span:
pipeline_span.set_attribute("user.query", query)
pipeline_span.set_attribute("user.query_length", len(query))
# Step 1: Retrieve documents
with tracer.start_as_current_span("retrieve_documents") as retrieve_span:
start = time.perf_counter()
documents = search_documents(query)
retrieve_latency = time.perf_counter() - start
retrieve_span.set_attribute("retrieve.doc_count", len(documents))
retrieve_span.set_attribute("retrieve.latency_ms", retrieve_latency * 1000)
retrieve_span.set_attribute("retrieve.index_name", "products-index")
# Step 2: Rerank results
with tracer.start_as_current_span("rerank_documents") as rerank_span:
ranked_docs = rerank(query, documents)
rerank_span.set_attribute("rerank.input_count", len(documents))
rerank_span.set_attribute("rerank.output_count", len(ranked_docs))
# Step 3: Generate response
with tracer.start_as_current_span("generate_response") as gen_span:
context = "\n".join(ranked_docs[:3])
result = generate_answer(query, context)
gen_span.set_attribute("generate.model", "gpt-4o")
gen_span.set_attribute("generate.prompt_tokens", result["prompt_tokens"])
gen_span.set_attribute(
"generate.completion_tokens", result["completion_tokens"]
)
gen_span.set_attribute("generate.total_tokens", result["total_tokens"])
gen_span.set_attribute("generate.finish_reason", result["finish_reason"])
# Step 4: Safety check (optional inline check)
with tracer.start_as_current_span("safety_check") as safety_span:
is_safe = check_content_safety(result["response"])
safety_span.set_attribute("safety.passed", is_safe)
if not is_safe:
safety_span.set_status(StatusCode.ERROR, "Content safety check failed")
pipeline_span.set_attribute("response.blocked", True)
return {"response": "I'm unable to answer that question.", "blocked": True}
pipeline_span.set_attribute("response.length", len(result["response"]))
pipeline_span.set_attribute("response.blocked", False)
return result
def search_documents(query: str) -> list[str]:
"""Placeholder for Azure AI Search retrieval."""
return ["Doc 1 content", "Doc 2 content", "Doc 3 content"]
def rerank(query: str, docs: list[str]) -> list[str]:
"""Placeholder for reranking logic."""
return docs
def generate_answer(query: str, context: str) -> dict:
"""Placeholder for Azure OpenAI call."""
return {
"response": f"Answer to: {query}",
"prompt_tokens": 150,
"completion_tokens": 80,
"total_tokens": 230,
"finish_reason": "stop",
}
def check_content_safety(response: str) -> bool:
"""Placeholder for content safety check."""
return True
Tip: Use semantic span names (
rag_pipeline,retrieve_documents,generate_response) so traces are readable in Application Insights. Add token counts as span attributes for cost tracking.
The Azure OpenAI SDK supports automatic OpenTelemetry instrumentation — no manual span creation needed for LLM calls.
"""Enable automatic tracing for Azure OpenAI SDK calls."""
import os
# Enable OpenAI instrumentation BEFORE creating the client
os.environ["AZURE_TRACING_GEN_AI_CONTENT_RECORDING_ENABLED"] = "true"
from azure.monitor.opentelemetry import configure_azure_monitor
configure_azure_monitor(
connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"],
)
# Now every Azure OpenAI call is automatically traced
from openai import AzureOpenAI
client = AzureOpenAI(
azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_version="2024-12-01-preview",
)
# This call will automatically generate OpenTelemetry spans
# with model name, token counts, latency, and status
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is Azure AI Foundry?"},
],
)
print(response.choices[0].message.content)
# Traces are automatically sent to Application Insights!
Tip: Set
AZURE_TRACING_GEN_AI_CONTENT_RECORDING_ENABLED=trueto capture prompt and completion text in traces. Disable this in production if prompts contain PII — set tofalseor omit the variable.
Track operational metrics that matter for AI applications.
"""Custom OpenTelemetry metrics for token usage and latency."""
from opentelemetry import metrics
import time
meter = metrics.get_meter("ai-app.metrics")
# Counters — monotonically increasing totals
token_counter = meter.create_counter(
name="ai.tokens.total",
description="Total tokens consumed across all requests",
unit="tokens",
)
prompt_token_counter = meter.create_counter(
name="ai.tokens.prompt",
description="Prompt tokens consumed",
unit="tokens",
)
completion_token_counter = meter.create_counter(
name="ai.tokens.completion",
description="Completion tokens consumed",
unit="tokens",
)
request_counter = meter.create_counter(
name="ai.requests.total",
description="Total AI requests processed",
unit="requests",
)
# Histograms — distribution of values
latency_histogram = meter.create_histogram(
name="ai.request.duration",
description="End-to-end request latency",
unit="ms",
)
ttft_histogram = meter.create_histogram(
name="ai.time_to_first_token",
description="Time from request to first token",
unit="ms",
)
def record_llm_metrics(
model: str,
prompt_tokens: int,
completion_tokens: int,
latency_ms: float,
status: str = "success",
):
"""Record metrics for an LLM call."""
attributes = {"model": model, "status": status}
token_counter.add(prompt_tokens + completion_tokens, attributes)
prompt_token_counter.add(prompt_tokens, attributes)
completion_token_counter.add(completion_tokens, attributes)
request_counter.add(1, attributes)
latency_histogram.record(latency_ms, attributes)
# Usage after each LLM call
record_llm_metrics(
model="gpt-4o",
prompt_tokens=150,
completion_tokens=80,
latency_ms=1250.5,
status="success",
)
Set up alerts to catch problems early.
# Create an alert rule for error rate > 5%
az monitor metrics alert create \
--name "ai-high-error-rate" \
--resource-group "rg-ai-monitoring" \
--scopes "/subscriptions/<sub-id>/resourceGroups/rg-ai-monitoring/providers/microsoft.insights/components/ai-app-insights" \
--condition "avg requests/failed > 5" \
--window-size "5m" \
--evaluation-frequency "1m" \
--severity 2 \
--description "AI app error rate exceeded 5%"
# Create alert for P95 latency > 5 seconds
az monitor metrics alert create \
--name "ai-high-latency-p95" \
--resource-group "rg-ai-monitoring" \
--scopes "/subscriptions/<sub-id>/resourceGroups/rg-ai-monitoring/providers/microsoft.insights/components/ai-app-insights" \
--condition "avg requests/duration > 5000" \
--window-size "5m" \
--evaluation-frequency "1m" \
--severity 3 \
--description "AI app P95 latency exceeded 5 seconds"
# Log-based alert using KQL for daily token budget
az monitor scheduled-query create \
--name "ai-token-budget-alert" \
--resource-group "rg-ai-monitoring" \
--scopes "/subscriptions/<sub-id>/resourceGroups/rg-ai-monitoring/providers/microsoft.insights/components/ai-app-insights" \
--condition "count > 0" \
--condition-query "customMetrics | where name == 'ai.tokens.total' | summarize DailyTokens = sum(value) by bin(timestamp, 1d) | where DailyTokens > 1000000" \
--window-size "1h" \
--evaluation-frequency "1h" \
--severity 3 \
--description "Daily token usage approaching budget limit"
Use these queries in Log Analytics or Application Insights to gain insights into your AI application.
customMetrics
| where name in ("ai.tokens.prompt", "ai.tokens.completion")
| summarize TotalTokens = sum(value) by bin(timestamp, 1h), name
| render timechart
customMetrics
| where name == "ai.request.duration"
| summarize
P50 = percentile(value, 50),
P95 = percentile(value, 95),
P99 = percentile(value, 99),
Avg = avg(value)
by bin(timestamp, 15m)
| render timechart
requests
| where success == false
| summarize ErrorCount = count() by resultCode, bin(timestamp, 1h)
| order by timestamp desc
| render barchart
requests
| where duration > 3000
| project timestamp, name, duration, resultCode,
customDimensions["user.query"] as Query,
customDimensions["generate.model"] as Model
| order by duration desc
| take 50
customEvents
| where name == "UserFeedback"
| extend
Feedback = tostring(customDimensions["feedback"]),
QueryId = tostring(customDimensions["query_id"])
| summarize
Positive = countif(Feedback == "positive"),
Negative = countif(Feedback == "negative")
by bin(timestamp, 1d)
| extend SatisfactionRate = round(100.0 * Positive / (Positive + Negative), 1)
| render timechart
customMetrics
| where name == "ai.request.duration"
| extend Model = tostring(customDimensions["model"])
| summarize
AvgLatency = avg(value),
P95Latency = percentile(value, 95),
RequestCount = count()
by Model, bin(timestamp, 1h)
| render timechart
Create a monitoring dashboard that gives your team a single-pane view.
┌─────────────────────────────────────────────────────────────┐
│ AI APPLICATION DASHBOARD │
├──────────────────────┬──────────────────────────────────────┤
│ REQUEST VOLUME │ LATENCY (P50 / P95 / P99) │
│ ┌────────────────┐ │ ┌──────────────────────────────┐ │
│ │ █████████████ │ │ │ ─── P50 ─── P95 ─── P99│ │
│ │ ██████████████ │ │ │ ─────────────────────── │ │
│ │ ████████████ │ │ │ ───────────────── │ │
│ └────────────────┘ │ └──────────────────────────────┘ │
├──────────────────────┼──────────────────────────────────────┤
│ TOKEN USAGE (24h) │ ERROR RATE │
│ ┌────────────────┐ │ ┌──────────────────────────────┐ │
│ │ Prompt: 1.2M │ │ │ Current: 0.3% │ │
│ │ Completion: 0.5M│ │ │ ────────────── threshold │ │
│ │ Total: 1.7M │ │ │ ───────── │ │
│ │ Est. cost: $8.50│ │ └──────────────────────────────┘ │
│ └────────────────┘ │ │
├──────────────────────┼──────────────────────────────────────┤
│ QUALITY SCORES │ USER SATISFACTION │
│ (sampled eval) │ ┌──────────────────────────────┐ │
│ Groundedness: 4.3 │ │ 👍 85% 👎 15% │ │
│ Relevance: 4.5 │ │ Trend: ↑ 2% from last week │ │
│ Coherence: 4.6 │ └──────────────────────────────┘ │
│ Safety pass: 99.8% │ │
├──────────────────────┴──────────────────────────────────────┤
│ RECENT ALERTS │
│ • 14:32 — P95 latency spike (6.2s) on gpt-4o │
│ • 09:15 — Token budget at 80% for the day │
└─────────────────────────────────────────────────────────────┘
# Create a workbook for the AI monitoring dashboard
az monitor app-insights workbook create \
--resource-group "rg-ai-monitoring" \
--name "AI App Monitoring" \
--location "eastus2" \
--kind "shared" \
--category "AI"
Tip: Use Azure Workbooks for interactive dashboards that combine KQL queries, metrics charts, and markdown text. Pin individual queries to an Azure Dashboard for a simpler overview.
Track and estimate costs in real-time based on token usage.
"""Estimate Azure OpenAI costs from observed token usage."""
# Pricing per 1K tokens (example — check current pricing)
MODEL_PRICING = {
"gpt-4o": {"prompt": 0.005, "completion": 0.015},
"gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
"gpt-4": {"prompt": 0.03, "completion": 0.06},
}
def estimate_cost(
model: str, prompt_tokens: int, completion_tokens: int
) -> dict:
"""Estimate cost for a single request."""
pricing = MODEL_PRICING.get(model, {"prompt": 0.01, "completion": 0.03})
prompt_cost = (prompt_tokens / 1000) * pricing["prompt"]
completion_cost = (completion_tokens / 1000) * pricing["completion"]
return {
"prompt_cost": round(prompt_cost, 6),
"completion_cost": round(completion_cost, 6),
"total_cost": round(prompt_cost + completion_cost, 6),
}
def daily_cost_report(daily_metrics: list[dict]) -> dict:
"""Generate a daily cost report from collected metrics."""
total_cost = 0.0
model_costs = {}
for entry in daily_metrics:
cost = estimate_cost(
model=entry["model"],
prompt_tokens=entry["prompt_tokens"],
completion_tokens=entry["completion_tokens"],
)
total_cost += cost["total_cost"]
model_costs.setdefault(entry["model"], 0.0)
model_costs[entry["model"]] += cost["total_cost"]
return {
"total_daily_cost": round(total_cost, 4),
"cost_by_model": {k: round(v, 4) for k, v in model_costs.items()},
"total_requests": len(daily_metrics),
"projected_monthly": round(total_cost * 30, 2),
}
# Example usage
daily_data = [
{"model": "gpt-4o", "prompt_tokens": 500, "completion_tokens": 200},
{"model": "gpt-4o", "prompt_tokens": 800, "completion_tokens": 150},
{"model": "gpt-4o-mini", "prompt_tokens": 300, "completion_tokens": 100},
]
report = daily_cost_report(daily_data)
print(f"Daily cost: ${report['total_daily_cost']}")
print(f"Projected monthly: ${report['projected_monthly']}")
for model, cost in report["cost_by_model"].items():
print(f" {model}: ${cost}")
Run scheduled evaluations on sampled production traffic to detect quality drift.
"""Schedule periodic quality evaluation on sampled production data."""
import json
import random
from datetime import datetime
from azure.ai.evaluation import evaluate, GroundednessEvaluator, RelevanceEvaluator
model_config = {
"azure_deployment": "gpt-4o",
"azure_endpoint": "https://<your-endpoint>.openai.azure.com/",
"api_version": "2024-12-01-preview",
}
def sample_production_data(log_path: str, sample_size: int = 50) -> str:
"""Sample recent production logs for evaluation."""
with open(log_path, "r") as f:
logs = [json.loads(line) for line in f]
# Sample from the last 24 hours
sampled = random.sample(logs, min(sample_size, len(logs)))
output_path = f"eval_sample_{datetime.now().strftime('%Y%m%d_%H%M')}.jsonl"
with open(output_path, "w") as f:
for entry in sampled:
f.write(json.dumps(entry) + "\n")
return output_path
def run_periodic_evaluation(sample_path: str) -> dict:
"""Run evaluation on sampled data and check for quality drift."""
result = evaluate(
data=sample_path,
evaluators={
"groundedness": GroundednessEvaluator(model_config=model_config),
"relevance": RelevanceEvaluator(model_config=model_config),
},
evaluator_config={
"groundedness": {
"query": "${data.query}",
"context": "${data.context}",
"response": "${data.response}",
},
"relevance": {
"query": "${data.query}",
"context": "${data.context}",
"response": "${data.response}",
},
},
)
metrics = result["metrics"]
# Check for quality drift (compare to baseline)
BASELINE = {"groundedness.groundedness": 4.2, "relevance.relevance": 4.3}
print(f"=== Periodic Evaluation ({datetime.now()}) ===")
drift_detected = False
for metric, baseline in BASELINE.items():
current = metrics.get(metric, 0)
delta = current - baseline
drift_flag = "⚠️ DRIFT" if delta < -0.3 else "✅ OK"
print(f" {metric}: {current:.2f} (baseline: {baseline}, Δ: {delta:+.2f}) {drift_flag}")
if delta < -0.3:
drift_detected = True
if drift_detected:
print("\n⚠️ Quality drift detected — investigate recent changes!")
# Trigger an alert or notification here
return metrics
# Schedule this function to run daily via Azure Functions timer trigger,
# CRON job, or Azure Data Factory pipeline.
Tip: Set your baseline scores from your last pre-deployment evaluation. A drop of 0.3+ points from baseline is a strong signal of quality drift — investigate recent data source changes, model updates, or prompt modifications.
| Component | Tool | Purpose |
|---|---|---|
| Tracing | OpenTelemetry + App Insights | End-to-end request visibility |
| Auto-instrumentation | Azure OpenAI SDK tracing | Zero-code LLM call traces |
| Custom metrics | OpenTelemetry meters | Token usage, latency, request counts |
| Alerts | Azure Monitor alerts | Proactive issue detection |
| KQL queries | Log Analytics | Deep-dive investigation |
| Dashboards | Azure Workbooks | Team visibility |
| Cost tracking | Token-based estimation | Budget management |
| Quality sampling | Periodic evaluate() |
Drift detection |
| ← Lesson 2: Custom Evaluators | Next → Knowledge Check |