chiefj/instrumentation-sdk-api repository overview

⁠LLM Observability Platform: Core Python Infrastructure

This guide covers the technical architecture and end-user usage for the Python-based observability components.

⁠Table of Contents

• 1. System Architecture⁠
  • High-Level Data Flow⁠
  • Technical Sequence⁠
• 2. End-User Usage Guide⁠
  • Installation⁠
  • Auto-Instrumentation (Zero-Code Changes)⁠
  • Remote Management API (REST)⁠
  • Basic Usage: Decorators⁠
  • Advanced Usage: Context Manager⁠
  • Token Counting (Pre-Call Token Counting)⁠
  • Streaming Observability (TTFT & Token Tracking)⁠
  • PII & Injection Scan (Aho-Corasick Redaction)⁠
  • Deterministic Sampling Gate (Modulo 100)⁠
  • MiniLM Embedding (Concurrent & Sampled)⁠
  • Prometheus Metrics & Grafana Dashboard⁠
  • Updating Config Files (Model Prices, PII Patterns, Infra)⁠
  • Docker Deployment⁠
  • Observability Launcher CLI (llm-observe)⁠
• 3. Implementation Call Chain⁠

⁠1. System Architecture

⁠High-Level Data Flow

This diagram illustrates the lifecycle of a span from application capture to background enrichment. The SDK now includes a REST Management API for remote control and discovery.

┌────────────────┐          ┌──────────────────┐          ┌───────────────────┐
│   User App     │ capture  │ instrumentation  │  queue   │  Cloudflare Queue │
│  (Python/JS)   ├─────────>│      -sdk        ├─────────>│ (span-enrichment) │
└────────────────┘          └─────────┬────────┘          └─────────┬─────────┘
                                      │                             │
                                      │ REST API (8000)             │ trigger
                                      v                             v
┌────────────────┐          ┌──────────────────┐          ┌───────────────────┐
│ Analytics DB   │ storage  │  Remote Control  │ response │  queue-embedding  │
│ (ClickHouse)   │<─────────┤  (Init/Detect)   │<─────────┤      -worker      │
└────────────────┘          └──────────────────┘          └─────────┬─────────┘
                                                                    │
                                                                    │ HTTP call
                                                                    v
                                                          ┌───────────────────┐
                                                          │ Cloudflare AI     │
                                                          │ (Workers AI API)  │
                                                          └───────────────────┘

⁠2. End-User Usage Guide

The instrumentation-sdk is designed to be developer-friendly, requiring minimal code changes to start capturing observability data.

⁠Installation

pip install instrumentation-sdk

⁠Auto-Instrumentation (Zero-Code Changes)

The fastest way to get observability is to use auto-instrumentation. This patches the underlying HTTP calls of popular LLM clients transparently.

from instrumentation_sdk import init_auto_instrumentation

# Initialize at the start of your application
init_auto_instrumentation()

# Now any call to OpenAI, Anthropic, LiteLLM, or LangChain is tracked automatically
import openai
client = openai.AsyncOpenAI()
response = await client.chat.completions.create(model="gpt-4o", messages=[...])

Supported Providers:

• OpenAI: openai.AsyncOpenAI
• Anthropic: anthropic.AsyncAnthropic
• LiteLLM: litellm.acompletion
• LangChain: Any model inheriting from BaseChatModel (via ainvoke)

⁠Remote Management API (REST)

The SDK provides a built-in FastAPI-based management layer for remote orchestration.

⁠Basic Usage: Decorators

Use the @llm_observe decorator to manually track functions.

from instrumentation_sdk import llm_observe

# (1) Decorate your LLM-calling functions
@llm_observe(service="payment-bot", endpoint="gpt-4o")
def get_llm_response(prompt: str):
    # Your existing LLM logic here
    # status, latency, and span_ids are captured automatically
    return response

# (2) Support for Async functions
@llm_observe(service="search-agent", endpoint="claude-3")
async def get_async_response(prompt: str):
    return await client.completions.create(...)

⁠Advanced Usage: Context Manager

For callers who need to set metadata mid-call (e.g., after routing to a specific model or determining usage), use the llm_span context manager. It supports both synchronous and asynchronous usage.

