Lab 14 · Prod cockpit
Instrument the Northwind support agent with metadata, cost and feedback, and print an on-call cockpit: error rate, p50/p95 latency, tokens, cost, tool health and thumbs-up rate.
What you will build
You are on call for the Northwind support agent, and right now all you can say about it is it seems fine. You build a small production cockpit: every ticket the agent answers is written to a JSONL run log with a metadata envelope, token counts, cost and tool outcomes, feedback is joined onto it, and one aggregator turns the log into the numbers you want on screen during an incident.
The report reads the local log by default. If you have a LangSmith key, the same aggregator reads live traces instead: one aggregator, two sources.
What you will learn
- How to design a metadata envelope you can group, filter and alert on
- How to pass
run_id,run_name,metadataandtagsin the invoke config - How to mask sensitive fields before they reach a tracing backend
- How to compute cost, error rate and nearest-rank p95 latency from run records
- How to join feedback scores onto runs through a shared
run_id
No LangSmith account needed. Without LANGSMITH_API_KEY the cockpit reports on a local file, run_log.jsonl. LangSmith is an optional extra at the end.
Check your prerequisites
You need three things on your machine and three values from your instructor.
- Python 3.11, 3.12 or 3.13. Check with the command below. On Windows, install it from python.org and tick Add python.exe to PATH.
- A code editor. VS Code with the Python extension works well.
- A terminal. PowerShell on Windows, Terminal on macOS or Linux.
python --versionpython3 --versionFrom your instructor you get an Azure OpenAI endpoint, an API key and a deployment name. Keep the key private: never paste it into code or share it in chat.
Create the project folder and virtual environment
Create a new folder for this lab, open it in a terminal, and create a virtual environment. A virtual environment is a private copy of Python for this project, so the libraries you install here never clash with other projects.
mkdir C:\labs\lab-14-prod-cockpit
cd C:\labs\lab-14-prod-cockpit
python -m venv .venv
.venv\Scripts\Activate.ps1mkdir -p ~/labs/lab-14-prod-cockpit
cd ~/labs/lab-14-prod-cockpit
python3 -m venv .venv
source .venv/bin/activateKeep lab folders on a short path such as C:\labs. Deep folders, for example inside OneDrive or Desktop, can exceed the Windows path limit. Some libraries then fail to load with The filename or extension is too long. OneDrive syncing can also lock files the labs write.
When activation works, your prompt starts with (.venv). Activate it again every time you open a new terminal for this lab. Open the folder in VS Code with code . if you like.
PowerShell says running scripts is disabled on this system? Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned, answer Y, and activate again.
This is what your folder will look like at the end of the lab:
Install the libraries
Create a file named requirements.txt in your lab folder with the content below. The versions are pinned to the ones this lab was tested with.
langchain==1.4.2
langchain-core==1.6.4
langchain-openai==1.6.2
python-dotenv==1.2.3
langsmith==0.14.0langsmith is the LangSmith SDK. LangChain already depends on it; it is pinned here because the script imports Client from it. You do not need a LangSmith account for this lab.
Then install everything into your virtual environment. It takes a minute or two.
python -m pip install --upgrade pip
pip install -r requirements.txtpython -m pip install --upgrade pip
pip install -r requirements.txtAdd your Azure OpenAI settings
Create a file named .env (the name starts with a dot) in your lab folder. Paste the template and replace the three placeholders with the values from your instructor.
# Azure OpenAI settings. Your instructor gives you these three values.
AZURE_OPENAI_ENDPOINT=https://<your-resource>.openai.azure.com/openai/v1
AZURE_OPENAI_API_KEY=<paste-your-key-here>
AZURE_OPENAI_DEPLOYMENT=<your-chat-deployment-name>
# Only used when your endpoint does NOT end in /openai/v1.
OPENAI_API_VERSION=2025-04-01-preview- AZURE_OPENAI_ENDPOINT is your resource URL. If it ends in
/openai/v1, keep it that way. If it is justhttps://<resource>.openai.azure.com/, paste that; the helper in the next step handles both. - AZURE_OPENAI_API_KEY is your secret key.
- AZURE_OPENAI_DEPLOYMENT is the deployment name, for example
gpt-5-mini. On Azure you call a deployment, not a model name.
Never commit .env to Git or share it. If you use Git, add a .gitignore file containing .env and .venv/.
Create the Azure OpenAI helper
Every lab talks to Azure OpenAI through one small helper file. Create model_setup.py and add the blocks below one after another. You write it once per lab folder; the lab code then only needs from model_setup import get_model.
Block 1Imports, .env loading and a console fix
load_dotenv() reads your .env file and puts every line into environment variables, so your key never appears in code.
The loop below it switches the console to UTF-8. Windows terminals default to an older encoding and crash when a model prints curly quotes or dashes.
"""Azure OpenAI model setup shared by every lab.
Import it in lab code as: from model_setup import get_model, get_embeddings
"""
import asyncio
import os
import sys
from dotenv import load_dotenv
from langchain_core.embeddings import Embeddings
from langchain_core.language_models import BaseChatModel
# Read AZURE_OPENAI_* values from the .env file next to your lab code.
load_dotenv()
# Windows consoles default to cp1252 and crash on the curly quotes models love to print.
for _stream in (sys.stdout, sys.stderr):
if hasattr(_stream, "reconfigure"):
_stream.reconfigure(encoding="utf-8", errors="replace")Block 2Read settings safely
_setting() fetches a value from the environment and stops with a friendly message if it is missing, instead of a long stack trace.
_uses_v1_api() checks the shape of your endpoint. Azure offers two:
https://<resource>.openai.azure.com/openai/v1is the newer OpenAI-compatible API.https://<resource>.openai.azure.com/is the classic API that needs anapi_version.
def _setting(name: str) -> str:
"""Return a required setting from .env, or stop with a readable message."""
value = os.getenv(name, "").strip()
if not value:
sys.exit(f"Missing {name} in your .env file. Open .env and fill it in.")
return value
def _uses_v1_api(endpoint: str) -> bool:
"""True for the OpenAI-compatible Azure endpoint that ends in /openai/v1."""
return endpoint.rstrip("/").endswith("/openai/v1")Block 3Build the chat model
get_model() returns a LangChain chat model connected to your deployment.
- With a
/openai/v1endpoint it usesChatOpenAIwithbase_url. On Azure the deployment name goes inmodel. - With a classic endpoint it uses
AzureChatOpenAIwithazure_deploymentandapi_version.
Either way the lab code just calls get_model() and never cares which one it got.
def get_model(deployment: str | None = None, **kwargs) -> BaseChatModel:
"""Return a chat model for your Azure OpenAI deployment."""
endpoint = _setting("AZURE_OPENAI_ENDPOINT")
api_key = _setting("AZURE_OPENAI_API_KEY")
name = deployment or _setting("AZURE_OPENAI_DEPLOYMENT")
if _uses_v1_api(endpoint):
from langchain_openai import ChatOpenAI
return ChatOpenAI(model=name, base_url=endpoint, api_key=api_key, **kwargs)
from langchain_openai import AzureChatOpenAI
return AzureChatOpenAI(
azure_deployment=name,
azure_endpoint=endpoint,
api_key=api_key,
api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
**kwargs,
)Block 4Build the embeddings model
get_embeddings() does the same for embeddings, which turn text into vectors for search. Only the retrieval labs use it, but keeping one helper for every lab is simpler.
def get_embeddings(deployment: str | None = None) -> Embeddings:
"""Return an embeddings model for your Azure OpenAI embeddings deployment."""
endpoint = _setting("AZURE_OPENAI_ENDPOINT")
api_key = _setting("AZURE_OPENAI_API_KEY")
name = deployment or _setting("AZURE_OPENAI_EMBEDDING_DEPLOYMENT")
if _uses_v1_api(endpoint):
from langchain_openai import OpenAIEmbeddings
return OpenAIEmbeddings(model=name, base_url=endpoint, api_key=api_key)
from langchain_openai import AzureOpenAIEmbeddings
return AzureOpenAIEmbeddings(
azure_deployment=name,
azure_endpoint=endpoint,
api_key=api_key,
api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
)Block 5A Windows fix for async code
use_stable_event_loop() matters only for the async MCP lab. The default Windows event loop can crash on exit after async network sessions close. On macOS and Linux it does nothing.
def use_stable_event_loop() -> None:
"""On Windows, use the Selector event loop so async MCP code exits cleanly."""
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())Check that the helper can reach your deployment with a one-line test:
python -c "from model_setup import get_model; print(get_model().invoke('Say hello in 5 words').text)"python -c "from model_setup import get_model; print(get_model().invoke('Say hello in 5 words').text)"You should see a short greeting. If you see an error instead, check the troubleshooting step at the end.
Download the lab files
The cockpit needs some traffic to report on before you have run anything. Download the sample run log and save it directly in your lab folder, next to where prod_cockpit.py will go (not in a subfolder).
Open it in your editor. Each line is one JSON object: a run record or a feedback record that points at a run through its run_id. That line shape is the contract the rest of the lab is built on.
Write the cockpit
An agent fails quietly: it picks the wrong tool, takes eleven seconds instead of four, or costs fifty times the median, and still returns HTTP 200. The cockpit makes those failures visible. Every ticket becomes one run record, and every number on the report is computed from those records.
Create an empty file named prod_cockpit.py in your lab folder. Then copy each block below into it, in order, one after another.
Block 1Imports
AIMessageandToolMessagelet you pick token usage and tool outcomes out of the finished conversation.Clientfromlangsmithis only used when you have a LangSmith key (feedback upload and the live report).get_modelcomes from your helper. Importingmodel_setupalso loads.env, so the next block can read your deployment name.
"""Lab 14: a production cockpit for the Northwind support agent.
Run: python prod_cockpit.py
"""
import json
import math
import os
import time
import uuid
from collections import defaultdict
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
from langchain.agents import create_agent
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool
from langsmith import Client
from model_setup import get_modelBlock 2The metadata contract
LOG_PATH puts run_log.jsonl next to the script, so it works from any working directory.
MODEL_NAME is your deployment name from .env. It goes into the metadata and picks the price row.
METADATA_KEYS lists the keys every run carries. Ask of each one: would I ever group a chart or filter an alert by this? If yes, it belongs in metadata. The two tuples below it are the only allowed values for tier and channel, and SENSITIVE_FIELDS names what must never be stored in a trace.
LOG_PATH = Path(__file__).resolve().parent / "run_log.jsonl"
# The model name used in metadata and for pricing: your Azure deployment name from .env.
MODEL_NAME = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-5-mini")
# The metadata contract: every run carries these keys, with string values. Adding a key is
# cheap; renaming one silently breaks every saved chart, filter and alert.
METADATA_KEYS = ("service", "environment", "release", "model", "customer_tier", "intent",
"channel", "thread_id")
TIERS = ("free", "pro", "enterprise")
CHANNELS = ("web", "email", "phone")
SENSITIVE_FIELDS = ("email", "phone", "card_number", "ssn")
REDACTED = "[redacted]"Block 3Build and validate the envelope
build_run_metadata returns exactly the contract keys, all as strings (tracing filters compare metadata values as strings, so "2" and 2 are different).
It raises ValueError for an unknown tier or channel. Validating at the producer is the cheapest place to catch a stray "Enterprise", which would otherwise show up as a second series on every chart next to "enterprise".
def build_run_metadata(customer_tier: str, intent: str, thread_id: str, model: str,
environment: str = "dev", channel: str = "web") -> dict[str, str]:
"""Build the metadata envelope for one run. Values are strings; vocabularies are checked."""
if customer_tier not in TIERS:
raise ValueError(f"customer_tier must be one of {TIERS}, got {customer_tier!r}")
if channel not in CHANNELS:
raise ValueError(f"channel must be one of {CHANNELS}, got {channel!r}")
if not thread_id:
raise ValueError("thread_id is required: it powers the LangSmith threads view")
metadata = {"service": "northwind-support", "environment": environment,
"release": "2026.09.1", "model": model, "customer_tier": customer_tier,
"intent": intent, "channel": channel, "thread_id": thread_id}
return {key: str(value) for key, value in metadata.items()}Block 4Mask sensitive fields
mask_payload walks dicts and lists recursively and replaces the value of any configured field, matching the key case-insensitively. It builds new objects instead of changing the input.
In LangSmith you pass a function like this to Client(hide_inputs=..., hide_outputs=...), so a bug here is a data-protection incident, not a chart glitch.
def mask_payload(payload: Any, fields: tuple[str, ...] = SENSITIVE_FIELDS) -> Any:
"""Replace the value of every configured field, at any depth, case-insensitively.
This is what you hand to Client(hide_inputs=...), so the original never leaves here.
"""
lowered = {field.lower() for field in fields}
if isinstance(payload, dict):
return {key: REDACTED if key.lower() in lowered else mask_payload(value, fields)
for key, value in payload.items()}
if isinstance(payload, list):
return [mask_payload(item, fields) for item in payload]
return payloadBlock 5Estimate cost
Prices are in USD per 1 million tokens, as input and output pairs. An unknown model falls back to a default price instead of reporting a free lunch.
If your deployment has a custom name, add a row for it to PRICE_PER_1M_USD with your real prices.
# USD per 1M tokens, (input, output). LangSmith keeps the real table server-side; this copy
# is what lets the offline report show a cost column.
PRICE_PER_1M_USD = {"gpt-5-mini": (0.25, 2.00), "gpt-5.2": (1.25, 10.00),
"claude-sonnet-4-6": (3.00, 15.00)}
DEFAULT_PRICE_PER_1M_USD = (1.00, 3.00)
def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float:
"""Estimate the USD cost of one run from its token counts and the price table."""
price_in, price_out = PRICE_PER_1M_USD.get(model, DEFAULT_PRICE_PER_1M_USD)
return round((input_tokens * price_in + output_tokens * price_out) / 1_000_000, 6)Block 6The agent's tools and prompt
Two tools, deliberately simple: lookup_order answers order questions from a dictionary, escalate_to_human hands off what the tools cannot solve. The prompt tells the model to look things up instead of guessing.
@tool
def lookup_order(order_id: str) -> str:
"""Look up an order's status, carrier and ETA by its id, such as A-100."""
orders = {"A-100": "shipped via DHL, arriving 2026-09-15, total 249.00 USD",
"A-101": "processing, expected 2026-09-18, total 89.50 USD"}
return f"Order {order_id.upper()}: {orders.get(order_id.upper(), 'not found')}."
@tool
def escalate_to_human(reason: str) -> str:
"""Hand the conversation to a human agent when the tools cannot resolve it."""
return f"Escalated ({reason}). A specialist will reply within two hours."
SYSTEM_PROMPT = (
"You are the Northwind support assistant. Look facts up with the tools instead of "
"guessing, escalate what they cannot answer, and reply in at most three sentences."
)Block 7The instrumented call
This is the heart of the lab. The invoke config carries four things:
run_id: a UUID you choose, so the trace and your own log line share one key.run_name: a stable name (support-ticket) to group runs by.metadata: the envelope from the previous block.tags: two flat labels for the quick-filter bar. Keep tags few and low-cardinality; metadata is the source of truth.
After the run, the code sums usage_metadata from every AIMessage for token counts, and reads ToolMessage.status ("success" or "error") for tool outcomes. The finished record is appended to the log.
def handle_ticket(question: str, customer_tier: str = "pro", intent: str = "order_status",
thread_id: str = "thread-1", model: Any = None,
model_name: str = MODEL_NAME, log_path: Path = LOG_PATH) -> dict[str, Any]:
"""Answer one ticket, fully instrumented, and append its record to the run log."""
agent = create_agent(model=model or get_model(), name="northwind-support",
tools=[lookup_order, escalate_to_human], system_prompt=SYSTEM_PROMPT)
metadata = build_run_metadata(customer_tier, intent, thread_id, model_name)
# The run_id is ours, so it names the LangSmith root run *and* our own log line below.
run_id = uuid.uuid4()
started = time.perf_counter()
result = agent.invoke(
{"messages": [{"role": "user", "content": question}]},
config={"run_id": run_id, "run_name": "support-ticket", "metadata": metadata,
"tags": [f"env:{metadata['environment']}", f"tier:{customer_tier}"]},
)
latency_ms = round((time.perf_counter() - started) * 1000, 1)
messages = result["messages"]
usage = [m.usage_metadata for m in messages if isinstance(m, AIMessage) and m.usage_metadata]
tokens = {"input": sum(u["input_tokens"] for u in usage),
"output": sum(u["output_tokens"] for u in usage)}
record = {
"type": "run", "run_id": str(run_id), "error": None, "metadata": metadata,
"ts": datetime.now(UTC).isoformat(timespec="seconds"), "latency_ms": latency_ms,
"tokens": tokens, "answer": messages[-1].text,
"cost_usd": estimate_cost_usd(model_name, tokens["input"], tokens["output"]),
"tools": [{"name": m.name, "ok": m.status != "error"}
for m in messages if isinstance(m, ToolMessage)],
}
append_log(record, log_path)
return recordBlock 8Write and read the run log
append_log writes one JSON object per line. record_feedback appends a feedback line that points at a run, and also sends it to LangSmith when a key is set.
records_from_log joins them back: every run gets a feedback dict built from the feedback lines with its run_id. It skips blank lines and feedback whose run is not in the file, so a hand-edited or truncated log does not crash the report.
def append_log(entry: dict[str, Any], log_path: Path = LOG_PATH) -> None:
"""Append one JSON object as one line. Append-only logs are easy to ship and replay."""
with Path(log_path).open("a", encoding="utf-8") as handle:
handle.write(json.dumps(entry) + "\n")
def record_feedback(run_id: str, key: str, score: float, log_path: Path = LOG_PATH) -> None:
"""Write one feedback score to the run log, and to LangSmith when a key is configured."""
append_log({"type": "feedback", "run_id": str(run_id), "key": key, "score": float(score),
"ts": datetime.now(UTC).isoformat(timespec="seconds")}, log_path)
if os.environ.get("LANGSMITH_API_KEY"):
Client().create_feedback(run_id, key=key, score=score)
def records_from_log(log_path: Path = LOG_PATH) -> list[dict[str, Any]]:
"""Read the run log, joining each run with the feedback scores written against it."""
entries = [json.loads(line) for line in Path(log_path).read_text("utf-8").splitlines()
if line.strip()]
runs = {e["run_id"]: {**e, "feedback": {}} for e in entries if e.get("type") == "run"}
for entry in entries:
run = runs.get(entry.get("run_id", ""))
if entry.get("type") == "feedback" and run is not None: # skip orphan feedback
run["feedback"][entry["key"]] = entry["score"]
return list(runs.values())Block 9The same records from LangSmith (optional)
This function produces the same record shape from live traces: root runs for latency, tokens, cost and feedback, and run_type="tool" runs for the tool breakdown. Because the shape matches, the aggregator below does not care where the records came from.
It only runs when LANGSMITH_API_KEY is set. You can type it in now and try it later.
def records_from_langsmith(project: str, hours: int = 24) -> list[dict[str, Any]]:
"""The same record shape, read from live traces. One aggregator, two sources."""
client = Client()
since = datetime.now(UTC) - timedelta(hours=hours)
tools_by_trace: dict[str, list[dict[str, Any]]] = defaultdict(list)
for run in client.list_runs(project_name=project, run_type="tool", start_time=since):
tools_by_trace[str(run.trace_id)].append({"name": run.name, "ok": not run.error})
records = []
for run in client.list_runs(project_name=project, is_root=True, start_time=since, limit=100):
end = run.end_time or run.start_time
records.append({
"run_id": str(run.id), "ts": run.start_time.isoformat(timespec="seconds"),
"latency_ms": round((end - run.start_time).total_seconds() * 1000, 1),
"error": run.error, "metadata": (run.extra or {}).get("metadata", {}),
"cost_usd": float(run.total_cost or 0),
"tokens": {"input": run.prompt_tokens or 0, "output": run.completion_tokens or 0},
"tools": tools_by_trace[str(run.trace_id)],
"feedback": {key: stats["avg"] for key, stats in (run.feedback_stats or {}).items()},
})
return recordsBlock 10Percentiles and the aggregator
percentile uses the nearest-rank method: for ten values, p50 is the 5th and p95 is the 10th. It returns 0.0 for an empty list, because a report that crashes when traffic drops to zero is useless at 03:00.
summarize_runs counts runs and errors, computes p50/p95 latency, totals tokens and cost, builds a per-tool breakdown with error rates, and averages each feedback key. Every ratio checks for division by zero.
def percentile(values: list[float], pct: float) -> float:
"""Nearest-rank percentile: p95 is the 95th slowest request out of a hundred, exactly."""
if not values: # a report that crashes when traffic drops to zero is useless at 03:00
return 0.0
ordered = sorted(values)
return ordered[math.ceil(pct / 100 * len(ordered)) - 1]
def summarize_runs(records: list[dict[str, Any]]) -> dict[str, Any]:
"""Aggregate run records into the numbers the cockpit shows."""
runs = len(records)
latencies = [record["latency_ms"] for record in records]
errors = [record for record in records if record["error"]]
cost = sum(record["cost_usd"] for record in records)
tools: dict[str, dict[str, int]] = defaultdict(lambda: {"calls": 0, "errors": 0})
scores: dict[str, list[float]] = defaultdict(list)
for record in records:
for event in record["tools"]:
tools[event["name"]]["calls"] += 1
tools[event["name"]]["errors"] += 0 if event["ok"] else 1
for key, score in record["feedback"].items():
scores[key].append(score)
return {
"runs": runs, "errors": len(errors), "cost_usd": round(cost, 6),
"error_rate": round(len(errors) / runs, 4) if runs else 0.0,
"cost_per_run_usd": round(cost / runs, 6) if runs else 0.0,
"latency_p50_ms": round(percentile(latencies, 50), 1),
"latency_p95_ms": round(percentile(latencies, 95), 1),
"tokens": {"input": sum(r["tokens"]["input"] for r in records),
"output": sum(r["tokens"]["output"] for r in records)},
"tools": {name: {**counts, "error_rate": round(counts["errors"] / counts["calls"], 4)}
for name, counts in sorted(tools.items())},
"feedback": {key: round(sum(values) / len(values), 4)
for key, values in sorted(scores.items())},
}Block 11Print the cockpit
A plain text report with one line per number. In production the same numbers would sit on a dashboard and drive alerts.
def print_report(summary: dict[str, Any]) -> None:
"""Print the cockpit: the numbers you want on screen during an incident."""
print(f"runs {summary['runs']} error rate {summary['error_rate']:.1%}")
print(f"latency p50 {summary['latency_p50_ms']:.0f} ms p95 {summary['latency_p95_ms']:.0f} ms")
print(f"tokens {summary['tokens']['input']} in / {summary['tokens']['output']} out")
print(f"cost {summary['cost_usd']:.6f} USD ({summary['cost_per_run_usd']:.6f} per run)")
for name, stats in summary["tools"].items():
print(f"tool {name:<20}{stats['calls']:>3} calls {stats['error_rate']:.0%} errors")
for key, mean in summary["feedback"].items():
print(f"feedback {key:<20}mean {mean}")Block 12Answer a ticket and report
The demo first shows the masking helper on a nested payload. Then it answers one real ticket, prints the record it logged, gives it a thumbs-up, and prints the cockpit over the whole log: the seven sample runs plus yours.
With LANGSMITH_API_KEY set it flushes the tracer and reports on your LangSmith project instead.
if __name__ == "__main__":
print("masking demo:", mask_payload({"order_id": "A-100", "customer": {"Email": "ada@example.com"}}))
record = handle_ticket("Where is order A-100?", customer_tier="enterprise", thread_id="t-7")
print(f"\nanswer {record['answer']}")
print(f"run_id {record['run_id']}")
print(f"tools {[t['name'] for t in record['tools']] or 'none'}")
print(f"latency {record['latency_ms']:.0f} ms tokens {record['tokens']} cost {record['cost_usd']:.6f} USD")
print(f"metadata tier={record['metadata']['customer_tier']} model={record['metadata']['model']} "
f"intent={record['metadata']['intent']}")
record_feedback(record["run_id"], "thumbs", 1.0)
if os.environ.get("LANGSMITH_API_KEY"):
from langchain_core.tracers.langchain import wait_for_all_tracers
wait_for_all_tracers() # traces upload in the background; flush before reading back
project = os.environ.get("LANGSMITH_PROJECT", "default")
print(f"\nLANGSMITH_API_KEY found: reporting on project {project!r} (last 24 h).\n")
summary = summarize_runs(records_from_langsmith(project))
else:
print(f"\nNo LANGSMITH_API_KEY: reporting on the local run log, {LOG_PATH.name}.\n")
summary = summarize_runs(records_from_log())
print_report(summary)Run the cockpit
Make sure run_log.jsonl is in the lab folder, then run the script.
python prod_cockpit.pypython3 prod_cockpit.pyYour output should look similar to this. Model answers vary a little from run to run, so the wording will not match exactly.
masking demo: {'order_id': 'A-100', 'customer': {'Email': '[redacted]'}}
answer Order A-100 was shipped via DHL and is scheduled to arrive on 2026-09-15. Would you like the tracking number or any additional details?
run_id c7fd3a0b-263f-4d3e-a6ec-129fcf4385d4
tools ['lookup_order']
latency 5608 ms tokens {'input': 482, 'output': 324} cost 0.000768 USD
metadata tier=enterprise model=gpt-5-mini intent=order_status
No LANGSMITH_API_KEY: reporting on the local run log, run_log.jsonl.
runs 8 error rate 12.5%
latency p50 2110 ms p95 9420 ms
tokens 3066 in / 687 out
cost 0.002139 USD (0.000267 per run)
tool escalate_to_human 1 calls 0% errors
tool lookup_order 6 calls 17% errors
feedback thumbs mean 0.7143Read the report the way you would during an incident:
- runs 8, error rate 12.5 %: the seven sample runs plus yours, and one of them timed out.
- p95 9420 ms: the slowest request dominates p95 with so few runs. That timeout alone breaks an 8 s latency SLO.
- lookup_order ... errors: one of the sample lookups failed. That is where a flaky tool shows up first.
- feedback thumbs mean: the share of thumbs-up across rated runs, including the one you just added.
Your ticket's line shows real token counts and a cost of a fraction of a cent. The sample runs came with their own numbers.
Each run appends to run_log.jsonl. Run the script again and the run count goes up by one. To start over, download the sample log again.
Check your work
You are done when
- The masking demo prints
[redacted]in place of the email address. - Your ticket's answer mentions DHL and 2026-09-15, and the tools line shows
lookup_order. - The report shows 8 runs on the first run, an error rate above zero, and p95 at or above p50.
- Running the script a second time increases the run count by one.
Troubleshooting
FileNotFoundError: run_log.jsonl
The sample log must sit in the same folder as prod_cockpit.py. Download it again from the files step.
ImportError: cannot import name 'UTC' from 'datetime'
datetime.UTC needs Python 3.11 or newer. Recreate the virtual environment with Python 3.11+.
The cost for my run shows the default price
The price table is keyed by model name, and the script uses your deployment name from .env. If it has a custom name, add a row for it to PRICE_PER_1M_USD.
With a LangSmith key the report shows 0 runs
Traces need LANGSMITH_TRACING=true as well as the key, and they can take a few seconds to be queryable. Run again after a moment, and check LANGSMITH_PROJECT names the project you traced into.
401 Unauthorized or 'Access denied due to invalid subscription key'
The key in .env is wrong or has spaces around it. Copy it again from your instructor, with no quotes.
404 'DeploymentNotFound' or 'The API deployment for this resource does not exist'
AZURE_OPENAI_DEPLOYMENT must be the deployment name exactly as your instructor gave it, and the endpoint must be the same resource.
429 Too Many Requests
The shared deployment is busy. Wait 30 seconds and run again.
ModuleNotFoundError: No module named 'model_setup' (or another package)
Run the command from inside your lab folder, with the virtual environment activated. Reinstall with pip install -r requirements.txt if a library is missing.
'python' is not recognized, or the wrong Python runs
Activate the virtual environment first. On Windows you can also try py -3.11 instead of python when creating the venv.
Go further
- SLO grades. Compare the summary with targets (p95 under 8000 ms, error rate under 2 %) and print
OKorBREACHper objective. The sample log breaches at least one. - Escalation rate. Add the share of runs that called
escalate_to_humanto the summary. It is an early warning: frustrated customers escalate before they rate anything. - Per-tier latency. Group p95 latency by
metadata.customer_tierto see whether enterprise customers suffer more. - LangSmith. Create a free LangSmith account, add
LANGSMITH_TRACING=true,LANGSMITH_API_KEY=...andLANGSMITH_PROJECT=northwind-support-devto.env, run a few times, and the same report reads live traces. Then build a dashboard chart of p95 latency grouped bycustomer_tier. - Alerts and online evaluators. In LangSmith, add an error-rate alert (>= 2 % over 15 minutes) and an LLM-as-judge online evaluator at 20 % sampling, with a spend limit set before you launch.
Stuck?
Download the finished project and compare it with yours. It contains every file from this guide except .env and .venv, so you still add your own settings.