How to Achieve Full‑Stack AI Observability: Tracking Prompts, Tool Calls, Traces, and Tokens

Why AI Observability Matters

When an AI product is launched, users often report inconsistent behavior – sometimes it works well, sometimes it does not. Operators cannot diagnose the problem because they lack visibility into the Prompt, the external tools the agent calls, the full request trace, and the token cost. Unlike traditional software that has mature APM, logging, and tracing, LLM‑based systems have unique challenges such as nondeterministic outputs, multi‑step workflows, and opaque pricing.

What Is AI Observability?

AI observability is the ability to understand the internal state and behavior of an AI system end‑to‑end. It extends traditional APM from HTTP request → DB query → response to user input → Prompt → LLM call → Tool Call → multi‑turn reasoning → final output . Four core dimensions are required; missing any dimension leaves the system a costly black box.

Four Core Dimensions

Prompt Tracking : record every prompt, including system prompts, few‑shot examples, and the final prompt sent to the model.

Tool Call Tracking : monitor every external API, database, or code execution the agent performs.

Trace (Chain) Tracking : capture the complete call graph from the initial request to the final response.

Token Tracking : measure input/output token counts and translate them into real monetary cost.

Dimension 1 – Prompt Tracking

A typical request may generate several prompts (intent detection, context retrieval, generation, quality check). Without prompt logs you cannot pinpoint which step produced a bad output.

Core Prompt Record Schema

prompt_trace:
  id: "prompt_abc123"               # unique ID
  timestamp: "2026-06-18T10:23:45Z"
  parent_id: "trace_xyz789"          # owning trace
  system_prompt: "You are a professional email assistant..."
  user_prompt: "User original input"
  full_prompt: "Complete prompt sent to LLM"
  model: "gpt-5.5"
  temperature: 0.7
  max_tokens: 2000
  response: "LLM full output"
  finish_reason: "stop"
  latency_ms: 1234
  tokens_input: 850
  tokens_output: 420
  metadata:
    user_id: "user_12345"
    session_id: "session_67890"
    feature_flag: "new_email_template_v2"

Practical Example with Langfuse

from langfuse import Langfuse
from openai import OpenAI

langfuse = Langfuse(public_key="pk-lf-...", secret_key="sk-lf-...", host="https://cloud.langfuse.com")
openai = OpenAI()

def generate_email(user_input: str, user_id: str):
    trace = langfuse.trace(name="email_generation", user_id=user_id, metadata={"source": "web"})
    # Prompt 1 – intent detection
    span1 = trace.span(name="intent_detection")
    intent_prompt = f"Determine user intent: {user_input}"
    response1 = openai.chat.completions.create(model="gpt-5.5", messages=[{"role": "user", "content": intent_prompt}])
    span1.end(output=response1.choices[0].message.content,
              metadata={"tokens_input": response1.usage.prompt_tokens,
                        "tokens_output": response1.usage.completion_tokens})
    intent = response1.choices[0].message.content
    # Prompt 2 – email drafting
    span2 = trace.span(name="email_drafting")
    email_prompt = f"Generate an email based on intent '{intent}'"
    response2 = openai.chat.completions.create(model="gpt-5.5", messages=[{"role": "user", "content": email_prompt}])
    span2.end(output=response2.choices[0].message.content,
              metadata={"tokens_input": response2.usage.prompt_tokens,
                        "tokens_output": response2.usage.completion_tokens})
    email = response2.choices[0].message.content
    trace.update(output=email, metadata={"total_spans": 2})
    return email

Prompt‑Tracking Best Practices

Record the full prompt – include system prompt, few‑shot examples, and the assembled prompt sent to the LLM.

Version management – tag each prompt template with a version number.

A/B test logging – store experiment group identifiers in the metadata.

Sensitive‑data redaction – mask personal information before persisting prompts.

Dimension 2 – Tool Call Tracking

Modern agents may invoke 5‑20 external tools per request (web search, database query, code execution, browser automation). Without tool‑call logs you cannot know what the agent actually did.

Core Tool‑Call Record Schema

tool_call_trace:
  id: "tool_call_def456"
  timestamp: "2026-06-18T10:23:46Z"
  parent_id: "span_abc123"
  tool_name: "web_search"
  tool_version: "2.1.0"
  input:
    query: "2026 AI market size"
    num_results: 5
    date_range: "last_month"
  output:
    success: true
    results: [{"title": "...", "url": "...", "snippet": "..."}]
    result_count: 5
  latency_ms: 890
  retries: 0
  error: null
  metadata:
    cache_hit: false
    rate_limited: false

