tessl/npm-langfuse--tracing

Langfuse instrumentation methods based on OpenTelemetry

Overview

Eval results

Files

Observe Decorator

Name: tessl/npm-langfuse--tracing
Author: tessl

The observe() function provides a decorator-style approach to adding observability to existing functions without modifying their implementation. It wraps functions with automatic tracing, input/output capture, and error tracking while preserving their original behavior and type signatures.

Core Function

observe

Decorator function that automatically wraps any function with Langfuse observability.

/**
 * Decorator function that automatically wraps any function with Langfuse observability.
 *
 * @param fn - The function to wrap with observability
 * @param options - Configuration for observation behavior
 * @returns An instrumented version of the function
 */
function observe<T extends (...args: any[]) => any>(
  fn: T,
  options?: ObserveOptions
): T;

interface ObserveOptions {
  /** Name for the observation (defaults to function name) */
  name?: string;
  /** Type of observation to create */
  asType?: LangfuseObservationType;
  /** Whether to capture function input as observation input. Default: true */
  captureInput?: boolean;
  /** Whether to capture function output as observation output. Default: true */
  captureOutput?: boolean;
  /** Parent span context to attach this observation to */
  parentSpanContext?: SpanContext;
  /** Whether to automatically end the observation when exiting. Default: true */
  endOnExit?: boolean;
}

Key Features

Zero Code Changes

Wrap existing functions without modifying their internal logic.

import { observe } from '@langfuse/tracing';

// Original function
async function fetchUserData(userId: string) {
  const response = await fetch(`/api/users/${userId}`);
  return response.json();
}

// Wrapped function - behavior unchanged, now with observability
const tracedFetchUserData = observe(fetchUserData, {
  name: 'fetch-user-data',
  asType: 'span'
});

// Use exactly as before
const user = await tracedFetchUserData('user-123');

Automatic I/O Capture

Function arguments and return values are automatically captured as input and output.

const processOrder = observe(
  async (orderId: string, items: CartItem[]) => {
    const validation = await validateOrder(orderId, items);
    const payment = await processPayment(validation);
    const shipping = await scheduleShipping(payment);

    return {
      orderId,
      status: 'confirmed',
      trackingId: shipping.id
    };
  },
  {
    name: 'process-order',
    captureInput: true,  // Captures [orderId, items]
    captureOutput: true  // Captures return value
  }
);

// Input and output automatically logged
const result = await processOrder('ord_123', cartItems);

Error Tracking

Errors are automatically captured with error level and status message.

const riskyOperation = observe(
  async (data: string) => {
    if (!data) {
      throw new Error('Data is required');
    }
    return processData(data);
  },
  { name: 'risky-operation' }
);

try {
  await riskyOperation('');
} catch (error) {
  // Error automatically captured in observation with level: 'ERROR'
}

Type Preservation

The wrapped function maintains the original signature and return types.

// Original function with specific types
async function calculateTotal(
  items: Item[],
  taxRate: number
): Promise<{ subtotal: number; tax: number; total: number }> {
  const subtotal = items.reduce((sum, item) => sum + item.price, 0);
  const tax = subtotal * taxRate;
  return { subtotal, tax, total: subtotal + tax };
}

// Wrapped function maintains exact types
const tracedCalculateTotal = observe(calculateTotal, {
  name: 'calculate-total'
});

// TypeScript knows the return type
const result: Promise<{ subtotal: number; tax: number; total: number }> =
  tracedCalculateTotal(items, 0.08);

Basic Usage

Simple Function Wrapping

import { observe } from '@langfuse/tracing';

// Wrap a synchronous function
const addNumbers = observe(
  (a: number, b: number) => a + b,
  { name: 'add-numbers' }
);

const sum = addNumbers(5, 3); // 8

// Wrap an async function
const fetchData = observe(
  async (url: string) => {
    const response = await fetch(url);
    return response.json();
  },
  { name: 'fetch-data' }
);

const data = await fetchData('https://api.example.com/data');

Named vs Anonymous Functions

The decorator uses the function name by default, but you can override it.

// Named function - uses function name as observation name
function processUser(userId: string) {
  return { id: userId, processed: true };
}