from instrumentation_sdk import llm_span

async def my_handler(req):
    # (1) Start a span with initial metadata
    async with llm_span(model="gpt-4o", user_id=req.user_id) as span:
        # (2) Perform your LLM call
        response = await client.chat.completions.create(...)
        
        # (3) Update metadata mid-call
        span.set_metadata("actual_model", response.model)
        span.set_metadata("prompt_tokens", response.usage.prompt_tokens)
        
    # Span is automatically reported on exit (even if an error occurs)

⁠Manual Reporting

If you prefer direct control over the span data, you can use the reporter manually.

from instrumentation_sdk import get_reporter

reporter = get_reporter()
reporter.report({
    "span_id": "unique-id",
    "service_name": "my-service",
    "status": "success",
    "text": "The prompt content to be enriched"
})

⁠Docker Deployment

The instrumentation SDK API is available as a production-ready, fully self-contained All-in-One Standalone Observability Container. This container bundles the FastAPI application, Grafana, and Tempo into a single image, eliminating the need to set up external databases or visualizers manually.

Image Name: chiefj/instrumentation-sdk-api:unstable (or chiefj/instrumentation-sdk-api:latest)

To pull and run the fully integrated all-in-one container locally:

# Pull the latest standalone image
docker pull chiefj/instrumentation-sdk-api:unstable

# Run the unified all-in-one telemetry stack
docker run -d \
  -p 8002:8000 \
  -p 3002:3000 \
  --name instrumentation-api-allinone \
  chiefj/instrumentation-sdk-api:unstable

• API Endpoints: Accessible at http://localhost:8002
• Grafana Portal: Accessible at http://localhost:3002 (Tempo is automatically provisioned as a read-only datasource and ready to query!)

⁠Observability Launcher CLI (llm-observe)

To simplify launching and managing the all-in-one container, the SDK provides a built-in command-line utility named llm-observe.

Once instrumentation-sdk is installed, you can manage the observability stack with simple commands:

# Start the stack on host ports: API (8002), Grafana (3002), OTLP (4317), Prometheus (9090)
llm-observe start

# Check the status of the container stack
llm-observe status

# Stop and clean up the container stack
llm-observe stop

⁠Customizing Ports & Names

You can customize the container name, ports, image, and tag:

llm-observe start --name my-observability --api-port 8005 --grafana-port 3005

For development with hot-reloading, use the provided Docker Compose:

docker compose -f packages/python/instrumentation-sdk/deploy/docker/docker-compose.dev.yaml up instrumentation-api

⁠Token Counting (Pre-Call Token Counting)

The SDK provides automatic pre-call token counting utilizing tiktoken with fallback character-based heuristics for non-OpenAI models. It supports plain text strings, complex chat message list schemas, and OpenAI's tile-based vision token calculation.

⁠Direct Token Counting

Use count_tokens to calculate tokens directly:

from instrumentation_sdk import count_tokens

tokens, method = count_tokens("hello world", "gpt-4")

⁠Context Manager with Automated Token Tracking

Use llm_span_with_tokens to automatically record prompt_tokens and token_count_method inside manual spans:

from instrumentation_sdk import llm_span_with_tokens

async def handle_request(req):
    async with llm_span_with_tokens(model="gpt-4", provider="openai", prompt="hello world") as span:
        pass

⁠REST Management API (REST)

The /v1/token-counting/count REST API endpoint supports counting prompt tokens:

curl -X POST http://localhost:8000/v1/token-counting/count \
  -H "Content-Type: application/json" \
  -d '{"prompt": "hello world", "model": "gpt-4"}'

⁠Streaming Observability (TTFT & Token Tracking)

The SDK provides specialized utilities for tracking streaming LLM calls. It wraps generators/iterators to:

• Capture the Time-to-First-Token (TTFT) latency when the first chunk is yielded.
• Accumulate the streamed chunks and automatically compute the completion token count (using Tiktoken/heuristics) upon stream completion or cancellation.
• Finalize and report the manual span only when the stream is exhausted, closed, or encounters an exception.

⁠Basic Streaming Usage

