Observe

Wrap functions with Laminar to capture inputs, outputs, and errors automatically.

TypeScript
Python

observe(options, fn, …args)

Wrap a function to create a span. Automatically captures inputs, outputs, and errors.

import { observe } from '@lmnr-ai/lmnr';

const result = await observe(
  { name: 'my_operation' },
  async (param1, param2) => {
    return param1 + param2;
  },
  1, 2
);
// Span input: [1, 2], output: 3

Parameters:

Name	Type	Default	Description
`name`	`string`	`fn.name`	Span name
`sessionId`	`string`	—	Associate trace with session
`userId`	`string`	—	Associate trace with user
`metadata`	`Record<string, any>`	—	Trace metadata (JSON-serializable)
`tags`	`string[]`	—	Span tags (deduplicated)
`traceType`	`'DEFAULT'` \| `'EVALUATION'`	`'DEFAULT'`	Trace type
`spanType`	`'DEFAULT'` \| `'LLM'` \| `'TOOL'` \| `'EXECUTOR'` \| `'EVALUATOR'` \| `'EVALUATION'`	`'DEFAULT'`	Span type
`input`	`any`	—	Explicit input (overrides function args)
`ignoreInput`	`boolean`	`false`	Don’t record input
`ignoreOutput`	`boolean`	`false`	Don’t record output
`parentSpanContext`	`string` \| `LaminarSpanContext`	—	Parent context for distributed tracing
`rolloutEntrypoint`	`boolean`	`false`	Return a wrapped function for rollout sessions instead of executing immediately

Returns: Promise<ReturnType<F>> (default) or wrapped function when rolloutEntrypoint: trueBehavior:

If one non-Map object argument, records that object as input; otherwise records array
Maps converted to entries, circular references become "[Circular reference]"
Exceptions recorded via recordException and re-thrown
Anonymous function name becomes "" if name not provided
Works with both sync and async functions
When rolloutEntrypoint is true, returns a wrapped function and does not execute immediately
In rollout sessions, Laminar attaches rollout.session_id from LMNR_ROLLOUT_SESSION_ID

Examples: Rollout entrypoint

import { observe } from '@lmnr-ai/lmnr';

export const generateStream = observe(
  { name: 'generateStream', rolloutEntrypoint: true },
  async (messages: string[]) => {
    // Your agent code here.
  },
);
// Start a session with: npx lmnr-cli dev path/to/entrypoint.js

Examples: Group multiple LLM calls under one parent span

import { observe } from '@lmnr-ai/lmnr';

export const handle = async (userMessage: string) =>
  observe({ name: 'requestHandler' }, async () => {
    const routerResponse = await openai.chat.completions.create({
      model: 'gpt-4o-mini',
      messages: [{ role: 'user', content: `Route this: ${userMessage}` }],
    });

    const modelResponse = await openai.chat.completions.create({
      model: 'gpt-4o',
      messages: [{ role: 'user', content: userMessage }],
    });

    return {
      route: routerResponse.choices[0].message.content,
      answer: modelResponse.choices[0].message.content,
    };
  });

Examples: Nested spans for multi-step workflows

import { observe } from '@lmnr-ai/lmnr';

await observe({ name: 'checkout', sessionId: `checkout-${userId}` }, async () => {
  await observe({ name: 'validateCart' }, async () => {
    // ...
  });

  await observe({ name: 'processPayment' }, async () => {
    // ...
  });

  await observe({ name: 'createOrder' }, async () => {
    // ...
  });
});

observeDecorator(config)

Class method decorator version of observe.

import { observeDecorator } from '@lmnr-ai/lmnr';

class MyService {
  @observeDecorator({ name: 'process_request' })
  async processRequest(input: string) {
    return await this.doWork(input);
  }
}

Parameters: Same as observe options. Config can be an object or a function (thisArg, ...args) => config for per-call configuration.Returns: Method decorator

@observe()

Decorate a function to create a span. Automatically captures inputs, outputs, and errors.

from lmnr import observe

@observe()
def my_function(param1, param2):
    return param1 + param2

my_function(1, 2)
# Span input: {"param1": 1, "param2": 2}
# Span output: 3

Parameters:

Name	Type	Default	Description
`name`	`str`	Function name	Span name
`session_id`	`str`	`None`	Associate trace with session
`user_id`	`str`	`None`	Associate trace with user
`metadata`	`dict`	`None`	Trace metadata (JSON-serializable)
`tags`	`list[str]`	`None`	Span tags
`span_type`	`'DEFAULT'` \| `'LLM'` \| `'TOOL'`	`'DEFAULT'`	Span type
`ignore_input`	`bool`	`False`	Don’t record input
`ignore_output`	`bool`	`False`	Don’t record output
`ignore_inputs`	`list[str]`	`None`	Specific parameters to skip
`input_formatter`	`Callable`	`None`	Custom input formatting function
`output_formatter`	`Callable`	`None`	Custom output formatting function
`preserve_global_context`	`bool`	`False`	Use global OTEL context instead of isolated
`rollout_entrypoint`	`bool`	`False`	Register function as rollout entrypoint when running in rollout mode

Returns: DecoratorBehavior:

Function arguments serialized via orjson to lmnr.span.input
Return value serialized to lmnr.span.output
Inputs/outputs >10MB are truncated with warning message
Exceptions recorded as span events with ERROR status
Generators/async generators: yielded parts aggregated for output
If Laminar not initialized, returns original function unchanged
Async functions supported automatically

Examples: Group multiple LLM calls under one parent span

from lmnr import observe

@observe()
def handle(user_message: str) -> dict:
    router = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Route this: {user_message}"}],
    )

    model = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_message}],
    )

    return {
        "route": router.choices[0].message.content,
        "answer": model.choices[0].message.content,
    }

Examples: Nested spans for multi-step workflows

from lmnr import Laminar, observe

@observe()
def checkout(user_id: str):
    Laminar.set_trace_session_id(f"checkout-{user_id}")
    validate_cart()
    process_payment()
    create_order()

@observe()
def validate_cart():
    # ...
    pass

@observe()
def process_payment():
    # ...
    pass

@observe()
def create_order():
    # ...
    pass

Examples: Rollout entrypoint

from lmnr import observe

@observe(rollout_entrypoint=True)
def generate_stream(messages):
    # Your agent code here.
    pass
# Start a session with: lmnr dev path/to/entrypoint.py

Core

Instrumentation

Types

observe(options, fn, …args)

observeDecorator(config)

@observe()

Core

Instrumentation

Types

​observe(options, fn, …args)

​observeDecorator(config)

​@observe()

observe(options, fn, …args)

observeDecorator(config)

@observe()