const traced = observe(processUser); // name: 'processUser'

// Anonymous function - provide explicit name
const tracedAnon = observe(
  (userId: string) => ({ id: userId, processed: true }),
  { name: 'process-user-anonymous' }
);

LLM Function Wrapping

Generation Observation

Wrap LLM API calls with generation-type observations.

const generateSummary = observe(
  async (document: string, maxWords: number = 100) => {
    const response = await openai.chat.completions.create({
      model: 'gpt-4-turbo',
      messages: [
        { role: 'system', content: `Summarize in ${maxWords} words or less` },
        { role: 'user', content: document }
      ],
      max_tokens: maxWords * 2
    });

    return response.choices[0].message.content;
  },
  {
    name: 'document-summarizer',
    asType: 'generation',
    captureInput: true,
    captureOutput: true
  }
);

// Use the wrapped function
const summary = await generateSummary(longDocument, 150);

Embedding Generation

const generateEmbeddings = observe(
  async (texts: string[]) => {
    const response = await openai.embeddings.create({
      model: 'text-embedding-ada-002',
      input: texts
    });

    return response.data.map(item => item.embedding);
  },
  {
    name: 'text-embedder',
    asType: 'embedding',
    captureInput: true,
    captureOutput: false // Don't log large vectors
  }
);

const vectors = await generateEmbeddings(['Hello', 'World']);

Specialized Observation Types

Agent Function

const researchAgent = observe(
  async (query: string, maxSources: number = 3) => {
    // Search for relevant documents
    const documents = await searchDocuments(query, maxSources * 2);

    // Filter and rank results
    const topDocs = documents
      .filter(d => d.score > 0.7)
      .slice(0, maxSources);

    // Generate comprehensive answer
    const context = topDocs.map(d => d.content).join('\n\n');
    const answer = await generateSummary(
      `Based on: ${context}\n\nQuestion: ${query}`,
      200
    );

    return {
      answer,
      sources: topDocs.map(d => d.source),
      confidence: Math.min(...topDocs.map(d => d.score))
    };
  },
  {
    name: 'research-agent',
    asType: 'agent',
    captureInput: true,
    captureOutput: true
  }
);

Tool Function

const searchDocuments = observe(
  async (query: string, topK: number = 5) => {
    const embedding = await embedText(query);
    const results = await vectorDb.search(embedding, topK);

    return results.map(r => ({
      content: r.metadata.content,
      score: r.score,
      source: r.metadata.source
    }));
  },
  {
    name: 'document-search',
    asType: 'retriever',
    captureInput: true,
    captureOutput: true
  }
);

Evaluator Function

const evaluateResponse = observe(
  (response: string, reference: string, metric: string = 'similarity') => {
    let score: number;

    switch (metric) {
      case 'similarity':
        score = calculateCosineSimilarity(response, reference);
        break;
      case 'bleu':
        score = calculateBleuScore(response, reference);
        break;
      default:
        throw new Error(`Unknown metric: ${metric}`);
    }

    return {
      score,
      passed: score > 0.8,
      metric,
      grade: score > 0.9 ? 'excellent' : score > 0.7 ? 'good' : 'needs_improvement'
    };
  },
  {
    name: 'response-evaluator',
    asType: 'evaluator',
    captureInput: true,
    captureOutput: true
  }
);

Guardrail Function

const moderateContent = observe(
  async (text: string, policies: string[] = ['profanity', 'spam']) => {
    const violations = [];

    for (const policy of policies) {
      const result = await checkPolicy(text, policy);
      if (result.violation) {
        violations.push({ policy, severity: result.severity });
      }
    }

    return {
      allowed: violations.length === 0,
      violations,
      confidence: 0.95
    };
  },
  {
    name: 'content-moderator',
    asType: 'guardrail',
    captureInput: true,
    captureOutput: true
  }
);

Class Method Decoration

Constructor Pattern

Wrap methods during class instantiation.

class UserService {
  private db: Database;

  constructor(database: Database) {
    this.db = database;

    // Wrap methods in constructor
    this.createUser = observe(this.createUser.bind(this), {
      name: 'create-user',
      asType: 'span',
      captureInput: false,  // Sensitive data
      captureOutput: true
    });

    this.fetchUser = observe(this.fetchUser.bind(this), {
      name: 'fetch-user',
      asType: 'span'
    });
  }

