langfuse

README

Eino callback for Langfuse

English | 简体中文

A CloudWeGo Eino callback that exports traces to Langfuse through native OpenTelemetry OTLP/HTTP ingestion.

This package targets Langfuse v4's observations-first data model. It sends one complete immutable OTEL span per Eino operation to /api/public/otel/v1/traces, uses Basic Auth, and includes x-langfuse-ingestion-version: 4 for real-time ingestion. It does not use the deprecated /api/public/ingestion event API.

Features

• Eino chat models become Langfuse generation observations.
• Eino ADK agents, tools, retrievers, embeddings, and prompts receive specific Langfuse observation types.
• Model name, parameters, token usage, streaming output, and completion start time are captured.
• ADK async event iterators remain open until the agent has actually completed.
• Parent/child relationships use standard OpenTelemetry context propagation.
• Same-name Eino ChatModelAgent implementation Chains can be collapsed with an opt-in setting.
• Trace name, user, session, tags, release, environment, version, and metadata are propagated to child observations for Langfuse v4.
• Input/output masking and size limits are configurable.
• OTLP batching, sampling, custom exporters, custom HTTP clients, flush, and graceful shutdown are supported.

Install

go get github.com/cloudwego/eino-ext/callbacks/langfuse/v2

The callback uses Go 1.23+, Eino 0.9.14, and OpenTelemetry Go 1.38.

Usage

package main

import (
	"context"
	"os"

	"github.com/cloudwego/eino/callbacks"
	langfuse "github.com/cloudwego/eino-ext/callbacks/langfuse/v2"
)

func main() {
	ctx := context.Background()
	handler, err := langfuse.NewHandler(ctx, &langfuse.Config{
		Host:        os.Getenv("LANGFUSE_HOST"),
		PublicKey:   os.Getenv("LANGFUSE_PUBLIC_KEY"),
		SecretKey:   os.Getenv("LANGFUSE_SECRET_KEY"),
		ServiceName: "my-eino-service",
		Environment: "production",
	})
	if err != nil {
		panic(err)
	}
	defer handler.Shutdown(ctx)

	callbacks.AppendGlobalHandlers(handler)

	traceCtx := handler.StartTrace(ctx,
		langfuse.WithName("chat-response"),
		langfuse.WithUserID("user-123"),
		langfuse.WithSessionID("conversation-456"),
		langfuse.WithInput("Hello"),
		langfuse.WithObservationType(langfuse.ObservationTypeAgent),
	)

	// Run Eino components with traceCtx. Their callbacks become child spans.

	handler.EndTrace(traceCtx, "Hello! How can I help?")
}

Config.Host accepts a Langfuse base URL, a URL ending in /api/public/otel, or the full /api/public/otel/v1/traces endpoint.

WithName sets both the Langfuse trace name and the application root observation's span name.

Discarded telemetry is never silent. Sampling, a full asynchronous queue, callbacks received after processor shutdown, callback and OTel attribute limits, metadata/payload serialization failures, and OTel event/link limits are aggregated into one warning per minute:

langfuse callback discarded telemetry since previous report (interval 1m0s): queue_full_spans=37 value_limit_truncations=2

Set Config.DropLogInterval to change the reporting interval. A negative value logs each discard immediately. Export failures are always logged immediately with the affected batch size because the processor will not retry the batch after the exporter returns its final error.

Set Config.ExportDiagnostics to append the serialized protobuf size, gzip request-body size, HTTP attempt count, and DNS/connect/TLS/write/header-wait timings to final export failure logs. Successful exports remain silent. The diagnostics never log span content or trace IDs, and inspect the first request body only, so enabling them has a small CPU and allocation cost.

The batch processor does not add a shorter export deadline. For the built-in OTLP exporter, Config.Timeout limits each HTTP request and the OpenTelemetry exporter's retry policy controls the total retry window. Explicit ForceFlush operations pass the caller's context to the exporter and remain cancellable.

