LLM Observability Docs

export DEADPIPE_API_KEY="dp_your_api_key"

pip install deadpipe

from deadpipe import track
from openai import OpenAI

client = OpenAI()

# Store params in variable for context capture
params = {
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "Hello!"}]
}

with track(prompt_id="my_agent") as t:
    response = client.chat.completions.create(**params)
    # Pass input params to capture full context (messages, tools, system prompt)
    t.record(response, input=params)

from deadpipe import wrap_openai
from pydantic import BaseModel, Field
from typing import Literal
from openai import OpenAI

class ProductRecommendation(BaseModel):
    product_id: str
    confidence: float = Field(ge=0, le=1)
    reasoning: str
    category: Literal["electronics", "clothing", "home"]
    alternatives: list[str] = []

# Schema validation with wrapper
client = wrap_openai(OpenAI(), app="my_app")

response = client.chat.completions.create(
    prompt_id="recommender",
    schema=ProductRecommendation,  # Per-call schema validation
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Recommend a laptop"}],
    response_format={"type": "json_object"}
)
# → Automatically validates output against ProductRecommendation
# → Tracks schema pass rates over time
# → Alerts when validation rates drop

from deadpipe import track

with track(
    prompt_id="pricing_agent",
    enum_fields={
        "currency": ["USD", "EUR", "GBP"],
        "tier": ["free", "pro", "enterprise"]
    },
    numeric_bounds={
        "price": (0, 10000),      # Must be between 0-10000
        "quantity": (1, 100)      # Must be between 1-100
    }
) as t:
    response = client.chat.completions.create(...)
    t.record(response)
    # → Automatically flags enum_out_of_range and numeric_out_of_bounds

from deadpipe import track

with track(prompt_id="streaming_agent") as t:
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Write a story"}],
        stream=True
    )

    chunks = []
    for chunk in stream:
        if chunk.choices[0].delta.content:
            t.mark_first_token()  # Call once when first content arrives
            chunks.append(chunk.choices[0].delta.content)
            print(chunk.choices[0].delta.content, end="")

    # Record with the stream object - captures timing
    t.record(stream)

from deadpipe import wrap, wrap_anthropic
from anthropic import Anthropic

# Option 1: Universal wrap (recommended)
client = wrap(Anthropic(), app="my_app")

# Option 2: Provider-specific wrapper
client = wrap_anthropic(Anthropic(), app="my_app")

# All calls automatically tracked - pass prompt_id per call
response = client.messages.create(
    prompt_id="claude_agent",
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude!"}]
)
# → Provider auto-detected, cost estimated automatically

with track(prompt_id="retrying_agent") as t:
    for attempt in range(3):
        try:
            t.mark_retry()  # Call before each retry
            response = client.chat.completions.create(...)
            t.record(response)
            break
        except Exception as e:
            if attempt == 2:
                raise

# Required
export DEADPIPE_API_KEY="dp_your_api_key"

# Optional
export DEADPIPE_APP_ID="my-app"
export DEADPIPE_ENVIRONMENT="production"
export DEADPIPE_VERSION="v1.2.3"  # or GIT_COMMIT

npm install deadpipe
# or
yarn add deadpipe
# or
pnpm add deadpipe

import { track } from 'deadpipe';
import OpenAI from 'openai';

const client = new OpenAI();

// Store params in variable for context capture
const params = {
  model: 'gpt-4o-mini',
  messages: [{ role: 'user', content: 'Hello!' }]
};

const response = await track('my_agent', async (t) => {
  const response = await client.chat.completions.create(params);
  // Pass input params to capture full context (messages, tools, system prompt)
  t.record(response, undefined, params);
  return response;
});

import { track, SchemaValidator } from 'deadpipe';
import { z } from 'zod';

const ProductSchema = z.object({
  productId: z.string(),
  confidence: z.number().min(0).max(1),
  reasoning: z.string(),
  category: z.enum(['electronics', 'clothing', 'home']),
  alternatives: z.array(z.string()).default([])
});

// Create Zod adapter for Deadpipe
const zodValidator: SchemaValidator = {
  validate: (data) => {
    const result = ProductSchema.safeParse(data);
    return {
      success: result.success,
      data: result.success ? result.data : undefined,
      errors: result.success ? undefined : result.error.errors.map(e => e.message)
    };
  }
};

const result = await track('recommender', async (t) => {
  const response = await client.chat.completions.create({
    model: 'gpt-4o-mini',
    messages: [{ role: 'user', content: 'Recommend a laptop' }],
    response_format: { type: 'json_object' }
  });
  return t.record(response);
}, { schema: zodValidator });

// result is typed as ProductSchema | null
if (result) {
  console.log(result.productId); // TypeScript knows this is ProductSchema
}

const result = await track('pricing_agent', async (t) => {
  const response = await client.chat.completions.create({
    model: 'gpt-4o-mini',
    messages: [{ role: 'user', content: 'Set pricing' }]
  });
  return t.record(response);
}, {
  enumFields: {
    currency: ['USD', 'EUR', 'GBP'],
    tier: ['free', 'pro', 'enterprise']
  },
  numericBounds: {
    price: [0, 10000],      // Must be between 0-10000
    quantity: [1, 100]      // Must be between 1-100
  }
});
// → Automatically flags enum_out_of_range and numeric_out_of_bounds