  async createUser(userData: UserData) {
    // Implementation automatically traced
    return await this.db.users.create(userData);
  }

  async fetchUser(userId: string) {
    // Implementation automatically traced
    return await this.db.users.findUnique({ where: { id: userId } });
  }
}

const service = new UserService(db);
const user = await service.createUser({ name: 'Alice' });

Factory Pattern

Create traced instances using a factory function.

class AIService {
  async generateText(prompt: string) {
    return await llm.generate(prompt);
  }

  async embedText(text: string) {
    return await embedder.embed(text);
  }
}

function createTracedAIService(): AIService {
  const service = new AIService();

  service.generateText = observe(service.generateText.bind(service), {
    name: 'ai-generate-text',
    asType: 'generation'
  });

  service.embedText = observe(service.embedText.bind(service), {
    name: 'ai-embed-text',
    asType: 'embedding'
  });

  return service;
}

const ai = createTracedAIService();

Input/Output Control

Selective Capture

Control what gets captured to avoid logging sensitive data or large payloads.

// Capture input but not output (sensitive results)
const fetchUserProfile = observe(
  async (userId: string) => {
    const user = await db.users.findUnique({ where: { id: userId } });
    const preferences = await db.preferences.findMany({ where: { userId } });
    return { ...user, preferences };
  },
  {
    name: 'fetch-user-profile',
    captureInput: false,   // Don't capture user IDs
    captureOutput: false   // Don't capture sensitive profile data
  }
);

// Capture output but not input (large payloads)
const processLargeFile = observe(
  async (fileBuffer: Buffer) => {
    const processed = await processFile(fileBuffer);
    return { size: processed.length, checksum: processed.checksum };
  },
  {
    name: 'process-file',
    captureInput: false,   // Don't log large buffer
    captureOutput: true    // Log summary info
  }
);

Argument Capture Behavior

Input capture handles different argument patterns automatically.

// Single argument - captured as-is
const singleArg = observe(
  (value: string) => value.toUpperCase(),
  { name: 'single-arg', captureInput: true }
);
singleArg('hello'); // input: "hello"

// Multiple arguments - captured as array
const multiArg = observe(
  (a: number, b: number, c: number) => a + b + c,
  { name: 'multi-arg', captureInput: true }
);
multiArg(1, 2, 3); // input: [1, 2, 3]

// No arguments - input is undefined
const noArg = observe(
  () => Date.now(),
  { name: 'no-arg', captureInput: true }
);
noArg(); // input: undefined

Function Composition

Observed functions remain fully composable.

// Individual observed functions
const fetchData = observe(
  async (url: string) => {
    const response = await fetch(url);
    return response.json();
  },
  { name: 'fetch-data', asType: 'tool' }
);

const processData = observe(
  async (data: any) => {
    return data.items.map(item => ({
      id: item.id,
      value: item.value * 2
    }));
  },
  { name: 'process-data', asType: 'span' }
);

const saveData = observe(
  async (data: any) => {
    return await db.items.createMany({ data });
  },
  { name: 'save-data', asType: 'span' }
);

// Compose into pipeline
const dataPipeline = observe(
  async (url: string) => {
    const raw = await fetchData(url);
    const processed = await processData(raw);
    const saved = await saveData(processed);
    return saved;
  },
  { name: 'data-pipeline', asType: 'chain' }
);

// Single call creates hierarchical trace
await dataPipeline('https://api.example.com/data');

Advanced Patterns

Conditional Tracing

Wrap functions conditionally based on environment.

function maybeObserve<T extends (...args: any[]) => any>(
  fn: T,
  options: ObserveOptions
): T {
  if (process.env.LANGFUSE_ENABLED === 'true') {
    return observe(fn, options);
  }
  return fn;
}

const processOrder = maybeObserve(
  async (orderId: string) => {
    return await performProcessing(orderId);
  },
  { name: 'process-order' }
);

Middleware Pattern

Create reusable observation middleware.

function withObservation<T extends (...args: any[]) => any>(
  options: ObserveOptions
) {
  return (fn: T): T => {
    return observe(fn, options);
  };
}

