Phase 8: Production AI Security Architecture · 70 min · LangSmith · Langfuse · Arize Phoenix
Security Monitoring & Observability
An attack you can't see in your dashboard is an incident you'll read about in a postmortem.
What you will learn
- Explain what LangSmith, Langfuse, Phoenix, and Helicone each instrument by default and what security-relevant signal you have to derive yourself
- Compute a rolling z-score anomaly detector on token cost time series to catch cost-spike attacks (context flooding, token exhaustion DoS)
- Compute a rolling injection-attempt rate and alert threshold from flagged-request telemetry
- Build a Markov-chain likelihood model over tool-call sequences to detect anomalous agent behavior that individual tool-call checks would miss
The Problem
LLM observability platforms — LangSmith, Langfuse, Arize Phoenix, Helicone — solved a real problem: you can now see every prompt, every completion, every token cost, and every tool call your production system makes. Teams adopt one of these tools, wire up tracing, and feel like they have security visibility.
They don't, not yet. These platforms give you the raw traces. They do not ship a default answer to "is this pattern of token cost anomalous," "is this rate of flagged requests an attack or noise," or "is this tool-call sequence something an attacker would produce." That's a modeling problem you have to solve yourself, on top of the tracing layer — and it's exactly the kind of problem AI Security Engineer and DevSecOps-for-AI interviews probe, because it separates people who've wired up a dashboard from people who've operationalized one.
Three signals matter most for security, specifically because each one catches an attack class that guardrails (Phase 5) and per-request input filtering (Phase 8, Lesson 1) structurally cannot:
- Token cost — a context-flooding or token-exhaustion DoS attack (OWASP LLM04) doesn't look like a bad prompt; it looks like a normal-looking request that costs 10x what your average request costs. No per-request filter catches "this is expensive" without a baseline to compare against.
- Injection attempt rate — a single flagged prompt injection attempt might be noise (an overzealous classifier, a confused user). A burst of them is a campaign. The difference is only visible in a rolling rate, not a per-request classification.
- Tool-call sequence — Phase 4 covered agent privilege escalation and tool poisoning: attacks that chain individually-authorized tool calls into something dangerous. A
send_email call and a delete_record call might each pass tool authorization independently. The sequence read_document -> send_email -> delete_record is what should trip an alarm, and no single-call check will ever see it.
Rolling Z-score: catching cost spikes before they're an incident
A z-score measures how many standard deviations a point is from a rolling mean. For token cost, this is the right primitive because "normal" cost varies by traffic pattern and time of day — an absolute dollar threshold either fires constantly or misses slow-building attacks.
def cost_anomalies(self, costs: list[float], window: int = 20) -> list[Alert]:
for i in range(window, len(costs)):
hist = costs[i - window : i]
mean = sum(hist) / window
std = (sum((x - mean) ** 2 for x in hist) / window) ** 0.5 or 1e-9
z = (costs[i] - mean) / std
if z >= self.cost_z_threshold:
... # alert
The threshold you pick (this lesson uses 3.0, i.e. 3 standard deviations) is a real tradeoff, not an arbitrary constant. Set it too low and normal traffic variance triggers alert fatigue — the demo run below shows exactly this, with two borderline 3.4-3.5 sigma events that are plausibly just noise alongside one unambiguous 67-sigma spike that is not. Set it too high and you'll miss a genuine attack that ramps cost gradually to stay under a wide threshold. In a real system, you validate the threshold against a labeled window of known-normal traffic and tune it until the false-positive rate is tolerable for whoever is on call.
Why does this lesson's cost-anomaly detector use a rolling z-score against a sliding window instead of a fixed absolute dollar threshold (e.g., "alert if any request costs more than $0.50")?
The lesson makes this explicit: "'normal' cost varies by traffic pattern and time of day — an absolute dollar threshold either fires constantly or misses slow-building attacks." A rolling z-score recalculates its own baseline (mean and standard deviation) from the preceding window at every point, so it flags a request that is unusual relative to recent behavior, not relative to a number picked once and never revisited. This is what lets it catch a sudden 67-sigma context-flooding spike and still stay quiet during a legitimate traffic surge that raises everyone's cost proportionally.
Rolling rate: catching a campaign, not a single flag
Injection detectors from Phase 1 and Phase 2 flag individual requests. But a single flagged request tells you almost nothing about whether you're under active attack — it could be one confused user typing something that pattern-matched a heuristic. What tells you it's a campaign is a rate of flagged requests over a short window that's far above baseline.
def injection_rate_alerts(self, flagged: list[bool], window: int = 50) -> list[Alert]:
for i in range(window, len(flagged) + 1):
rate = sum(flagged[i - window : i]) / window
if rate >= self.injection_rate_threshold:
... # alert, severity scales with how far above threshold
This is the same reason a fraud detection system alerts on "10 failed logins in 2 minutes" rather than "1 failed login" — the individual signal is weak, the rate is strong. Averaged over a full day of traffic, a 30-request injection burst inside 10,000 total requests moves the daily average by a fraction of a percent — invisible. Over a 50-request rolling window, that same burst is an obvious 30-45% flag rate against a ~3% baseline.
Don't let your own detector's false-positive rate become your attack signal
If your injection classifier has a 5% false-positive rate on legitimate traffic, your rolling-rate baseline needs to account for that — set the alert threshold meaningfully above the expected false-positive rate (this lesson uses 15% against a ~3% baseline), or you'll page someone every day for nothing and they'll start ignoring the alert. This is the security-monitoring equivalent of alert fatigue, and it's the single most common reason real security teams stop trusting their own dashboards.
Markov-chain likelihood: catching sequences, not calls
This is the signal that catches what per-call authorization misses. Fit a first-order Markov chain over a corpus of known-benign tool-call sessions — the transition probability P(next_tool | current_tool) — then score any new session by its log-likelihood under that model. A session built from individually-legitimate tools, in an ordering that never occurs in benign usage, gets a very low likelihood even though every call in it would pass an allowlist check.
An agent has tool-authorization rules that allow send_email for any authenticated user and delete_record for any user with edit permission. Both calls individually pass authorization. A compromised session executes read_document -> send_email -> delete_record — a pattern that has never occurred in 80 sessions of benign baseline traffic. Which detection layer from this course is designed to catch this specific attack?
Explanation: Per-call authorization (Phase 4/5) is necessary but structurally blind to sequence-level attacks: it answers "is this call allowed" one call at a time, never "does this session's pattern of calls look like anything we've seen before." A Markov-chain likelihood model scores the full transition sequence against a baseline of normal sessions, so a chain built entirely from individually-authorized calls can still be flagged as anomalous if the ordering itself is statistically unlikely — exactly the exfiltration/privilege-escalation pattern (read data, then send it externally, then delete evidence) that this lesson's demo catches.
Building the chain requires smoothing, because a real system will always encounter tool-call transitions it hasn't seen in the baseline yet:
def fit_tool_call_baseline(self, sequences: list[list[str]]) -> dict[str, dict[str, float]]:
# counts[a][b] = how often tool b directly follows tool a in benign sessions
# Laplace (+1) smoothing over the full observed vocabulary ensures every
# possible transition gets a small non-zero probability instead of a
# divide-by-zero or an artificially "impossible" score for something
# merely rare.
...
Without smoothing, any transition absent from the baseline gets probability zero, and a log-likelihood of log(0) is undefined — the model would crash on the first novel-but-benign sequence instead of scoring it as "unusual but not impossible."
What LangSmith, langfuse, phoenix, and helicone actually give you
Each of these platforms instruments the trace layer — every LLM call, its inputs, outputs, latency, and token usage — with different strengths: LangSmith integrates tightly with LangChain/LangGraph agent traces; Langfuse is open-source and self-hostable, strong on session/user-level analytics; Arize Phoenix focuses on embeddings-based drift and evaluation; Helicone is a thin proxy-based layer that's fast to bolt onto any OpenAI-compatible API. None of them ship "this is a prompt injection campaign" out of the box — they give you the queryable trace data that the SecurityMonitor in this lesson would run against in production, instead of the synthetic lists used here for a runnable, dependency-free demo.
Build
Build it
The code artifact below implements all three signals from scratch — z-score cost anomalies, rolling injection rate, and Markov-chain tool-sequence likelihood — plus synthetic telemetry generators that inject a realistic cost spike, injection burst, and two adversarial tool-call chains, and a markdown dashboard renderer that reports all three signals together with an overall status.
"""
lesson_08_02_security_monitor.py
A SecurityMonitor that computes three real, statistically grounded security
signals over synthetic production telemetry -- the same class of signal that
LangSmith, Langfuse, Arize Phoenix, and Helicone expose traces for, but which
none of them compute as a security alert out of the box. You have to define
"anomalous" yourself.
Signals implemented from scratch:
1. Rolling z-score anomaly detection on token cost (catches context-flooding
/ token-exhaustion DoS and runaway agent loops).
2. Rolling injection-attempt rate (catches a burst of flagged requests that
an average-over-all-time metric would smooth away).
3. Tool-call sequence anomaly via a Markov-chain likelihood model (catches
agent behavior that no single tool call would flag, but that is wildly
unlikely given the baseline of normal sessions).
Run:
python lesson_08_02_security_monitor.py
No external services, no API keys, no network calls -- all data is synthetic
and generated deterministically.
"""
from __future__ import annotations
import math
import random
from collections import defaultdict
from dataclasses import dataclass
random.seed(42)
# ---------------------------------------------------------------------------
# Alert model
# ---------------------------------------------------------------------------
@dataclass
class Alert:
signal: str
severity: str
message: str
index: int | None = None
def _dedupe_consecutive(alerts: list[Alert]) -> list[Alert]:
"""Collapse runs of consecutive-index alerts on the same signal into the first one.
A sustained anomaly (e.g. a 30-request injection burst) would otherwise raise
30 separate alerts for what is operationally one event.
"""
if not alerts:
return []
out = [alerts[0]]
last_raw_index = alerts[0].index
for a in alerts[1:]:
prev = out[-1]
if (
a.signal == prev.signal
and a.index is not None
and last_raw_index is not None
and a.index - last_raw_index <= 1
):
last_raw_index = a.index
continue
out.append(a)
last_raw_index = a.index
return out
# ---------------------------------------------------------------------------
# SecurityMonitor
# ---------------------------------------------------------------------------
class SecurityMonitor:
def __init__(
self,
cost_z_threshold: float = 3.0,
injection_rate_threshold: float = 0.15,
markov_likelihood_threshold: float = -3.5,
) -> None:
self.cost_z_threshold = cost_z_threshold
self.injection_rate_threshold = injection_rate_threshold
self.markov_likelihood_threshold = markov_likelihood_threshold
# -- Signal 1: token cost z-score -------------------------------------
def cost_anomalies(self, costs: list[float], window: int = 20) -> list[Alert]:
"""Rolling z-score of each cost point against the preceding window.
A cost spike from context flooding or a runaway tool loop shows up as
a large positive z-score well before it would move a daily/weekly
cost average enough to notice.
"""
alerts: list[Alert] = []
for i in range(window, len(costs)):
hist = costs[i - window : i]
mean = sum(hist) / window
var = sum((x - mean) ** 2 for x in hist) / window
std = math.sqrt(var) or 1e-9
z = (costs[i] - mean) / std
if z >= self.cost_z_threshold:
alerts.append(
Alert(
"token_cost",
"high",
f"index {i}: cost=${costs[i]:.4f} is {z:.1f} std devs above rolling mean "
f"${mean:.4f} (window={window})",
i,
)
)
return _dedupe_consecutive(alerts)
# -- Signal 2: injection attempt rate ----------------------------------
def injection_rate_alerts(self, flagged: list[bool], window: int = 50) -> list[Alert]:
"""Rolling rate of injection-flagged requests over a sliding window.
Averaged over a full day, a 30-request injection burst inside 10,000
normal requests is invisible. Over a 50-request rolling window, it's a
rate spike from ~2% baseline to 40%+.
"""
alerts: list[Alert] = []
for i in range(window, len(flagged) + 1):
hist = flagged[i - window : i]
rate = sum(hist) / window
if rate >= self.injection_rate_threshold:
alerts.append(
Alert(
"injection_rate",
"critical" if rate >= 0.30 else "high",
f"window ending at request {i - 1}: injection-flagged rate {rate:.0%} "
f"over last {window} requests (threshold {self.injection_rate_threshold:.0%})",
i - 1,
)
)
return _dedupe_consecutive(alerts)
# -- Signal 3: tool-call sequence anomaly (Markov chain) ---------------
def fit_tool_call_baseline(self, sequences: list[list[str]]) -> dict[str, dict[str, float]]:
"""Fit a first-order Markov chain over observed benign tool-call sequences.
Returns transition probabilities P(next_tool | current_tool), with
<START> and <END> sentinels and add-one (Laplace) smoothing so unseen
transitions get a small non-zero probability instead of an outright
divide-by-zero.
"""
counts: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
vocab: set[str] = {"<START>", "<END>"}
for seq in sequences:
vocab.update(seq)
for seq in sequences:
tokens = ["<START>"] + seq + ["<END>"]
for a, b in zip(tokens, tokens[1:]):
counts[a][b] += 1
probs: dict[str, dict[str, float]] = {}
for a in vocab:
row_counts = counts.get(a, {})
total = sum(row_counts.values()) + len(vocab) # +1 smoothing per possible next-token
row = {b: (row_counts.get(b, 0) + 1) / total for b in vocab}
row["_unseen_state_default"] = 1.0 / total
probs[a] = row
return probs
def sequence_log_likelihood(self, seq: list[str], transition_probs: dict[str, dict[str, float]]) -> float:
"""Log-likelihood of a tool-call sequence under the fitted baseline chain,
normalized by sequence length so long and short sessions are comparable."""
tokens = ["<START>"] + seq + ["<END>"]
log_l = 0.0
for a, b in zip(tokens, tokens[1:]):
row = transition_probs.get(a)
if row is None:
p = 1e-6
else:
p = row.get(b, row.get("_unseen_state_default", 1e-6))
log_l += math.log(p)
return log_l / max(len(tokens) - 1, 1)
def tool_sequence_alerts(
self,
sessions: list[list[str]],
transition_probs: dict[str, dict[str, float]],
) -> list[Alert]:
"""Flag sessions whose normalized log-likelihood under the baseline chain
falls below threshold -- i.e. tool-call orderings that almost never happen
in normal usage, even if every individual call is independently authorized."""
alerts: list[Alert] = []
for i, seq in enumerate(sessions):
ll = self.sequence_log_likelihood(seq, transition_probs)
if ll <= self.markov_likelihood_threshold:
alerts.append(
Alert(
"tool_sequence",
"high",
f"session {i}: sequence {seq} -> normalized log-likelihood {ll:.2f} "
f"(threshold {self.markov_likelihood_threshold}) — statistically anomalous ordering",
i,
)
)
return alerts
# -- Dashboard -----------------------------------------------------------
def dashboard(
self,
cost_alerts: list[Alert],
injection_alerts: list[Alert],
tool_alerts: list[Alert],
total_requests: int,
total_sessions: int,
) -> str:
lines = ["# Security Monitoring Dashboard", ""]
lines.append(f"Requests analyzed: {total_requests} · Agent sessions analyzed: {total_sessions}")
lines.append("")
lines.append(f"## Signal 1 — Token Cost Anomalies ({len(cost_alerts)} alert(s))")
if cost_alerts:
for a in cost_alerts:
lines.append(f"- [{a.severity.upper()}] {a.message}")
else:
lines.append("- No cost anomalies detected.")
lines.append("")
lines.append(f"## Signal 2 — Injection Attempt Rate ({len(injection_alerts)} alert(s))")
if injection_alerts:
for a in injection_alerts:
lines.append(f"- [{a.severity.upper()}] {a.message}")
else:
lines.append("- Injection rate within baseline for all windows.")
lines.append("")
lines.append(f"## Signal 3 — Tool-Call Sequence Anomalies ({len(tool_alerts)} alert(s))")
if tool_alerts:
for a in tool_alerts:
lines.append(f"- [{a.severity.upper()}] {a.message}")
else:
lines.append("- All sessions consistent with baseline tool-call patterns.")
lines.append("")
total_alerts = len(cost_alerts) + len(injection_alerts) + len(tool_alerts)
status = "CLEAN" if total_alerts == 0 else ("REVIEW" if total_alerts <= 2 else "INVESTIGATE")
lines.append(f"**Overall status: {status}** ({total_alerts} total alert(s) across 3 signals)")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Synthetic telemetry generators
# ---------------------------------------------------------------------------
def generate_cost_series(n: int = 200) -> list[float]:
"""Baseline token cost per request ~ $0.02 +/- noise, with an injected
cost-spike attack (simulated context-flooding) around index 150-154."""
costs = [max(0.001, random.gauss(0.02, 0.004)) for _ in range(n)]
for i in range(150, 155):
costs[i] = random.gauss(0.24, 0.02) # ~10x spike: context-flooding DoS
return costs
def generate_injection_flags(n: int = 300) -> list[bool]:
"""Baseline ~3% of requests get flagged by the input-layer injection
detector (some false positives are normal). A sustained attack burst
hits 100-129 at a ~45% flag rate."""
flags = [random.random() < 0.03 for _ in range(n)]
for i in range(100, 130):
flags[i] = random.random() < 0.45
return flags
TOOL_TEMPLATES = [
["search_docs", "read_document", "summarize", "respond_to_user"],
["query_database", "summarize", "respond_to_user"],
["call_calendar_api", "respond_to_user"],
["search_docs", "read_document", "write_file", "respond_to_user"],
]
def generate_baseline_sessions(n: int = 60) -> list[list[str]]:
sessions = []
for _ in range(n):
template = random.choice(TOOL_TEMPLATES)
seq = list(template)
# small natural variation: occasionally repeat a read/search step
if random.random() < 0.15 and "read_document" in seq:
idx = seq.index("read_document")
seq.insert(idx, "read_document")
sessions.append(seq)
return sessions
def generate_test_sessions() -> list[list[str]]:
"""A mix of normal sessions plus two adversarial tool-call chains that
an agent should never legitimately produce: exfiltration and destructive
privilege escalation, each built entirely from individually-authorized
tools -- the reason per-call authorization (guardrails layer) alone
would miss both."""
normal = generate_baseline_sessions(6)
adversarial = [
["search_docs", "read_document", "send_email", "delete_record"],
["query_database", "delete_record", "execute_code"],
]
return normal + adversarial
# ---------------------------------------------------------------------------
# Demo
# ---------------------------------------------------------------------------
if __name__ == "__main__":
monitor = SecurityMonitor()
print("=" * 70)
print("Generating synthetic production telemetry...")
print("=" * 70)
costs = generate_cost_series()
flags = generate_injection_flags()
baseline_sessions = generate_baseline_sessions(80)
test_sessions = generate_test_sessions()
cost_alerts = monitor.cost_anomalies(costs)
injection_alerts = monitor.injection_rate_alerts(flags)
transition_probs = monitor.fit_tool_call_baseline(baseline_sessions)
tool_alerts = monitor.tool_sequence_alerts(test_sessions, transition_probs)
dashboard = monitor.dashboard(
cost_alerts,
injection_alerts,
tool_alerts,
total_requests=len(costs),
total_sessions=len(test_sessions),
)
print(dashboard)
print("\n" + "=" * 70)
print("Verifying the Markov detector caught the two adversarial sessions")
print("=" * 70)
for i, seq in enumerate(test_sessions):
ll = monitor.sequence_log_likelihood(seq, transition_probs)
tag = "ADVERSARIAL" if i >= len(test_sessions) - 2 else "normal"
print(f" session {i} [{tag}]: ll={ll:.2f} seq={seq}")
Unlock the full lesson
You've read the first 2 sections. The rest of this lesson covers Check Yourself — plus a hands-on lab, quiz, and project artifact.
Create a free account — the opening phases of 29 of 35 courses are free, no credit card.
Browse all courses · View pricing · DeVenture Academy