Decorator for Automatic Tool‑Call Tracing

from langfuse import Langfuse
from typing import Callable, Any

langfuse = Langfuse()

def tracked_tool_call(trace, tool_name: str, func: Callable, **kwargs):
    """Decorator that creates a span, runs the tool, and records success/failure."""
    span = trace.span(name=f"tool_call:{tool_name}", input=kwargs)
    try:
        result = func(**kwargs)
        span.end(output=result, metadata={"success": True})
        return result
    except Exception as e:
        span.end(output=None, metadata={"success": False,
                                      "error_type": type(e).__name__,
                                      "error_message": str(e)})
        raise

Key Tool‑Call Metrics

Success rate > 99 %

Latency distribution (P50, P95, P99) to spot slow tools

Error‑type breakdown to identify fragile integrations

Call frequency to prioritize optimization

Common Pitfalls (and Solutions)

Tool timeout – Agent hangs; fix with timeout + retry.

Tool hallucination – Agent calls a non‑existent tool; fix with a strict tool registry.

Tool overuse – Simple tasks call many tools; fix by optimizing tool‑selection logic.

Tool dependency failures – One tool’s error cascades; fix by recording dependency graph and handling cascade errors.

Dimension 3 – Trace (Chain) Tracking

A Trace is the complete call graph for a single user request, analogous to an HTTP request trace in micro‑service APM.

Hierarchical Example

Trace: User request "Help me analyze this CSV file"
├── Span 1: Intent detection (LLM call, 234 ms, 850 tokens)
├── Span 2: File reading (Tool call, 45 ms, output 1000 rows)
├── Span 3: Data analysis
│   ├── LLM call (1.2 s, 2100 tokens)
│   └── Tool call: python_executor (890 ms)
├── Span 4: Report generation (LLM call, 1.8 s, 3200 tokens)
└── Span 5: Result formatting (LLM call, 456 ms)
Total: 4 LLM calls, 2 tool calls, 4.6 s latency, 7350 tokens

Why Traces Matter

Problem localisation – Slow response can be traced to a specific span.

Performance optimisation – Identify the longest‑running spans.

Cost analysis – Attribute token cost to each span.

Quality assessment – Compare successful vs. failed traces.

Langfuse Trace Construction Example

from langfuse import Langfuse
import time

langfuse = Langfuse()

def analyze_csv(user_input: str, file_path: str, user_id: str):
    trace = langfuse.trace(name="csv_analysis", user_id=user_id,
                          metadata={"input": user_input, "file_path": file_path})
    start_time = time.time()
    # Span 1 – intent detection
    with trace.span(name="intent_detection") as span1:
        intent = detect_intent(user_input)
        span1.update(output=intent)
    # Span 2 – file reading
    with trace.span(name="file_reading") as span2:
        csv_data = read_csv_file(file_path)
        span2.update(output={"rows": len(csv_data), "columns": len(csv_data.columns)},
                     metadata={"file_size_mb": get_file_size(file_path)})
    # Span 3 – data analysis
    with trace.span(name="data_analysis") as span3:
        analysis_prompt = f"Analyze the following data:
{csv_data.head()}"
        analysis = call_llm(analysis_prompt)
        # Sub‑span – code execution
        with span3.span(name="code_execution") as sub_span:
            code = extract_code(analysis)
            result = execute_python(code)
            sub_span.update(output=result)
        span3.update(output=analysis)
    # Span 4 – report generation
    with trace.span(name="report_generation") as span4:
        report = generate_report(analysis, result)
        span4.update(output=report)
    # Span 5 – formatting
    with trace.span(name="formatting") as span5:
        final_output = format_response(report)
        span5.update(output=final_output)
    # Final trace update
    total_time = time.time() - start_time
    trace.update(output=final_output,
                 metadata={"total_latency_ms": int(total_time * 1000),
                           "total_spans": 5,
                           "total_llm_calls": 4})
    return final_output

Advanced Trace Analysis

1. Comparative analysis – fetch successful and failed traces, compare average LLM call counts.

successful_traces = langfuse.fetch_traces(name="csv_analysis", metadata={"status": "success"}, limit=100)
failed_traces = langfuse.fetch_traces(name="csv_analysis", metadata={"status": "failed"}, limit=100)
# Observation: success avg 3 LLM calls, failure avg 7 LLM calls (likely retries)

2. Performance bottleneck detection – compute average and P95 latency per span.

traces = langfuse.fetch_traces(name="csv_analysis", limit=1000)
span_latencies = {}
for trace in traces:
    for span in trace.spans:
        span_latencies.setdefault(span.name, []).append(span.latency_ms)