// Create middleware
const asGeneration = withObservation({ asType: 'generation' });
const asTool = withObservation({ asType: 'tool' });

// Apply to functions
const generateText = asGeneration(async (prompt: string) => {
  return await llm.generate(prompt);
});

const searchWeb = asTool(async (query: string) => {
  return await webApi.search(query);
});

Decorator Factory

Create domain-specific decorators.

function observeLLM(name: string, model: string) {
  return <T extends (...args: any[]) => any>(fn: T): T => {
    return observe(fn, {
      name,
      asType: 'generation',
      captureInput: true,
      captureOutput: true
    });
  };
}

function observeTool(name: string) {
  return <T extends (...args: any[]) => any>(fn: T): T => {
    return observe(fn, {
      name,
      asType: 'tool',
      captureInput: true,
      captureOutput: true
    });
  };
}

// Use decorators
const chatGPT = observeLLM('chat-gpt', 'gpt-4')(
  async (prompt: string) => await openai.chat(prompt)
);

const webSearch = observeTool('web-search')(
  async (query: string) => await google.search(query)
);

Error Handling

Automatic Error Capture

Errors are automatically logged with error level.

const riskyOperation = observe(
  async (data: string) => {
    if (!data) {
      throw new Error('Data required');
    }

    const processed = await processData(data);

    if (!processed.valid) {
      throw new Error('Processing failed validation');
    }

    return processed;
  },
  { name: 'risky-operation' }
);

try {
  await riskyOperation('');
} catch (error) {
  // Observation automatically updated with:
  // - level: 'ERROR'
  // - statusMessage: error.message
  // - output: { error: error.message }
}

Error Propagation

Errors are re-thrown after being captured, preserving error handling.

const operation = observe(
  async () => {
    throw new Error('Operation failed');
  },
  { name: 'failing-operation' }
);

try {
  await operation();
} catch (error) {
  // Error was logged in observation
  console.error('Caught error:', error.message);
  // Normal error handling continues
}

Best Practices

Descriptive Names

Use clear, descriptive names that indicate the operation's purpose.

// Good: Specific and clear
const generateProductDescription = observe(fn, {
  name: 'generate-product-description'
});

// Avoid: Generic and unclear
const process = observe(fn, { name: 'process' });

Appropriate Observation Types

Choose the correct observation type for the operation.

// LLM calls -> generation
const llmCall = observe(fn, { asType: 'generation' });

// API calls -> tool
const apiCall = observe(fn, { asType: 'tool' });

// Multi-step workflows -> chain
const pipeline = observe(fn, { asType: 'chain' });

// General operations -> span (default)
const operation = observe(fn, { asType: 'span' });

Sensitive Data

Disable capture for functions handling sensitive information.

const processPayment = observe(
  async (cardNumber: string, cvv: string) => {
    return await paymentGateway.charge({ cardNumber, cvv });
  },
  {
    name: 'process-payment',
    captureInput: false,   // Don't log card details
    captureOutput: false   // Don't log payment response
  }
);

Performance Considerations

For high-frequency operations, consider disabling capture.

const logMetric = observe(
  (metric: string, value: number) => {
    metrics.record(metric, value);
  },
  {
    name: 'log-metric',
    captureInput: false,   // Reduce overhead
    captureOutput: false
  }
);

Composition Over Deep Nesting

Prefer composing observed functions over deeply nested callbacks.

// Good: Composed, traceable functions
const step1 = observe(async (data) => { /* ... */ }, { name: 'step-1' });
const step2 = observe(async (data) => { /* ... */ }, { name: 'step-2' });
const step3 = observe(async (data) => { /* ... */ }, { name: 'step-3' });

const pipeline = observe(
  async (input) => {
    const a = await step1(input);
    const b = await step2(a);
    return await step3(b);
  },
  { name: 'pipeline', asType: 'chain' }
);

// Avoid: Deep nesting in single function
const monolithic = observe(
  async (input) => {
    const a = await /* complex step 1 */;
    const b = await /* complex step 2 */;
    return await /* complex step 3 */;
  },
  { name: 'monolithic' }
);

Install with Tessl CLI