Use llm_streaming_span, wrap_stream (for synchronous generators), and wrap_async_stream (for asynchronous generators):

from instrumentation_sdk import llm_streaming_span, wrap_stream, wrap_async_stream

# 1. Synchronous Streaming
with llm_streaming_span(model="gpt-4", provider="openai", prompt="Say hello") as span_ctx:
    raw_generator = ["Hello", " world", "!"]
    wrapped_stream = wrap_stream(raw_generator, span_context=span_ctx, model="gpt-4")
    for chunk in wrapped_stream:
        print(chunk)

# 2. Asynchronous Streaming
async with llm_streaming_span(model="gpt-4", provider="openai", prompt="Say hello") as span_ctx:
    async def async_generator():
        yield "Hello"
        yield " world"
    wrapped_stream = wrap_async_stream(async_generator(), span_context=span_ctx, model="gpt-4")
    async for chunk in wrapped_stream:
        print(chunk)

⁠Mid-Stream Updates & Abort Resilience

• You can dynamically update span metadata using span_ctx.set_metadata("custom_field", "value") mid-stream.
• If the stream is closed early (via wrapped_stream.close() or .aclose()), the SDK captures and reports all completion tokens generated up to that point.

⁠REST Verification Endpoint

The /v1/streaming/test-stream-call endpoint streams SSE events back to the client while validating end-to-end streaming tracing:

curl -X POST http://localhost:8000/v1/streaming/test-stream-call \
  -H "Content-Type: application/json" \
  -d '{"provider": "openai", "chunks": ["A", "B", "C"]}'

⁠PII & Injection Scan (Aho-Corasick Redaction)

The SDK features an inline Aho-Corasick trie-based scanner that runs on all prompts inside manual span contexts (LLMSpanContext and LLMSpanWithTokensContext). It intercepts prompts, detects PII and SQL/prompt injection, and updates telemetry accordingly.

⁠Redaction & Interception Behavior

• PII Detected: The prompt and downstream fields (like hashes and embeddings) are completely redacted (None or empty). The custom span attribute llm.pii_detected is set to True.
• Injection Detected: The prompt is preserved, but the custom span attribute llm.injection_attempt is set to True.
• Fail-Safe execution: Any exception raised inside the scanning engine is caught internally, allowing client code or FastAPI handler to execute without crashes.

⁠Programmatic Scan Usage

You can import and call scan_prompt directly to inspect a prompt:

from instrumentation_sdk import scan_prompt

# Returns (pii_detected: bool, injection_attempt: bool)
pii, inj = scan_prompt("my email is [email protected]")
print(f"PII: {pii}, Injection: {inj}")

⁠REST Scanning Endpoint

The /v1/pii-injection/scan REST API endpoint supports checking prompt contents:

curl -X POST http://localhost:8000/v1/pii-injection/scan \
  -H "Content-Type: application/json" \
  -d '{"prompt": "my email is [email protected]"}'

⁠Deterministic Sampling Gate (Modulo 100)

The SDK implements deterministic sampling decided at span creation time. It hashes the span_id using SHA256 and evaluates whether the hash value modulo 100 is equal to 0.

• Sampled (is_sampled is True): The span is processed normally, performing prompt hashing and embedding generation.
• Unsampled (is_sampled is False): The span drops/skips both the SHA256 hashing and the MiniLM embedding generation, saving computational resources.

⁠Programmatic Usage

You can query the sampling logic directly:

from instrumentation_sdk import should_sample

sampled = should_sample("test-span-id")

⁠REST Endpoint

Query the /v1/sampling/should-sample endpoint to check sampling:

curl -X POST http://localhost:8000/v1/sampling/should-sample \
  -H "Content-Type: application/json" \
  -d '{"span_id": "test-span-id"}'

⁠MiniLM Embedding (Concurrent & Sampled)

The SDK asynchronously calls the embedding-worker HTTP endpoint (POST /embed) to generate a 384-dimensional vector embedding of the prompt text.

