Add guard observability hooks to Safety Agent SDKs

[dushyantzz]

Description

Adds observability hooks to the Safety Agent guard flow in both TS and Python SDKs (start/segment/result/error), with docs and tests.

Type of Change

-[ ] Bug fix
-[x] New feature
-[ ] Breaking change
-[x] Documentation update

Testing

cd sdk/typescript && npm test (failed: missing provider API keys)
cd sdk/python && PYTHONPATH=./src python -m pytest (failed: missing provider API keys; hook tests pass)

Checklist

-[x] Code follows project style guidelines
-[ ] Tests pass locally
-[x] Documentation updated (if needed)

[vercel]

@dushyantzz is attempting to deploy a commit to the Superagent Team on Vercel.

A member of the Team first needs to authorize it.

[dushyantzz]

The CI failures are occuring due to missing API keys in the GitHub Actions environment.

[homanp]

@cursor review

[homanp]

@dushyantzz how does this hook into existing observability providers like Langfuse etc.? Care to give some details?

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

[cursor]

Async hook exceptions not caught in Python

Medium Severity

When an async hook raises an exception, it's not caught or suppressed. The asyncio.create_task(result) call schedules the coroutine to run in the background, but provides no exception handler for the task. If the async hook raises an exception during execution, Python logs an "Task exception was never retrieved" warning when the task is garbage collected, violating the stated intent to "swallow hook errors to avoid impacting guard execution."

 

[homanp]

@cursor how does this hook into existing observability tools like Langfuse etc.?

[cursor]

Perfect! Now I have a complete understanding of the observability hooks implementation. Let me create a comprehensive response explaining how these hooks integrate with observability tools like Langfuse.

Based on the PR implementation, I can now provide a detailed answer about how the observability hooks work and how they can integrate with tools like Langfuse. The hooks provide lifecycle events (start, segment, result, error) that can be used to send telemetry data to any observability platform.

How the Observability Hooks Work

The observability hooks added to the Safety Agent SDKs provide a callback-based integration pattern that allows you to instrument the guard() method with your own observability tooling. Here's how they integrate with tools like Langfuse:

Hook Architecture

The implementation provides 4 lifecycle hooks for the guard flow:

1. onStart - Fired when guard execution begins (includes model, input type, segment count, etc.)
2. onSegment - Fired after each chunk/page/image is analyzed (includes result, duration, token usage)
3. onResult - Fired when the final aggregated result is ready (includes total duration and segment count)
4. onError - Fired if any error occurs during execution (includes error details and context)

Integration with Langfuse (Example)

Here's how you would integrate with Langfuse or similar observability platforms:

TypeScript:

import { createClient } from 'safety-agent';
import { Langfuse } from 'langfuse';

const langfuse = new Langfuse({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY,
  secretKey: process.env.LANGFUSE_SECRET_KEY,
});

const client = createClient({ apiKey: process.env.SUPERAGENT_API_KEY });

const trace = langfuse.trace({ name: 'safety-guard-check' });

const result = await client.guard({
  input: "user message to analyze",
  model: "openai/gpt-4o-mini",
  hooks: {
    onStart: (event) => {
      trace.update({
        metadata: {
          model: event.model,
          inputType: event.inputType,
          inputSize: event.inputSize,
          segmentCount: event.segmentCount,
        }
      });
    },
    
    onSegment: (event) => {
      // Log each segment as a span
      trace.span({
        name: `guard-segment-${event.kind}`,
        input: { size: event.segmentSize, units: event.segmentUnits },
        output: { 
          classification: event.result.classification,
          violations: event.result.violation_types 
        },
        metadata: {
          index: event.index,
          totalCount: event.count,
          durationMs: event.durationMs,
        },
        usage: {
          promptTokens: event.result.usage.promptTokens,
          completionTokens: event.result.usage.completionTokens,
          totalTokens: event.result.usage.totalTokens,
        }
      });
    },
    
    onResult: (event) => {
      trace.update({
        output: {
          classification: event.result.classification,
          reasoning: event.result.reasoning,
          violations: event.result.violation_types,
          cweCodes: event.result.cwe_codes,
        },
        usage: {
          promptTokens: event.result.usage.promptTokens,
          completionTokens: event.result.usage.completionTokens,
          totalTokens: event.result.usage.totalTokens,
        },
        metadata: {
          durationMs: event.durationMs,
          segmentCount: event.segmentCount,
        }
      });
    },
    
    onError: (event) => {
      trace.update({
        level: 'ERROR',
        statusMessage: event.error.message,
      });
    },
  },
});

