Skip to content

chore(observability): migrate Langfuse observability to v4 OTLP ingestion #490

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Mike3285 wants to merge 4 commits into
base: main
Choose a base branch
from
Open

chore(observability): migrate Langfuse observability to v4 OTLP ingestion #490

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1d96f4c
feat(langfuse): migrate observability client and ingestion to Langfus…
Mike3285 Nov 10, 2026
0f75eaf
fix(controller): propagate Langfuse v4 trace context and root span in…
Mike3285 Nov 10, 2026
67b8613
chore(compose): update default Langfuse web and worker images to v4
Mike3285 Nov 10, 2026
09348d3
fix(compose): complete Langfuse v4 infrastructure and migration confi…
Mike3285 Nov 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions .env.example
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,19 @@ NEO4J_IMAGE=neo4j:5.26.2
GRAPHITI_IMAGE=vxcontrol/graphiti:latest

## Langfuse stack images (docker-compose-langfuse.yml)
LANGFUSE_WORKER_IMAGE=langfuse/langfuse-worker:3
LANGFUSE_WEB_IMAGE=langfuse/langfuse:3
CLICKHOUSE_IMAGE=clickhouse/clickhouse-server:24
LANGFUSE_WORKER_IMAGE=langfuse/langfuse-worker:4
LANGFUSE_WEB_IMAGE=langfuse/langfuse:4
CLICKHOUSE_IMAGE=clickhouse/clickhouse-server:26.4
MINIO_IMAGE=cgr.dev/chainguard/minio:latest
REDIS_IMAGE=redis:7
POSTGRES_IMAGE=postgres:16

## Langfuse v4 defaults; see backend/docs/langfuse.md before upgrading existing data.
LANGFUSE_MIGRATION_V4_WRITE_MODE=events_only
LANGFUSE_MIGRATION_V4_NATIVE_OTEL_BEHAVIOUR=direct
LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN=true
LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL=true

## Observability stack images (docker-compose-observability.yml)
GRAFANA_IMAGE=grafana/grafana:11.4.0
NODE_EXPORTER_IMAGE=prom/node-exporter:v1.8.2
Expand Down
8 changes: 8 additions & 0 deletions backend/docs/langfuse.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,14 @@ This document provides a comprehensive guide to the Langfuse integration in Pent

## Introduction

### Upgrading the self-hosted stack to Langfuse v4