• Concurrent Execution: To prevent blocking client requests, the SDK uses asyncio.create_task() to fire the embedding generation concurrently with span finalization.
• Conditionality: The embedding is only generated if the span is sampled (is_sampled is True) and no PII is detected in the prompt (pii_detected is False).
• Timeout and Resilience: The embedding HTTP request has a timeout of 500ms. If the request times out or fails, the SDK falls back to None for the embedding field while the rest of the span details are still successfully emitted.

⁠Programmatic Usage

from instrumentation_sdk import get_embedding

embedding = await get_embedding("your text here")

⁠REST Endpoint

curl -X POST http://localhost:8000/v1/embeddings/embed \
  -H "Content-Type: application/json" \
  -d '{"text": "your text here"}'

⁠Prometheus Metrics & Grafana Dashboard

The SDK integrates a Prometheus metrics collection pipeline to track operational metrics for LLM calls (latency, TTFT, token usage, cost, and security violations).

⁠Configuration & Initialization

Initialize the Prometheus metrics scraping endpoint:

curl -X POST http://localhost:8000/v1/metrics/init \
  -H "Content-Type: application/json" \
  -d '{"port": 9464}'

⁠Metrics Endpoints

• Initialize Pipeline: POST /v1/metrics/init
• Health Check: GET /v1/metrics/health
• Record Single Span Metrics: POST /v1/metrics/record
• Record Batch Spans Metrics: POST /v1/metrics/record-batch

⁠Grafana Dashboard

The dashboard is built-in and automatically provisioned on port 3000 (or 3002 in standalone mode). It includes:

• LLM Latency & TTFT: Histogram distribution of request latency and time-to-first-token.
• Token Usage: Track prompt and completion tokens.
• Cost Analysis: Live cost calculation in micro-USD.
• Security Scans: Record rates of PII exposure and prompt injections.

⁠Updating Config Files (Model Prices, PII Patterns, Infra)

The SDK reads config files once at startup. After any change, a container restart is required (except dashboard JSON files which are hot-reloaded).

⁠Adding or updating a model price

Edit config/model_prices.yaml:

- model: gpt-5
  provider: openai
  input_price_per_1m: 10.00
  output_price_per_1m: 30.00
  version: "2026-01-01"

Required fields: model, provider, input_price_per_1m, output_price_per_1m, version.
Prices must be >= 0. Duplicate (model, provider) pairs are rejected by CI.

Then restart:

docker restart instrumentation-sdk-api

⁠Adding or updating a PII / Injection pattern

Edit config/patterns.yaml:

patterns:
  - name: phone_number
    regex: "\\b\\d{3}[-.]?\\d{3}[-.]?\\d{4}\\b"
    type: PII_STRUCTURAL   # or INJECTION_ATTEMPT

Valid type values: PII_STRUCTURAL, INJECTION_ATTEMPT.
The CI validates that every regex compiles and no pattern name is duplicated.

Then restart:

docker restart instrumentation-sdk-api

⁠Updating a Grafana dashboard

Edit any file under build/dashboards/*.json.
No restart required — Grafana polls and hot-reloads dashboards every 30 seconds automatically.

⁠Updating infra configs (Grafana datasource, Prometheus, Tempo)

Edit:

• build/grafana-datasource.yaml — add/change datasources
• build/grafana-dashboard-provider.yaml — change dashboard provider path/folder
• build/prometheus.yml — add scrape targets
• build/tempo-config.yaml — change Tempo storage or OTLP port

After editing, rebuild the image and restart:

DOCKER_PAT=<your-pat> ./scripts/deploy_docker.sh
docker stop instrumentation-sdk-api && docker rm instrumentation-sdk-api
docker pull chiefj/instrumentation-sdk-api:latest
docker run -d -p 8000:8000 -p 3000:3000 -p 4317:4317 -p 9464:9464 \
  --name instrumentation-sdk-api chiefj/instrumentation-sdk-api:latest

⁠CI — what runs automatically

The grafana-config-validate.yml workflow triggers only when one of the 10 watched files changes. It validates:

⁠Running the load test locally

cd packages/python/instrumentation-sdk
.venv/bin/python -m pytest tests/performance/ -m performance -v

This sends 1000 spans (100 individual + 10×50 batch) covering all 6 model/provider combos, error ratios, PII flags, and high token counts.

⁠3. Implementation Call Chain