for name, lat in span_latencies.items():
    avg = sum(lat) / len(lat)
    p95 = sorted(lat)[int(len(lat) * 0.95)]
    print(f"{name}: avg={avg}ms, p95={p95}ms")

3. Cost attribution – sum token usage per span and convert to dollars.

trace = langfuse.fetch_trace(trace_id="xxx")
total_cost = 0
for span in trace.spans:
    if span.type == "llm":
        tokens = span.metadata.get("tokens_input", 0) + span.metadata.get("tokens_output", 0)
        cost = calculate_cost(span.model, tokens)
        total_cost += cost
        print(f"{span.name}: {tokens} tokens, ${cost:.4f}")
print(f"Total cost: ${total_cost:.4f}")

Dimension 4 – Token Tracking

LLM pricing in 2026 varies widely. Example prices (per 1 M tokens):

GPT‑5.5 – $15 input, $60 output

Claude Opus 4 – $15 input, $75 output

Gemini 2.5 Ultra – $12.5 input, $50 output

DeepSeek‑V4 – $0.55 input, $2.19 output

A complex agent task can consume >10 000 tokens, costing $0.5‑$1. With 10 000 daily requests, low‑cost models cost $50‑$100 per day, while high‑cost models can exceed $5 000 per day. Without token tracking you only discover the bill at month‑end.

Core Token Metrics

tokens_per_minute, cost_per_minute (real‑time monitoring)

tokens_by_user / tokens_by_feature (granular cost allocation)

daily_tokens trend, budget vs. spent, projected end‑of‑month cost

Langfuse Token Tracker Implementation

from langfuse import Langfuse
from dataclasses import dataclass
from typing import Dict
import json

langfuse = Langfuse()

@dataclass
class TokenUsage:
    model: str
    input_tokens: int
    output_tokens: int
    cost: float

class TokenTracker:
    def __init__(self):
        self.usage_by_user: Dict[str, TokenUsage] = {}
        self.usage_by_feature: Dict[str, TokenUsage] = {}
        self.daily_usage: Dict[str, TokenUsage] = {}

    def record_usage(self, user_id: str, feature: str, model: str, input_tokens: int, output_tokens: int):
        cost = self._calculate_cost(model, input_tokens, output_tokens)
        usage = TokenUsage(model, input_tokens, output_tokens, cost)
        # aggregate by user
        if user_id not in self.usage_by_user:
            self.usage_by_user[user_id] = TokenUsage(model, 0, 0, 0)
        self.usage_by_user[user_id].input_tokens += input_tokens
        self.usage_by_user[user_id].output_tokens += output_tokens
        self.usage_by_user[user_id].cost += cost
        # aggregate by feature
        if feature not in self.usage_by_feature:
            self.usage_by_feature[feature] = TokenUsage(model, 0, 0, 0)
        self.usage_by_feature[feature].input_tokens += input_tokens
        self.usage_by_feature[feature].output_tokens += output_tokens
        self.usage_by_feature[feature].cost += cost
        # daily aggregation
        today = datetime.now().strftime("%Y-%m-%d")
        if today not in self.daily_usage:
            self.daily_usage[today] = TokenUsage(model, 0, 0, 0)
        self.daily_usage[today].input_tokens += input_tokens
        self.daily_usage[today].output_tokens += output_tokens
        self.daily_usage[today].cost += cost
        # send to Langfuse for visualization
        langfuse.score(name="token_cost", value=cost,
                       metadata={"user_id": user_id, "feature": feature, "model": model})

    def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        pricing = {
            "gpt-5.5": {"input": 15, "output": 60},
            "claude-opus-4": {"input": 15, "output": 75},
            "gemini-2.5-ultra": {"input": 12.5, "output": 50},
            "deepseek-v4": {"input": 0.55, "output": 2.19},
        }
        if model not in pricing:
            return 0
        input_cost = (input_tokens / 1_000_000) * pricing[model]["input"]
        output_cost = (output_tokens / 1_000_000) * pricing[model]["output"]
        return input_cost + output_cost

    def get_daily_report(self):
        today = datetime.now().strftime("%Y-%m-%d")
        usage = self.daily_usage.get(today, TokenUsage("", 0, 0, 0))
        return {
            "date": today,
            "total_tokens": usage.input_tokens + usage.output_tokens,
            "total_cost": usage.cost,
            "by_feature": {f: {"tokens": u.input_tokens + u.output_tokens, "cost": u.cost}
                           for f, u in self.usage_by_feature.items()},
        }