The Compose stack uses Langfuse v4 and ClickHouse 26.4. Before upgrading an existing deployment, follow the [official v3 to v4 upgrade guide](https://langfuse.com/self-hosting/upgrade/upgrade-guides/upgrade-v3-to-v4): back up your data, upgrade ClickHouse first while still running Langfuse v3, and verify the required ClickHouse permissions. Update existing `.env` image overrides as well; changing Compose defaults does not replace them.

The default configuration uses `events_only`, `direct` OTel ingestion, the new read experience, and historic backfill enabled. This matches v4 defaults and the PentAGI OTLP client. Backfill requires approximately three times the current ClickHouse data volume in available capacity.

For a staged migration with older producers or legacy API consumers, set `LANGFUSE_MIGRATION_V4_WRITE_MODE=dual`, `LANGFUSE_MIGRATION_V4_NATIVE_OTEL_BEHAVIOUR=dual_write`, and `LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL=false`. Keep `LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN=true` to allow the new read experience. Enable backfill after dual writing is confirmed healthy, then switch to `events_only` and `direct` after migrating producers, consumers, and historic data. Do not leave backfill disabled unless intentionally using the retention-based rollover described in the official guide.

Langfuse is an open-source observability platform specifically designed for LLM-powered applications. The PentAGI integration with Langfuse provides:

- **Comprehensive tracing** for AI agent flows and tasks
Expand Down
88 changes: 49 additions & 39 deletions backend/pkg/controller/assistant.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,27 +180,29 @@ func buildAssistantWorker(
pub.AssistantUpdated(cleanupCtx, failed)
}()

ctx, observation := obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow %d assistant worker", awc.cfg.TenantLabel(), awc.flowID, assistant.ID)),
langfuse.WithTraceUserID(tenantUserID(awc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(awc.cfg, "controller", "assistant")),
langfuse.WithTraceInput(awc.input),
langfuse.WithTraceSessionID(awc.cfg.ScopedName(fmt.Sprintf("assistant-%d-flow-%d", assistant.ID, awc.flowID))),
langfuse.WithTraceMetadata(tenantMeta(awc.cfg, langfuse.Metadata{
"assistant_id": assistant.ID,
"flow_id": awc.flowID,
"user_id": awc.userID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": awc.prvname.String(),
"provider_type": awc.prvtype.String(),
})),
),
traceCtxOpt := langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow %d assistant worker", awc.cfg.TenantLabel(), awc.flowID, assistant.ID)),
langfuse.WithTraceUserID(tenantUserID(awc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(awc.cfg, "controller", "assistant")),
langfuse.WithTraceInput(awc.input),
langfuse.WithTraceSessionID(awc.cfg.ScopedName(fmt.Sprintf("assistant-%d-flow-%d", assistant.ID, awc.flowID))),
langfuse.WithTraceMetadata(tenantMeta(awc.cfg, langfuse.Metadata{
"assistant_id": assistant.ID,
"flow_id": awc.flowID,
"user_id": awc.userID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": awc.prvname.String(),
"provider_type": awc.prvtype.String(),
})),
)
ctx, observation := obs.Observer.NewObservation(ctx, traceCtxOpt)
assistantSpan := observation.Span(
langfuse.WithSpanName("prepare assistant worker"),
langfuse.WithSpanInput(awc.input),
)
assistantSpan := observation.Span(langfuse.WithSpanName("prepare assistant worker"))
ctx, _ = assistantSpan.Observation(ctx)

aslw, err := awc.aslc.NewFlowAssistantLog(ctx, awc.flowID, assistant.ID, pub)
Expand Down Expand Up @@ -274,7 +276,10 @@ func buildAssistantWorker(
executor.SetGraphitiClient(awc.provs.GraphitiClient())

ctx, cancel := context.WithCancel(context.Background())
ctx, _ = obs.Observer.NewObservation(ctx, langfuse.WithObservationTraceID(observation.TraceID()))
ctx, _ = obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceID(observation.TraceID()),
traceCtxOpt,
)
aw := &assistantWorker{
id: assistant.ID,
flowID: awc.flowID,
Expand Down Expand Up @@ -364,24 +369,26 @@ func LoadAssistantWorker(
return nil, fmt.Errorf("failed to get flow primary container: %w", err)
}

traceCtxOpt := langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow %d assistant worker", awc.cfg.TenantLabel(), awc.flowID, assistant.ID)),
langfuse.WithTraceUserID(tenantUserID(awc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(awc.cfg, "controller", "assistant")),
langfuse.WithTraceSessionID(awc.cfg.ScopedName(fmt.Sprintf("assistant-%d-flow-%d", assistant.ID, awc.flowID))),
langfuse.WithTraceMetadata(tenantMeta(awc.cfg, langfuse.Metadata{
"assistant_id": assistant.ID,
"flow_id": awc.flowID,
"user_id": awc.userID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": assistant.ModelProviderName,
"provider_type": assistant.ModelProviderType,
})),
)
ctx, observation := obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow %d assistant worker", awc.cfg.TenantLabel(), awc.flowID, assistant.ID)),
langfuse.WithTraceUserID(tenantUserID(awc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(awc.cfg, "controller", "assistant")),
langfuse.WithTraceSessionID(awc.cfg.ScopedName(fmt.Sprintf("assistant-%d-flow-%d", assistant.ID, awc.flowID))),
langfuse.WithTraceMetadata(tenantMeta(awc.cfg, langfuse.Metadata{
"assistant_id": assistant.ID,
"flow_id": awc.flowID,
"user_id": awc.userID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": assistant.ModelProviderName,
"provider_type": assistant.ModelProviderType,
})),
),
langfuse.WithObservationTraceID(assistant.TraceID.String),
traceCtxOpt,
)
assistantSpan := observation.Span(langfuse.WithSpanName("prepare assistant worker"))
ctx, _ = assistantSpan.Observation(ctx)
Expand Down Expand Up @@ -445,7 +452,10 @@ func LoadAssistantWorker(
}

ctx, cancel := context.WithCancel(context.Background())
ctx, _ = obs.Observer.NewObservation(ctx, langfuse.WithObservationTraceID(observation.TraceID()))
ctx, _ = obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceID(observation.TraceID()),
traceCtxOpt,
)
aw := &assistantWorker{
id: assistant.ID,
flowID: awc.flowID,
Expand Down
83 changes: 46 additions & 37 deletions backend/pkg/controller/flow.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,26 +188,28 @@ func buildFlowWorker(
return nil, fmt.Errorf("failed to get user %d: %w", fwc.userID, err)
}

ctx, observation := obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow worker", fwc.cfg.TenantLabel(), flow.ID)),
langfuse.WithTraceUserID(tenantUserID(fwc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(fwc.cfg, "controller", "flow")),
langfuse.WithTraceInput(fwc.input),
langfuse.WithTraceSessionID(fwc.cfg.ScopedName(fmt.Sprintf("flow-%d", flow.ID))),
langfuse.WithTraceMetadata(tenantMeta(fwc.cfg, langfuse.Metadata{
"flow_id": flow.ID,
"user_id": fwc.userID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": fwc.prvname.String(),
"provider_type": fwc.prvtype.String(),
})),
),
traceCtxOpt := langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow worker", fwc.cfg.TenantLabel(), flow.ID)),
langfuse.WithTraceUserID(tenantUserID(fwc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(fwc.cfg, "controller", "flow")),
langfuse.WithTraceInput(fwc.input),
langfuse.WithTraceSessionID(fwc.cfg.ScopedName(fmt.Sprintf("flow-%d", flow.ID))),
langfuse.WithTraceMetadata(tenantMeta(fwc.cfg, langfuse.Metadata{
"flow_id": flow.ID,
"user_id": fwc.userID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": fwc.prvname.String(),
"provider_type": fwc.prvtype.String(),
})),
)
ctx, observation := obs.Observer.NewObservation(ctx, traceCtxOpt)
flowSpan := observation.Span(
langfuse.WithSpanName("prepare flow worker"),
langfuse.WithSpanInput(fwc.input),
)
flowSpan := observation.Span(langfuse.WithSpanName("prepare flow worker"))
ctx, _ = flowSpan.Observation(ctx)

prompter, err := newUserPrompter(ctx, fwc.db, fwc.userID)
Expand Down Expand Up @@ -277,7 +279,10 @@ func buildFlowWorker(
Screenshot: workers.sw,
}
ctx, cancel := context.WithCancel(context.Background())
ctx, _ = obs.Observer.NewObservation(ctx, langfuse.WithObservationTraceID(observation.TraceID()))
ctx, _ = obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceID(observation.TraceID()),
traceCtxOpt,
)
fw := &flowWorker{
tc: NewTaskController(flowCtx),
wg: &sync.WaitGroup{},
Expand Down Expand Up @@ -367,24 +372,25 @@ func LoadFlowWorker(ctx context.Context, flow database.Flow, fwc flowWorkerCtx)
return nil, fmt.Errorf("failed to get user %d: %w", flow.UserID, err)
}

