callbacks

type CallbackInput ¶

type CallbackInput = callbacks.CallbackInput

CallbackInput is the value passed to OnStart and OnStartWithStreamInput handlers. The concrete type is defined by the component — for example, ChatModel callbacks carry *model.CallbackInput. Use the component package's ConvCallbackInput helper (e.g. model.ConvCallbackInput) to cast safely; it returns nil if the type does not match, so you can ignore irrelevant component types:

modelInput := model.ConvCallbackInput(in)
if modelInput == nil {
    return ctx // not a model invocation, skip
}
log.Printf("prompt: %v", modelInput.Messages)

type CallbackOutput ¶

type CallbackOutput = callbacks.CallbackOutput

CallbackOutput is the value passed to OnEnd and OnEndWithStreamOutput handlers. Like CallbackInput, the concrete type is component-defined. Use the component package's ConvCallbackOutput helper to cast safely.

type CallbackTiming ¶

type CallbackTiming = callbacks.CallbackTiming

CallbackTiming enumerates the lifecycle moments at which a callback handler is invoked. Implement TimingChecker on your handler and return false for timings you do not handle, so the framework skips the overhead of stream copying and goroutine spawning for those timings.

const (
	// TimingOnStart fires just before the component begins processing.
	// Receives a fully-formed input value (non-streaming).
	TimingOnStart CallbackTiming = iota
	// TimingOnEnd fires after the component returns a result successfully.
	// Receives the output value. Only fires on success — not on error.
	TimingOnEnd
	// TimingOnError fires when the component returns a non-nil error.
	// Stream errors (mid-stream panics) are NOT reported here; they surface
	// as errors inside the stream reader.
	TimingOnError
	// TimingOnStartWithStreamInput fires when the component receives a
	// streaming input (Collect / Transform paradigms). The handler receives a
	// copy of the input stream and must close it after reading.
	TimingOnStartWithStreamInput
	// TimingOnEndWithStreamOutput fires after the component returns a
	// streaming output (Stream / Transform paradigms). The handler receives a
	// copy of the output stream and must close it after reading. This is
	// typically where you implement streaming metrics or logging.
	TimingOnEndWithStreamOutput
)

type Handler ¶

type Handler = callbacks.Handler

Handler is the unified callback handler interface. Implement all five methods (OnStart, OnEnd, OnError, OnStartWithStreamInput, OnEndWithStreamOutput) or use NewHandlerBuilder to set only the timings you care about.

Each method receives the context returned by the previous timing of the SAME handler, which lets a single handler pass state between its OnStart and OnEnd calls via context.WithValue. There is NO guaranteed execution order between DIFFERENT handlers, and the context chain does not flow from one handler to the next — do not rely on handler ordering.

Implement TimingChecker (the Needed method) on your handler so the framework can skip timings you have not registered; this avoids unnecessary stream copies and goroutine allocations on every component invocation.

Stream handlers (OnStartWithStreamInput, OnEndWithStreamOutput) receive a *schema.StreamReader that has already been copied; they MUST close their copy after reading. If any handler's copy is not closed, the original stream cannot be freed, causing a goroutine/memory leak for the entire pipeline.

Important: do NOT mutate the Input or Output values. All downstream nodes and handlers share the same pointer (direct assignment, not a deep copy). Mutations cause data races in concurrent graph execution.

type HandlerBuilder ¶

type HandlerBuilder struct {
	// contains filtered or unexported fields
}

HandlerBuilder constructs a Handler by registering callback functions for individual timings. Only set the timings you care about; the built handler implements TimingChecker and returns false for unregistered timings, so the framework skips those timings with no overhead.

The input/output values are untyped (CallbackInput / CallbackOutput). To work with a specific component's payload, use the component package's ConvCallbackInput / ConvCallbackOutput helpers inside your function. For a higher-level API that dispatches by component type automatically, see utils/callbacks.NewHandlerHelper.

Example:

handler := callbacks.NewHandlerBuilder().
    OnStartFn(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context {
        mi := model.ConvCallbackInput(input)
        if mi != nil {
            log.Printf("[%s] model start: %d messages", info.Name, len(mi.Messages))
        }
        return ctx
    }).
    OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        mo := model.ConvCallbackOutput(output)
        if mo != nil && mo.Message.ResponseMeta != nil {
            log.Printf("[%s] tokens: %d", info.Name, mo.Message.ResponseMeta.Usage.TotalTokens)
        }
        return ctx
    }).
    Build()

func NewHandlerBuilder() *HandlerBuilder

func (hb *HandlerBuilder) Build() Handler

func (hb *HandlerBuilder) OnEndFn(
	fn func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context) *HandlerBuilder

func (hb *HandlerBuilder) OnEndWithStreamOutputFn(
	fn func(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context) *HandlerBuilder

func (hb *HandlerBuilder) OnErrorFn(
	fn func(ctx context.Context, info *RunInfo, err error) context.Context) *HandlerBuilder

func (hb *HandlerBuilder) OnStartFn(
	fn func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context) *HandlerBuilder

func (hb *HandlerBuilder) OnStartWithStreamInputFn(
	fn func(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context) *HandlerBuilder

type RunInfo ¶

type RunInfo = callbacks.RunInfo

RunInfo describes the entity that triggered a callback. Always nil-check before dereferencing — a component that calls OnStart without first calling EnsureRunInfo or InitCallbacks will leave RunInfo absent in the context.

Fields:

• Name: business-meaningful name specified by the user. For nodes in a graph this is the node name (compose.WithNodeName). For standalone components it must be set explicitly via InitCallbacks or ReuseHandlers; it is empty string if not set.
• Type: implementation identity, e.g. "OpenAI". Set by the component via components.Typer; falls back to reflection (struct/func name) if the interface is not implemented. Empty for Graph itself.
• Component: category constant, e.g. components.ComponentOfChatModel. Fixed value "Lambda" for lambdas, "Graph"/"Chain"/"Workflow" for graphs. Use this to branch on component kind without caring about implementation.

Handlers should filter using RunInfo rather than assuming a fixed execution order — there is no guaranteed ordering between different Handlers.

type TimingChecker ¶

type TimingChecker = callbacks.TimingChecker

TimingChecker is an optional interface for Handler implementations. When a handler implements Needed, the framework calls it before each component invocation to decide whether to set up callback infrastructure (stream copying, goroutine allocation) for that timing. Returning false avoids unnecessary overhead.

Handlers built with NewHandlerBuilder or utils/callbacks.NewHandlerHelper automatically implement TimingChecker based on which callback functions were set.