const params = {
  model: 'gpt-4',
  messages: [{ role: 'user', content: 'Tell me a story' }],
  stream: true,
};

const response = await track('streaming_agent', async (t) => {
  const stream = await client.chat.completions.create(params);

  let fullContent = '';
  for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
      t.markFirstToken(); // Call once on first token
      fullContent += chunk.choices[0].delta.content;
    }
  }

  // Record manually for streams - pass input params to capture context
  t.record({
    model: 'gpt-4',
    choices: [{ message: { content: fullContent } }],
    usage: { prompt_tokens: 10, completion_tokens: 100, total_tokens: 110 }
  }, undefined, params);

  return fullContent;
});

import { wrap, wrapAnthropic } from 'deadpipe';
import Anthropic from '@anthropic-ai/sdk';

// Option 1: Universal wrap (recommended)
const client = wrap(new Anthropic(), { app: 'my_app' });

// Option 2: Provider-specific wrapper
const client = wrapAnthropic(new Anthropic(), { app: 'my_app' });

// All calls automatically tracked - pass promptId per call
const response = await client.messages.create({
  promptId: 'claude_agent',
  model: 'claude-sonnet-4-20250514',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Hello, Claude!' }]
});
// → Provider auto-detected, cost estimated automatically

const response = await track('retrying_agent', async (t) => {
  for (let attempt = 0; attempt < 3; attempt++) {
    try {
      t.markRetry(); // Call before each retry
      const response = await client.chat.completions.create({...});
      t.record(response);
      return response;
    } catch (error) {
      if (attempt === 2) throw error;
    }
  }
});

# Required
export DEADPIPE_API_KEY="dp_your_api_key"

# Optional
export DEADPIPE_APP_ID="my-app"
export DEADPIPE_ENVIRONMENT="production"
export DEADPIPE_VERSION="1.2.3"
export GIT_COMMIT="abc123"  # Fallback for version

await track('my-prompt', async (t) => {
  // Mark when first token arrives (for streaming)
  t.markFirstToken();

  // Mark retry attempts
  t.markRetry();

  // Record the response (required)
  t.record(response, parsedOutput?, input?);

  // Or record an error manually
  t.recordError(error);

  // Check if already recorded
  if (!t.isRecorded()) { /* ... */ }

  // Get current telemetry
  const telemetry = t.getTelemetry();
});

from deadpipe import wrap
from openai import OpenAI

# If Deadpipe is down, your code continues working
client = wrap(OpenAI(), app="my_app")

try:
    response = client.chat.completions.create(prompt_id="agent", ...)
    # Works normally even if telemetry fails
except Exception as e:
    # Only your LLM error, never Deadpipe
    pass

curl -X POST https://deadpipe.com/api/v1/prompt \
  -H "Content-Type: application/json" \
  -H "X-API-Key: dp_your_api_key" \
  -d '{
    "prompt_id": "review-analyzer",
    "model": "gpt-4o-mini",
    "provider": "openai",
    "request_start": 1704067200000,
    "end_time": 1704067201500,
    "total_latency_ms": 1500,
    "input_tokens": 150,
    "output_tokens": 75,
    "total_tokens": 225,
    "estimated_cost": 0.00045,
    "http_status": 200,
    "output_length": 512,
    "empty_output": false,
    "json_parse_success": true,
    "schema_validation_pass": true,
    "refusal_flag": false,
    "output_hash": "abc123..."
  }'

{
  "received": true,
  "prompt_id": "review-analyzer",
  "event_id": "evt_abc123",
  "baseline": {
    "latency_mean": 1200,
    "latency_p95": 2500,
    "token_mean": 200,
    "schema_pass_rate": 0.98,
    "sample_count": 1500
  },
  "anomalies": []
}

curl https://deadpipe.com/api/v1/prompt?prompt_id=review-analyzer \
  -H "X-API-Key: dp_your_api_key"

from pydantic import BaseModel, Field
from typing import Literal, List

class ProductRecommendation(BaseModel):
    product_id: str
    confidence: float = Field(ge=0, le=1)
    reasoning: str
    category: Literal["electronics", "clothing", "home"]
    alternatives: List[str] = []

with dp.track("recommender", schema=ProductRecommendation) as t:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Recommend..."}],
        response_format={"type": "json_object"}
    )
    result = t.record(response)
    
    # result.schema_pass tells you if validation passed
    # result.validation_error contains the error message if failed
    # result.validated contains the parsed Pydantic model if passed

import { z } from 'zod';

const ProductSchema = z.object({
  productId: z.string(),
  confidence: z.number().min(0).max(1),
  reasoning: z.string(),
  category: z.enum(['electronics', 'clothing', 'home']),
  alternatives: z.array(z.string()).default([])
});

const result = await dp.track('recommender', {
  schema: ProductSchema,
  fn: async () => {
    return await client.chat.completions.create({
      model: 'gpt-4o-mini',
      messages: [{ role: 'user', content: 'Recommend...' }],
      response_format: { type: 'json_object' }
    });
  }
});

if (result.validated) {
  // TypeScript knows this is ProductSchema type
  console.log(result.validated.productId);
}