





















A user submits a prompt. The agent burns through 50,000 tokens looping on a malformed tool response. Another user trips a model rate limit and the agent silently fails. A bad prompt update goes out at 4 PM Friday and degrades success rate to 60%. Your APM dashboard shows green the entire time because none of that is a 500.
This post walks through the LLMOps stack we built into a working reference sample on Azure App Service: the SLIs that matter for agents, a budget circuit breaker, prompt-repair retries, and a fully automated slot-swap rollback when things go sideways. Every code snippet is from the deployable sample at the end of the post.
📦 Sample repo: seligj95/app-service-self-healing-agent-python —
azd upand you've got the whole stack live in your subscription in under 10 minutes.
Your web app's reliability model assumes a request maps to bounded work — a SQL query, a cache hit, a templated response. You alert on Http5xx, p95 latency, and dependency failures. Done.
An agent breaks that model in four ways:
Web-app SLOs (uptime, latency, error rate) are necessary but not sufficient. Agents need agent-shaped SLOs.
Before instrumenting anything, write down what "healthy" means. Here are the four SLIs we chose for the sample. None of them are Http5xx.
| SLI | What it measures | Why it matters |
|---|---|---|
| Task success rate | % of /chat requests that the agent self-classifies as completed | Catches silent failures the HTTP layer misses |
| Cost per task | $ spent (input + output tokens × model rate) per /chat | The unbounded-loop problem in one number |
| Tool success rate | % of tool invocations that didn't raise | Tool layer is where most agent failures live |
| Repair retries | Times we re-prompted the model after a schema-validation failure | Leading indicator of prompt drift |
In our reference middleware these come out as agent.task.success, agent.cost.usd, agent.tool.success, and agent.repair.retry — eleven custom metrics in total. We emit them via OpenTelemetry so they land in App Insights customMetrics and the included KQL workbook visualizes them as SLO tiles.
App Service makes the observability story unusually easy because you get App Insights wired up automatically by azd — no agent install, no DaemonSet, no sidecar. The only thing you bring is the SDK init for your custom metrics:
# llmops_middleware/sli.py
from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry import metrics
def configure_azure_monitor_if_available() -> bool:
if not os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING"):
return False
configure_azure_monitor()
return True
meter = metrics.get_meter("agent")
tokens_in = meter.create_counter("agent.tokens.in")
cost_usd = meter.create_counter("agent.cost.usd")
task_latency = meter.create_histogram("agent.task.latency")
tool_success = meter.create_counter("agent.tool.success")
# ...
We compute cost from a per-model rate card so the metric is in real dollars, not abstract tokens:
COST_PER_1K_TOKENS = {
"gpt-4o": {"in": 0.0025, "out": 0.01},
"gpt-4o-mini": {"in": 0.00015, "out": 0.0006},
}
def record_cost(model: str, tokens_in_count: int, tokens_out_count: int, tenant: str) -> float:
rate = COST_PER_1K_TOKENS[model]
cost = (tokens_in_count * rate["in"] + tokens_out_count * rate["out"]) / 1000
cost_usd.add(cost, {"model": model, "tenant": tenant})
return cost
Once those flow, the KQL queries write themselves:
// Top cost-burning tenants in the last hour
customMetrics
| where timestamp > ago(1h)
| where name == "agent.cost.usd"
| extend tenant = tostring(customDimensions["tenant"])
| summarize spend_usd = sum(valueSum) by tenant
| top 10 by spend_usd desc
The sample ships a 6-tile workbook (observability/workbook.json) deployed via Bicep. It renders SLO compliance, cost burn-down, tool failure breakdown, latency percentiles, budget breaches, and healing signals out of the box.
The deployed workbook in App Insights. The SLO panel dips during a chaos run and recovers as the agent self-heals — exactly the signal you want on a glass-pane dashboard.
Custom metrics tell you about cost after you spent it. To prevent runaways, you need a circuit breaker that bites before the model call happens.
The middleware in llmops_middleware/budget.py keeps a per-tenant counter in memory (per month) and returns a decision:
class BudgetDecision(Enum):
ALLOW = "allow" # under budget
DOWNSHIFT = "downshift" # ≥80% — switch to cheaper model
BLOCK = "block" # ≥100% — refuse the request
def evaluate(tenant: str) -> BudgetDecision:
spent = _spend.get((tenant, _current_period()), 0.0)
if spent >= BUDGET_USD_PER_TENANT:
return BudgetDecision.BLOCK
if spent >= BUDGET_USD_PER_TENANT * 0.80:
return BudgetDecision.DOWNSHIFT
return BudgetDecision.ALLOW
The agent loop reads that decision and downshifts from gpt-4o to gpt-4o-mini — a 16× cost reduction ($0.0025 / 1K input tokens vs $0.00015) — when a tenant crosses 80% of their monthly budget. The user keeps getting answers; the bill stops climbing.
def _pick_model(tenant: str) -> str:
decision = budget.evaluate(tenant)
if decision == BudgetDecision.DOWNSHIFT:
sli.model_downshift.add(1, {"tenant": tenant})
return DOWNSHIFT_MODEL
return PRIMARY_MODEL
For the demo we keep state in memory; production should swap the dict for Redis (atomic INCRBY) or Cosmos with optimistic concurrency. The interface in budget.py is intentionally tiny so this is a 10-line change.
There are three patterns in the sample, each addressing a different failure class.
The most common agent failure isn't a tool exception — it's the model returning malformed JSON that fails schema validation on tool args. The fix is to feed the validation error back into the model and ask it to repair the call:
# llmops_middleware/repair.py
async def retry_with_repair(call_fn, args, *, max_attempts=2):
for attempt in range(max_attempts):
try:
return await call_fn(args)
except (ValidationError, RepairableError) as exc:
sli.repair_retry.add(1, {"attempt": str(attempt)})
args = await _ask_model_to_repair(args, str(exc))
raise
This single pattern recovers 50–70% of "the agent returned garbage" cases without escalating.
When a primary tool times out or fails open, try a cheaper or simpler one:
async def tool_fallback_chain(primary, *fallbacks, args):
for fn in (primary, *fallbacks):
try:
return await fn(args)
except ToolUnavailable:
sli.tool_success.add(1, {"tool": fn.__name__, "status": "fallback"})
raise NoToolAvailable()
Lookup-style tools especially benefit: web search → cached snapshot → static knowledge base.
Here's the killer feature App Service brings that's a slog on K8s: deployment slots. You always have a known-good previous version warmed up and one ARM API call away from production traffic. We wire that up to fire automatically when our SLI breaches.
The chain is:
Http5xx > 5 in 5 minutes (the platform metric, free)POST /sites/{name}/slots/staging/slotsswap via its managed identity (granted Website Contributor on the target web app)The whole healer is one trigger + two actions: receive the alert webhook, call ARM slotsswap, return a status payload to the caller.
The two actions in Bicep:
SwapSlots: {
type: 'Http'
inputs: {
method: 'POST'
uri: '${environment().resourceManager}@{parameters(\'targetSiteId\')}/slots/staging/slotsswap?api-version=2024-04-01'
body: { targetSlot: 'production' }
authentication: {
type: 'ManagedServiceIdentity'
audience: environment().resourceManager
}
}
}
No code to deploy, no secrets to manage, no second runtime to babysit. From alert-fire to swapped-slot is about 4 minutes in our tests — under the SLA most agent products have for "user-visible degraded mode."
Why not a Function App? We started there. The Logic App is 60 lines of Bicep and zero application code. For a one-action workflow like "swap a slot," the Function adds packaging, deployment, and a runtime to monitor for no benefit.
You can't trust a self-healing system you haven't broken. The sample ships a chaos CLI and an in-process injection point so you can practice failures on demand.
In-process: llmops_middleware/chaos.py exposes four modes (off, throttle, malformed, outage) togglable via POST /admin/chaos. When set, tool calls roll a die and raise the matching exception with the configured probability:
class ChaosController:
def maybe_inject(self) -> None:
if random.random() > self.probability:
return
if self.mode == "outage":
raise ChaosOutage("simulated tool outage")
if self.mode == "throttle":
raise ChaosThrottled("simulated 429")
if self.mode == "malformed":
raise ChaosMalformed("simulated bad tool output")
External: chaos/inject.py is a small async load driver that sets /admin/chaos then drives /chat at a target RPS, tallying response codes:
python chaos/inject.py \
--base-url https://my-agent.azurewebsites.net \
--mode outage --probability 1.0 --rps 10 --duration 300
Running that for 5 minutes against the deployed sample reliably:
customMetrics(name="agent.task.failure") over 50/minHttp5xx > 5 metric alert (~90 seconds after threshold breach)/health instance ID changesThe repo's observability/queries.kql has the canonical KQL for each of these signals, and observability/workbook.json is the deployable workbook that visualizes them.
Everything in this post is in seligj95/app-service-self-healing-agent-python. The Python package llmops_middleware/ is the part you'd vendor into your own agent — sli.py, budget.py, repair.py, chaos.py. The agent loop and the Bicep are demo-quality but production-shaped.
Run it yourself:
git clone https://github.com/seligj95/app-service-self-healing-agent-python
cd app-service-self-healing-agent-python
azd auth login
azd up
You'll have an agent + AOAI + workbook + healer running in about 8 minutes. Then run the chaos script and watch the slot flip.
Deployable workbook JSON, dropped into the resource group by Bicep. Six panels:
agent.task.success was emitted (grouped by tenant)agent.task.latency histogramagent.repair.retry + agent.model.downshift + agent.chaos.injected over timeIt's observability/workbook.json — loadTextContent-ed into infra/shared/monitoring.bicep so you get it deployed automatically.
After building this, the appeal of App Service for agents is clearer than I expected going in:
disableLocalAuth: true on the AOAI account — there literally is no key.customMetrics and your KQL queries work day one.If you're standing up a new agent and you don't already have a Kubernetes platform you love, App Service is a strong default.
If you take three things from this post:
The sample has all of it wired up.
The middleware in this sample (SLIs + telemetry, cost guardrails, policy/audit hooks) is exactly the kind of thing we're evaluating as first-class App Service platform features — opt-in sidecars or built-in capabilities so you don't have to vendor a middleware package into every agent you ship. Concretely, we're tracking ideas like:
If any of those would land for your team — or if you're solving these problems differently and want to push back on the shape — we want to hear it. Drop a comment on this post: the roadmap is genuinely shaped by feedback at this stage.
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。