These processor-level diagnostics apply when the callback creates its own OTel tracer provider. With a caller-owned TracerProvider, queueing and exporting are owned by that provider and must be diagnosed by its span processors/exporters; callback-level truncation and serialization diagnostics still apply.

Call EndTrace when the root operation completes. It records the final output and waits for active Eino child callbacks to finish before ending the root. StartTrace also watches the supplied context: when it is cancelled or reaches its deadline, the root waits for active child callbacks and then ends automatically. A non-cancellable context such as context.Background() requires EndTrace; Shutdown closes any roots that remain active during process shutdown.

User-initiated context.Canceled callbacks are exported with the default Langfuse level, a cancelled status message, and cancellation metadata instead of being counted as errors. context.DeadlineExceeded and other callback failures remain errors.

Resumable Eino tool, graph, subgraph, and ADK business interrupts are exported with the default Langfuse level and an interrupted status instead of ERROR. The interrupt cause remains available in structured observation output and metadata. OTel instrumentation scope includes the callback release version so traces can distinguish callback upgrades from the application release.

Process resource attributes are excluded by default to keep observation metadata small and avoid exposing local owners, executable paths, or command arguments. Set Config.IncludeProcessResourceAttributes only when those diagnostics are needed. Service name, OTel SDK identity, and instrumentation scope remain present.

Set Config.CollapseAgentInternalSpans to remove the same-name internal Chain created beneath an Eino ChatModelAgent's Agent observation, its internal ReAct Graph and Init Lambda, and anonymous or default-named Lambda wrappers. It is disabled by default to preserve complete framework traces. These names are only collapsed beneath an active Agent; matching components elsewhere and named business Lambdas are retained. Descendant generations, tools, and sub-agents remain attached to the nearest retained parent through standard OTel context propagation.

MaxAttributeValueLength applies a JSON-safe per-value limit. MaxSpanAttributeBytes limits the combined callback-owned attribute keys and values on each span. JSON inputs and outputs remain valid after either limit is applied.

Use MaskFunc before exporting production inputs and outputs that may contain secrets or personal data.

Migrating from v1

The v1 callback remains available at github.com/cloudwego/eino-ext/callbacks/langfuse. The v2 module uses Langfuse's OTLP ingestion API and intentionally replaces the legacy NewLangfuseHandler, SetTrace, and UpdateTraceOutput lifecycle with NewHandler, StartTrace, EndTrace, Flush, and Shutdown. OTLP spans are immutable after export, so retain the context returned by StartTrace and pass the final output to EndTrace.

Compatibility notes

Langfuse supports OTLP over HTTP/protobuf and HTTP/JSON, but not OTLP/gRPC. This package uses HTTP/protobuf with gzip. The tracing transport is standard OTLP; the langfuse.* span attributes are Langfuse's vendor-specific semantic mapping.

This repository focuses on runtime Eino instrumentation. Langfuse management APIs such as prompts, datasets, scores, projects, and API keys are outside its scope.

License

Apache-2.0

Documentation

Overview

Package langfuse converts CloudWeGo Eino callbacks into Langfuse-compatible OpenTelemetry spans and exports them over OTLP/HTTP.

Index