Token Optimisation Strategies

Model selection – use cheaper models for low‑complexity tasks.

def select_model(task_complexity: str) -> str:
    if task_complexity == "high":
        return "gpt-5.5"   # $15/$60
    elif task_complexity == "medium":
        return "gemini-2.5-flash"   # $0.30/$1.20
    else:
        return "deepseek-v4"   # $0.55/$2.19

Prompt caching – reuse static system prompts to avoid re‑tokenising.

# Before: each request sends the full system prompt (≈2000 tokens)
# After: cache the system prompt once and reuse it, halving token cost.

Output length control – set max_tokens according to task needs.

# Before: no limit, may emit 2000 tokens
response = call_llm(prompt)  # up to 2000 tokens
# After: enforce a ceiling
response = call_llm(prompt, max_tokens=500)  # max 500 tokens

Full AI Observability Architecture

The recommended stack combines an open‑source observability platform (Langfuse) with optional OpenTelemetry extensions for deeper customisation. Commercial alternatives (LangSmith, Helicone) are listed for reference but the core concepts remain the same.

Open‑source stack : Langfuse (cloud‑hosted or self‑hosted) + PostgreSQL (metadata) + ClickHouse (time‑series) + Redis (caching) + optional Grafana dashboards.

Custom stack : OpenTelemetry SDKs for LLMs + self‑built storage/visualisation.

Commercial options : LangSmith (deep LangChain integration) and Helicone (LLM‑focused monitoring).

Production Deployment Checklist

Infrastructure: Langfuse v2+, PostgreSQL 15+, ClickHouse 23+, Redis 7+, optional Grafana.

Integration points: OpenAI/Anthropic providers, LangChain/LlamaIndex agents, tool registry, alerting (Slack/PagerDuty/Email).

Alerting rules: high latency (P95 > 5 s), error rate > 5 %, cost > 80 % of budget, token spikes > 10 k per request.

Data retention: traces 90 days, metrics 1 year, logs 30 days.

Real‑World Refactor Case Study

An e‑commerce customer‑service AI suffered from occasional slow or incorrect answers. Before the refactor it lacked all four observability dimensions.

After integrating Langfuse and adding the four dimensions, key metrics improved dramatically:

Problem‑localisation time reduced from 2‑4 hours to 5‑10 minutes (95 % decrease).

P95 latency dropped from 8.5 s to 3.2 s (62 % decrease).

Daily cost fell from $450 to $280 (38 % decrease).

User satisfaction rose from 3.2/5 to 4.3/5 (34 % increase).

Key findings:

Slow responses were caused by a tool call to the order‑system API (P95 = 4.2 s); optimisation reduced it to 0.8 s.

Incorrect answers stemmed from outdated few‑shot examples in the Prompt; updating them raised accuracy by 25 %.

30 % of traffic could be handled by a cheaper model, cutting cost by 40 % after migration.

10‑Rule Best‑Practice Guide for AI Observability

Integrate observability from day 1; treat it as a development requirement.

Track everything but prioritise high‑cost, high‑latency, and high‑error paths.

Structure traces with clear naming (e.g., trace_name: feature_action, span_name: step_type).

Record full context for each span: inputs, outputs, model parameters, token usage, latency, errors, and business metadata.

Set sensible alerts (latency P95 > 5 s, error rate > 5 %, cost > 80 % of budget, token spikes > 10 k).

Review traces weekly – focus on the slowest, most expensive, and failed traces.

Maintain cost awareness: surface per‑feature, per‑user, and per‑call costs.

Continuously optimise prompts (trim unnecessary text, update few‑shot examples, tune temperature).

Version‑control all artefacts – prompt templates, model versions, tool versions, and code.

Ensure security and privacy – redact PII, enforce access controls, and respect data‑retention policies.

Conclusion

AI observability is no longer optional; it is essential for reliable, performant, and cost‑effective AI services. By instrumenting Prompt, Tool Call, Trace, and Token dimensions, teams can move from guessing to data‑driven debugging, optimisation, and governance.

References

Langfuse documentation – https://langfuse.com/docs

LangSmith documentation – https://docs.smith.langchain.com

OpenTelemetry for GenAI – https://opentelemetry.io/blog/2024/genai

AI Trust OS: Continuous Governance for Autonomous AI – https://arxiv.org/abs/2604.04749 (2026.04)

Governance‑Aware Agent Telemetry – https://arxiv.org/abs/2604.05119 (2026.04)

From Confident Closing to Silent Failure – https://arxiv.org/abs/2606.09863 (2026.06)