await langfuse.flushAsync();

Python:

from safety_agent import create_client, GuardHooks
from langfuse import Langfuse

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)

client = create_client(api_key=os.environ["SUPERAGENT_API_KEY"])

trace = langfuse.trace(name="safety-guard-check")

def on_start(event):
    trace.update(metadata={
        "model": event.model,
        "input_type": event.input_type,
        "input_size": event.input_size,
        "segment_count": event.segment_count,
    })

def on_segment(event):
    trace.span(
        name=f"guard-segment-{event.kind}",
        input={"size": event.segment_size, "units": event.segment_units},
        output={
            "classification": event.result.classification,
            "violations": event.result.violation_types,
        },
        metadata={
            "index": event.index,
            "total_count": event.count,
            "duration_ms": event.duration_ms,
        },
        usage={
            "prompt_tokens": event.result.usage.prompt_tokens,
            "completion_tokens": event.result.usage.completion_tokens,
            "total_tokens": event.result.usage.total_tokens,
        }
    )

def on_result(event):
    trace.update(
        output={
            "classification": event.result.classification,
            "reasoning": event.result.reasoning,
            "violations": event.result.violation_types,
            "cwe_codes": event.result.cwe_codes,
        },
        usage={
            "prompt_tokens": event.result.usage.prompt_tokens,
            "completion_tokens": event.result.usage.completion_tokens,
            "total_tokens": event.result.usage.total_tokens,
        },
        metadata={
            "duration_ms": event.duration_ms,
            "segment_count": event.segment_count,
        }
    )

hooks = GuardHooks(
    on_start=on_start,
    on_segment=on_segment,
    on_result=on_result,
)

result = await client.guard(
    input="user message to analyze",
    model="openai/gpt-4o-mini",
    hooks=hooks,
)

langfuse.flush()

Key Benefits

1. Provider-Agnostic: The hooks are generic callbacks that work with any observability platform (Langfuse, OpenTelemetry, Datadog, custom logging, etc.)

2. Granular Visibility: Track individual chunks/pages/segments separately, useful for understanding performance bottlenecks in large inputs

3. Token Tracking: Each segment event includes token usage, enabling accurate cost tracking per segment

4. Error Context: Error events include segment context (which chunk/page failed) for better debugging

5. Non-Blocking: Hooks are fire-and-forget - errors in hooks don't impact guard execution (as noted in the bugbot issue about async error handling)

Integration Patterns

The hooks support multiple integration patterns:

• Langfuse/LangSmith: Trace and span creation for LLM observability
• OpenTelemetry: Custom span instrumentation
• Datadog/New Relic: APM metric emission
• Custom Analytics: Send events to your own analytics pipeline
• Logging: Simple console/file logging for debugging

The design is intentionally minimal and flexible - you receive structured event objects and can transform/send them however you need for your observability stack.

 

[dushyantzz]

@homanp The hooks are provider agnostic and don’t hardcode any observability vendor. It emits structured lifecycle events (onStart, onSegment, onResult, onError) with timing + usage + classifications.
You can forward those events into Langfuse, OpenTelemetry, Datadog, etc. from your app by wiring the hook callbacks to your client SDK. That keeps this SDK light while still enabling integrations

[homanp]

@homanp The hooks are provider agnostic and don’t hardcode any observability vendor. It emits structured lifecycle events (onStart, onSegment, onResult, onError) with timing + usage + classifications. You can forward those events into Langfuse, OpenTelemetry, Datadog, etc. from your app by wiring the hook callbacks to your client SDK. That keeps this SDK light while still enabling integrations

I was thinking one could hook it into OpenTelemetry calls that already exists into upstream apps. But perhaps this is a better way to just keep it agnostic. A good start. Will test it out shortly!