• type CallbackHandler
  • func NewHandler(ctx context.Context, cfg *Config) (*CallbackHandler, error)
  • func (c *CallbackHandler) EndTrace(ctx context.Context, output string)
  • func (c *CallbackHandler) Flush(ctx context.Context) error
  • func (c *CallbackHandler) Needed(ctx context.Context, info *callbacks.RunInfo, _ callbacks.CallbackTiming) bool
  • func (c *CallbackHandler) OnEnd(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context
  • func (c *CallbackHandler) OnEndWithStreamOutput(ctx context.Context, info *callbacks.RunInfo, ...) context.Context
  • func (c *CallbackHandler) OnError(ctx context.Context, info *callbacks.RunInfo, callbackErr error) context.Context
  • func (c *CallbackHandler) OnStart(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context
  • func (c *CallbackHandler) OnStartWithStreamInput(ctx context.Context, info *callbacks.RunInfo, ...) context.Context
  • func (c *CallbackHandler) Shutdown(ctx context.Context) error
  • func (c *CallbackHandler) StartTrace(ctx context.Context, opts ...TraceOption) context.Context
• type Config
• type ObservationType
• type TraceOption
  • func WithEnvironment(environment string) TraceOption
  • func WithID(id string) TraceOption
  • func WithInput(input string) TraceOption
  • func WithMetadata(metadata map[string]string) TraceOption
  • func WithMetadataValues(metadata map[string]any) TraceOption
  • func WithName(name string) TraceOption
  • func WithObservationType(observationType ObservationType) TraceOption
  • func WithPublic(public bool) TraceOption
  • func WithRelease(release string) TraceOption
  • func WithSessionID(sessionID string) TraceOption
  • func WithTags(tags ...string) TraceOption
  • func WithUserID(userID string) TraceOption
  • func WithVersion(version string) TraceOption

Types

type CallbackHandler

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

CallbackHandler converts Eino callbacks into Langfuse-compatible OTEL spans.

func NewHandler

func NewHandler(ctx context.Context, cfg *Config) (*CallbackHandler, error)

NewHandler creates an Eino callback that exports OTLP/HTTP traces using the Langfuse v4 ingestion path.

func (*CallbackHandler) EndTrace

func (c *CallbackHandler) EndTrace(ctx context.Context, output string)

EndTrace records the final root-observation output and requests the trace to end. The root waits for active Eino child callbacks before it is exported. Calling EndTrace more than once is safe.

func (*CallbackHandler) Flush

func (c *CallbackHandler) Flush(ctx context.Context) error

Flush waits for callback-owned stream collectors and flushes completed spans.

func (*CallbackHandler) Needed

func (c *CallbackHandler) Needed(ctx context.Context, info *callbacks.RunInfo, _ callbacks.CallbackTiming) bool

Needed implements callbacks.TimingChecker. When explicitly enabled, it removes Eino Agent implementation details while preserving named business components and their standard context-based parent relationships.

func (*CallbackHandler) OnEnd

func (c *CallbackHandler) OnEnd(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context

func (*CallbackHandler) OnEndWithStreamOutput

func (c *CallbackHandler) OnEndWithStreamOutput(ctx context.Context, info *callbacks.RunInfo, output *schema.StreamReader[callbacks.CallbackOutput]) context.Context

func (*CallbackHandler) OnError

func (c *CallbackHandler) OnError(ctx context.Context, info *callbacks.RunInfo, callbackErr error) context.Context

func (*CallbackHandler) OnStart

func (c *CallbackHandler) OnStart(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context

func (*CallbackHandler) OnStartWithStreamInput

func (c *CallbackHandler) OnStartWithStreamInput(ctx context.Context, info *callbacks.RunInfo, input *schema.StreamReader[callbacks.CallbackInput]) context.Context

func (*CallbackHandler) Shutdown

func (c *CallbackHandler) Shutdown(ctx context.Context) error

Shutdown completes active root observations, drains stream collectors, and shuts down the callback-owned provider.

func (*CallbackHandler) StartTrace

func (c *CallbackHandler) StartTrace(ctx context.Context, opts ...TraceOption) context.Context

StartTrace creates the application root observation. It is ended explicitly by EndTrace, when ctx is done and all active Eino child callbacks have ended, or when the handler shuts down.

type Config

type Config struct {
	// Host accepts a Langfuse base URL, an OTLP base URL ending in
	// /api/public/otel, or the full /api/public/otel/v1/traces endpoint.
	Host string

	PublicKey string
	SecretKey string

	ServiceName string
	Environment string
	Release     string
	Version     string
	Tags        []string
	Timeout     time.Duration
	SampleRate  float64

	// Name, UserID, and SessionID configure default root trace attributes.
	// Prefer the corresponding StartTrace options for each trace.
	Name      string
	UserID    string
	SessionID string
	Public    bool

	// IncludeProcessResourceAttributes adds OTel process.* attributes such as
	// PID, owner, executable path, command arguments, and runtime details. It is
	// disabled by default to reduce metadata noise and accidental disclosure.
	IncludeProcessResourceAttributes bool
	// CollapseAgentInternalSpans suppresses implementation-detail spans emitted
	// by Eino agents: the same-name Chain beneath an Agent, the ReAct Graph and
	// Init Lambda it owns, and unnamed/default-named Lambda wrappers. It is
	// disabled by default so the callback preserves complete framework traces
	// unless the caller explicitly opts into a compact hierarchy. Matching
	// components outside an Agent and named business Lambdas are always retained.
	CollapseAgentInternalSpans bool

	MaxQueueSize       int
	MaxExportBatchSize int
	BatchTimeout       time.Duration
	// DropLogInterval controls how often discarded telemetry is aggregated into
	// one warning. Zero uses one minute. A negative value logs each discard
	// immediately; export failures are always logged immediately.
	DropLogInterval time.Duration
	// MaxAttributeValueLength limits each serialized input, output, or metadata
	// value. Set a negative value to disable this callback-level limit.
	MaxAttributeValueLength int
	// MaxSpanAttributeBytes limits the combined keys and values written by this
	// callback to one span. Set a negative value to disable the span-level budget.
	MaxSpanAttributeBytes int
	MaskFunc              func(string) string
	HTTPClient            *http.Client
	// ExportDiagnostics adds protobuf and gzip payload sizes, HTTP attempt
	// counts, and network-stage timings to the final export failure log. It
	// never logs span contents or trace IDs. Collection has a small CPU and
	// allocation cost because the first compressed request body is inspected.
	ExportDiagnostics bool

	// SpanExporter is intended for custom transports and tests. When set, Host
	// and API keys are not used to construct an exporter.
	SpanExporter sdktrace.SpanExporter

	// TracerProvider lets callers supply an existing provider. The callback does
	// not shut down a caller-owned provider. Custom trace IDs requested through
	// WithID are only guaranteed when the callback creates the provider.
	TracerProvider *sdktrace.TracerProvider
}

Config configures the Langfuse OTLP/HTTP callback.

type ObservationType

type ObservationType string

const (
	ObservationTypeEvent      ObservationType = "event"
	ObservationTypeSpan       ObservationType = "span"
	ObservationTypeGeneration ObservationType = "generation"
	ObservationTypeAgent      ObservationType = "agent"
	ObservationTypeTool       ObservationType = "tool"
	ObservationTypeChain      ObservationType = "chain"
	ObservationTypeRetriever  ObservationType = "retriever"
	ObservationTypeEvaluator  ObservationType = "evaluator"
	ObservationTypeEmbedding  ObservationType = "embedding"
	ObservationTypeGuardrail  ObservationType = "guardrail"
)

type TraceOption

type TraceOption func(*traceOptions)

func WithEnvironment

func WithEnvironment(environment string) TraceOption

func WithID

func WithID(id string) TraceOption

func WithInput

func WithInput(input string) TraceOption

func WithMetadata

func WithMetadata(metadata map[string]string) TraceOption

func WithMetadataValues

func WithMetadataValues(metadata map[string]any) TraceOption

func WithName

func WithName(name string) TraceOption

func WithObservationType

func WithObservationType(observationType ObservationType) TraceOption

func WithPublic

func WithPublic(public bool) TraceOption

func WithRelease

func WithRelease(release string) TraceOption

func WithSessionID

func WithSessionID(sessionID string) TraceOption

func WithTags

func WithTags(tags ...string) TraceOption

func WithUserID

func WithUserID(userID string) TraceOption

func WithVersion

func WithVersion(version string) TraceOption