traceCtxOpt := langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow worker", fwc.cfg.TenantLabel(), flow.ID)),
langfuse.WithTraceUserID(tenantUserID(fwc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(fwc.cfg, "controller", "flow")),
langfuse.WithTraceSessionID(fwc.cfg.ScopedName(fmt.Sprintf("flow-%d", flow.ID))),
langfuse.WithTraceMetadata(tenantMeta(fwc.cfg, langfuse.Metadata{
"flow_id": flow.ID,
"user_id": flow.UserID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": flow.ModelProviderName,
"provider_type": flow.ModelProviderType,
})),
)
ctx, observation := obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceID(flow.TraceID.String),
langfuse.WithObservationTraceContext(
langfuse.WithTraceName(fmt.Sprintf("%s%d flow worker", fwc.cfg.TenantLabel(), flow.ID)),
langfuse.WithTraceUserID(tenantUserID(fwc.cfg, user.Mail)),
langfuse.WithTraceTags(tenantTags(fwc.cfg, "controller", "flow")),
langfuse.WithTraceSessionID(fwc.cfg.ScopedName(fmt.Sprintf("flow-%d", flow.ID))),
langfuse.WithTraceMetadata(tenantMeta(fwc.cfg, langfuse.Metadata{
"flow_id": flow.ID,
"user_id": flow.UserID,
"user_email": user.Mail,
"user_name": user.Name,
"user_hash": user.Hash,
"user_role": user.RoleName,
"provider_name": flow.ModelProviderName,
"provider_type": flow.ModelProviderType,
})),
),
traceCtxOpt,
)
flowSpan := observation.Span(langfuse.WithSpanName("prepare flow worker"))
ctx, _ = flowSpan.Observation(ctx)
Expand Down Expand Up @@ -445,7 +451,10 @@ func LoadFlowWorker(ctx context.Context, flow database.Flow, fwc flowWorkerCtx)
Screenshot: workers.sw,
}
ctx, cancel := context.WithCancel(context.Background())
ctx, _ = obs.Observer.NewObservation(ctx, langfuse.WithObservationTraceID(observation.TraceID()))
ctx, _ = obs.Observer.NewObservation(ctx,
langfuse.WithObservationTraceID(observation.TraceID()),
traceCtxOpt,
)
fw := &flowWorker{
tc: NewTaskController(flowCtx),
wg: &sync.WaitGroup{},
Expand Down
Loading