diff --git a/.env.example b/.env.example index cd9914b9a..b2ea2dba8 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/backend/docs/langfuse.md b/backend/docs/langfuse.md index b7d634cf8..2f3e9b8a1 100644 --- a/backend/docs/langfuse.md +++ b/backend/docs/langfuse.md @@ -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 diff --git a/backend/pkg/controller/assistant.go b/backend/pkg/controller/assistant.go index 81998c9d2..75e31846c 100644 --- a/backend/pkg/controller/assistant.go +++ b/backend/pkg/controller/assistant.go @@ -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) @@ -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, @@ -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) @@ -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, diff --git a/backend/pkg/controller/flow.go b/backend/pkg/controller/flow.go index 1fbd816e6..0682fd672 100644 --- a/backend/pkg/controller/flow.go +++ b/backend/pkg/controller/flow.go @@ -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) @@ -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{}, @@ -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) @@ -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{}, diff --git a/backend/pkg/observability/langfuse/agent.go b/backend/pkg/observability/langfuse/agent.go index 1cdb96cbe..f98d0d9ba 100644 --- a/backend/pkg/observability/langfuse/agent.go +++ b/backend/pkg/observability/langfuse/agent.go @@ -4,10 +4,9 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - "pentagi/pkg/observability/langfuse/api" - "github.com/vxcontrol/langchaingo/llms" ) @@ -41,7 +40,9 @@ type agent struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type AgentOption func(*agent) @@ -58,6 +59,12 @@ func withAgentParentObservationID(parentObservationID string) AgentOption { } } +func withAgentTraceContext(traceCtx *TraceContext) AgentOption { + return func(a *agent) { + a.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithAgentID sets on creation time func WithAgentID(id string) AgentOption { return func(a *agent) { @@ -151,65 +158,43 @@ func newAgent(observer enqueue, opts ...AgentOption) Agent { opt(a) } - obsCreate := &api.IngestionEvent{IngestionEventTen: &api.IngestionEventTen{ - ID: newSpanID(), - Timestamp: getTimeRefString(a.StartTime), - Type: api.IngestionEventTenType(ingestionCreateAgent).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(a.ObservationID), - TraceID: getStringRef(a.TraceID), - ParentObservationID: getStringRef(a.ParentObservationID), - Name: getStringRef(a.Name), - Metadata: a.Metadata, - Input: convertInput(a.Input, a.Tools), - Output: convertOutput(a.Output), - StartTime: a.StartTime, - EndTime: a.EndTime, - Level: a.Level.ToLangfuse(), - StatusMessage: a.Status, - Version: a.Version, - Model: a.Model, - ModelParameters: a.ModelParameters.ToLangfuse(), - }, - }} - - a.observer.enqueue(obsCreate) - return a } func (a *agent) End(opts ...AgentOption) { - id := a.ObservationID - startTime := a.StartTime - a.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(a) - } - - // preserve the original observation ID and start time - a.ObservationID = id - a.StartTime = startTime - - agentUpdate := &api.IngestionEvent{IngestionEventTen: &api.IngestionEventTen{ - ID: newSpanID(), - Timestamp: getTimeRefString(a.EndTime), - Type: api.IngestionEventTenType(ingestionCreateAgent).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(a.ObservationID), - Name: getStringRef(a.Name), - Metadata: a.Metadata, - Input: convertInput(a.Input, a.Tools), - Output: convertOutput(a.Output), - EndTime: a.EndTime, - Level: a.Level.ToLangfuse(), - StatusMessage: a.Status, - Version: a.Version, - Model: a.Model, - ModelParameters: a.ModelParameters.ToLangfuse(), - }, - }} + a.endOnce.Do(func() { + id := a.ObservationID + startTime := a.StartTime + a.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(a) + } + + // preserve the original observation ID and start time + a.ObservationID = id + a.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: a.TraceID, + ObservationID: a.ObservationID, + ParentObservationID: a.ParentObservationID, + Name: a.Name, + ObservationType: otelObservationTypeAgent, + StartTime: a.StartTime, + EndTime: a.EndTime, + Level: a.Level, + Status: a.Status, + Version: a.Version, + Input: convertInput(a.Input, a.Tools), + Output: convertOutput(a.Output), + Metadata: a.Metadata, + Model: a.Model, + ModelParameters: a.ModelParameters, + TraceCtx: a.traceCtx, + }) - a.observer.enqueue(agentUpdate) + a.observer.enqueue(otelSpan) + }) } func (a *agent) String() string { @@ -221,10 +206,15 @@ func (a *agent) MarshalJSON() ([]byte, error) { } func (a *agent) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := a.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: a.TraceID, ObservationID: a.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: a.observer, } diff --git a/backend/pkg/observability/langfuse/chain.go b/backend/pkg/observability/langfuse/chain.go index 377a3b1e9..21d12d94f 100644 --- a/backend/pkg/observability/langfuse/chain.go +++ b/backend/pkg/observability/langfuse/chain.go @@ -4,9 +4,8 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - - "pentagi/pkg/observability/langfuse/api" ) const ( @@ -36,7 +35,9 @@ type chain struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type ChainOption func(*chain) @@ -53,6 +54,12 @@ func withChainParentObservationID(parentObservationID string) ChainOption { } } +func withChainTraceContext(traceCtx *TraceContext) ChainOption { + return func(c *chain) { + c.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithChainID sets on creation time func WithChainID(id string) ChainOption { return func(c *chain) { @@ -128,61 +135,41 @@ func newChain(observer enqueue, opts ...ChainOption) Chain { opt(c) } - obsCreate := &api.IngestionEvent{IngestionEventTwelve: &api.IngestionEventTwelve{ - ID: newSpanID(), - Timestamp: getTimeRefString(c.StartTime), - Type: api.IngestionEventTwelveType(ingestionCreateChain).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(c.ObservationID), - TraceID: getStringRef(c.TraceID), - ParentObservationID: getStringRef(c.ParentObservationID), - Name: getStringRef(c.Name), - Metadata: c.Metadata, - Input: convertInput(c.Input, nil), - Output: convertOutput(c.Output), - StartTime: c.StartTime, - EndTime: c.EndTime, - Level: c.Level.ToLangfuse(), - StatusMessage: c.Status, - Version: c.Version, - }, - }} - - c.observer.enqueue(obsCreate) - return c } func (c *chain) End(opts ...ChainOption) { - id := c.ObservationID - startTime := c.StartTime - c.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(c) - } - - // preserve the original observation ID and start time - c.ObservationID = id - c.StartTime = startTime - - chainUpdate := &api.IngestionEvent{IngestionEventTwelve: &api.IngestionEventTwelve{ - ID: newSpanID(), - Timestamp: getTimeRefString(c.EndTime), - Type: api.IngestionEventTwelveType(ingestionCreateChain).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(c.ObservationID), - Name: getStringRef(c.Name), - Metadata: c.Metadata, - Input: convertInput(c.Input, nil), - Output: convertOutput(c.Output), - EndTime: c.EndTime, - Level: c.Level.ToLangfuse(), - StatusMessage: c.Status, - Version: c.Version, - }, - }} + c.endOnce.Do(func() { + id := c.ObservationID + startTime := c.StartTime + c.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(c) + } + + // preserve the original observation ID and start time + c.ObservationID = id + c.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: c.TraceID, + ObservationID: c.ObservationID, + ParentObservationID: c.ParentObservationID, + Name: c.Name, + ObservationType: otelObservationTypeChain, + StartTime: c.StartTime, + EndTime: c.EndTime, + Level: c.Level, + Status: c.Status, + Version: c.Version, + Input: convertInput(c.Input, nil), + Output: convertOutput(c.Output), + Metadata: c.Metadata, + TraceCtx: c.traceCtx, + }) - c.observer.enqueue(chainUpdate) + c.observer.enqueue(otelSpan) + }) } func (c *chain) String() string { @@ -194,10 +181,15 @@ func (c *chain) MarshalJSON() ([]byte, error) { } func (c *chain) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := c.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: c.TraceID, ObservationID: c.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: c.observer, } diff --git a/backend/pkg/observability/langfuse/client.go b/backend/pkg/observability/langfuse/client.go index 569de36ce..f0a321316 100644 --- a/backend/pkg/observability/langfuse/client.go +++ b/backend/pkg/observability/langfuse/client.go @@ -15,7 +15,11 @@ import ( "pentagi/pkg/observability/langfuse/api/option" ) -const InstrumentationVersion = "2.0.0" +const ( + InstrumentationVersion = "2.0.0" + IngestionVersion = "4" + IngestionVersionHeader = "x-langfuse-ingestion-version" +) type AnnotationQueuesClient interface { Createqueue(ctx context.Context, request *api.CreateAnnotationQueueRequest, opts ...option.RequestOption) (*api.AnnotationQueue, error) @@ -315,6 +319,7 @@ func NewClient(opts ...ClientContextOption) (*Client, error) { "x_langfuse_sdk_version": []string{InstrumentationVersion}, "x_langfuse_public_key": []string{clientCtx.PublicKey}, "x_langfuse_project_id": []string{clientCtx.ProjectID}, + IngestionVersionHeader: []string{IngestionVersion}, }), } diff --git a/backend/pkg/observability/langfuse/context.go b/backend/pkg/observability/langfuse/context.go index 99aedc77b..debd98007 100644 --- a/backend/pkg/observability/langfuse/context.go +++ b/backend/pkg/observability/langfuse/context.go @@ -7,8 +7,9 @@ type ObservationContextKey int var observationContextKey ObservationContextKey type observationContext struct { - TraceID string `json:"trace_id"` - ObservationID string `json:"observation_id"` + TraceID string `json:"trace_id"` + ObservationID string `json:"observation_id"` + TraceCtx *TraceContext `json:"trace_ctx,omitempty"` } func getObservationContext(ctx context.Context) (observationContext, bool) { diff --git a/backend/pkg/observability/langfuse/embedding.go b/backend/pkg/observability/langfuse/embedding.go index 866cabc54..9e377acdc 100644 --- a/backend/pkg/observability/langfuse/embedding.go +++ b/backend/pkg/observability/langfuse/embedding.go @@ -4,9 +4,8 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - - "pentagi/pkg/observability/langfuse/api" ) const ( @@ -37,7 +36,9 @@ type embedding struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type EmbeddingOption func(*embedding) @@ -54,6 +55,12 @@ func withEmbeddingParentObservationID(parentObservationID string) EmbeddingOptio } } +func withEmbeddingTraceContext(traceCtx *TraceContext) EmbeddingOption { + return func(e *embedding) { + e.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithEmbeddingID sets on creation time func WithEmbeddingID(id string) EmbeddingOption { return func(e *embedding) { @@ -135,63 +142,42 @@ func newEmbedding(observer enqueue, opts ...EmbeddingOption) Embedding { opt(e) } - obsCreate := &api.IngestionEvent{IngestionEventFifteen: &api.IngestionEventFifteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(e.StartTime), - Type: api.IngestionEventFifteenType(ingestionCreateEmbedding).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(e.ObservationID), - TraceID: getStringRef(e.TraceID), - ParentObservationID: getStringRef(e.ParentObservationID), - Name: getStringRef(e.Name), - Metadata: e.Metadata, - Input: convertInput(e.Input, nil), - Output: convertOutput(e.Output), - StartTime: e.StartTime, - EndTime: e.EndTime, - Level: e.Level.ToLangfuse(), - StatusMessage: e.Status, - Version: e.Version, - Model: e.Model, - }, - }} - - e.observer.enqueue(obsCreate) - return e } func (e *embedding) End(opts ...EmbeddingOption) { - id := e.ObservationID - startTime := e.StartTime - e.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(e) - } - - // preserve the original observation ID and start time - e.ObservationID = id - e.StartTime = startTime - - embeddingUpdate := &api.IngestionEvent{IngestionEventFifteen: &api.IngestionEventFifteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(e.EndTime), - Type: api.IngestionEventFifteenType(ingestionCreateEmbedding).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(e.ObservationID), - Name: getStringRef(e.Name), - Metadata: e.Metadata, - Input: convertInput(e.Input, nil), - Output: convertOutput(e.Output), - EndTime: e.EndTime, - Level: e.Level.ToLangfuse(), - StatusMessage: e.Status, - Version: e.Version, - Model: e.Model, - }, - }} + e.endOnce.Do(func() { + id := e.ObservationID + startTime := e.StartTime + e.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(e) + } + + // preserve the original observation ID and start time + e.ObservationID = id + e.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: e.TraceID, + ObservationID: e.ObservationID, + ParentObservationID: e.ParentObservationID, + Name: e.Name, + ObservationType: otelObservationTypeEmbedding, + StartTime: e.StartTime, + EndTime: e.EndTime, + Level: e.Level, + Status: e.Status, + Version: e.Version, + Input: convertInput(e.Input, nil), + Output: convertOutput(e.Output), + Metadata: e.Metadata, + Model: e.Model, + TraceCtx: e.traceCtx, + }) - e.observer.enqueue(embeddingUpdate) + e.observer.enqueue(otelSpan) + }) } func (e *embedding) String() string { @@ -203,10 +189,15 @@ func (e *embedding) MarshalJSON() ([]byte, error) { } func (e *embedding) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := e.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: e.TraceID, ObservationID: e.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: e.observer, } diff --git a/backend/pkg/observability/langfuse/evaluator.go b/backend/pkg/observability/langfuse/evaluator.go index 50285dece..d03b88e60 100644 --- a/backend/pkg/observability/langfuse/evaluator.go +++ b/backend/pkg/observability/langfuse/evaluator.go @@ -4,10 +4,9 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - "pentagi/pkg/observability/langfuse/api" - "github.com/vxcontrol/langchaingo/llms" ) @@ -41,7 +40,9 @@ type evaluator struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type EvaluatorOption func(*evaluator) @@ -58,6 +59,12 @@ func withEvaluatorParentObservationID(parentObservationID string) EvaluatorOptio } } +func withEvaluatorTraceContext(traceCtx *TraceContext) EvaluatorOption { + return func(e *evaluator) { + e.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithEvaluatorID sets on creation time func WithEvaluatorID(id string) EvaluatorOption { return func(e *evaluator) { @@ -151,65 +158,43 @@ func newEvaluator(observer enqueue, opts ...EvaluatorOption) Evaluator { opt(e) } - obsCreate := &api.IngestionEvent{IngestionEventFourteen: &api.IngestionEventFourteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(e.StartTime), - Type: api.IngestionEventFourteenType(ingestionCreateEvaluator).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(e.ObservationID), - TraceID: getStringRef(e.TraceID), - ParentObservationID: getStringRef(e.ParentObservationID), - Name: getStringRef(e.Name), - Metadata: e.Metadata, - Input: convertInput(e.Input, e.Tools), - Output: convertOutput(e.Output), - StartTime: e.StartTime, - EndTime: e.EndTime, - Level: e.Level.ToLangfuse(), - StatusMessage: e.Status, - Version: e.Version, - Model: e.Model, - ModelParameters: e.ModelParameters.ToLangfuse(), - }, - }} - - e.observer.enqueue(obsCreate) - return e } func (e *evaluator) End(opts ...EvaluatorOption) { - id := e.ObservationID - startTime := e.StartTime - e.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(e) - } - - // preserve the original observation ID and start time - e.ObservationID = id - e.StartTime = startTime - - evaluatorUpdate := &api.IngestionEvent{IngestionEventFourteen: &api.IngestionEventFourteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(e.EndTime), - Type: api.IngestionEventFourteenType(ingestionCreateEvaluator).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(e.ObservationID), - Name: getStringRef(e.Name), - Metadata: e.Metadata, - Input: convertInput(e.Input, e.Tools), - Output: convertOutput(e.Output), - EndTime: e.EndTime, - Level: e.Level.ToLangfuse(), - StatusMessage: e.Status, - Version: e.Version, - Model: e.Model, - ModelParameters: e.ModelParameters.ToLangfuse(), - }, - }} + e.endOnce.Do(func() { + id := e.ObservationID + startTime := e.StartTime + e.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(e) + } + + // preserve the original observation ID and start time + e.ObservationID = id + e.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: e.TraceID, + ObservationID: e.ObservationID, + ParentObservationID: e.ParentObservationID, + Name: e.Name, + ObservationType: otelObservationTypeEvaluator, + StartTime: e.StartTime, + EndTime: e.EndTime, + Level: e.Level, + Status: e.Status, + Version: e.Version, + Input: convertInput(e.Input, e.Tools), + Output: convertOutput(e.Output), + Metadata: e.Metadata, + Model: e.Model, + ModelParameters: e.ModelParameters, + TraceCtx: e.traceCtx, + }) - e.observer.enqueue(evaluatorUpdate) + e.observer.enqueue(otelSpan) + }) } func (e *evaluator) String() string { @@ -221,10 +206,15 @@ func (e *evaluator) MarshalJSON() ([]byte, error) { } func (e *evaluator) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := e.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: e.TraceID, ObservationID: e.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: e.observer, } diff --git a/backend/pkg/observability/langfuse/event.go b/backend/pkg/observability/langfuse/event.go index 46d7dc241..bab13baec 100644 --- a/backend/pkg/observability/langfuse/event.go +++ b/backend/pkg/observability/langfuse/event.go @@ -5,8 +5,6 @@ import ( "encoding/json" "fmt" "time" - - "pentagi/pkg/observability/langfuse/api" ) const ( @@ -34,7 +32,8 @@ type event struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + observer enqueue `json:"-"` } type EventOption func(*event) @@ -51,6 +50,12 @@ func withEventParentObservationID(parentObservationID string) EventOption { } } +func withEventTraceContext(traceCtx *TraceContext) EventOption { + return func(e *event) { + e.traceCtx = cloneTraceContext(traceCtx) + } +} + func WithEventName(name string) EventOption { return func(e *event) { e.Name = name @@ -113,26 +118,24 @@ func newEvent(observer enqueue, opts ...EventOption) Event { opt(e) } - obsCreate := &api.IngestionEvent{IngestionEventSix: &api.IngestionEventSix{ - ID: newSpanID(), - Timestamp: getTimeRefString(e.Time), - Type: api.IngestionEventSixType(ingestionCreateEvent).Ptr(), - Body: &api.CreateEventBody{ - ID: getStringRef(e.ObservationID), - TraceID: getStringRef(e.TraceID), - ParentObservationID: getStringRef(e.ParentObservationID), - Name: getStringRef(e.Name), - StartTime: e.Time, - Metadata: e.Metadata, - Input: e.Input, - Output: e.Output, - Level: e.Level.ToLangfuse(), - StatusMessage: e.Status, - Version: e.Version, - }, - }} - - e.observer.enqueue(obsCreate) + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: e.TraceID, + ObservationID: e.ObservationID, + ParentObservationID: e.ParentObservationID, + Name: e.Name, + ObservationType: otelObservationTypeEvent, + StartTime: e.Time, + EndTime: e.Time, + Level: e.Level, + Status: e.Status, + Version: e.Version, + Input: convertInput(e.Input, nil), + Output: convertOutput(e.Output), + Metadata: e.Metadata, + TraceCtx: e.traceCtx, + }) + + e.observer.enqueue(otelSpan) return e } @@ -147,10 +150,15 @@ func (e *event) MarshalJSON() ([]byte, error) { } func (e *event) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := e.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: e.TraceID, ObservationID: e.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: e.observer, } diff --git a/backend/pkg/observability/langfuse/generation.go b/backend/pkg/observability/langfuse/generation.go index a2c9ff715..637369da5 100644 --- a/backend/pkg/observability/langfuse/generation.go +++ b/backend/pkg/observability/langfuse/generation.go @@ -4,10 +4,9 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - "pentagi/pkg/observability/langfuse/api" - "github.com/vxcontrol/langchaingo/llms" ) @@ -45,7 +44,9 @@ type generation struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type GenerationOption func(*generation) @@ -62,6 +63,12 @@ func withGenerationParentObservationID(parentObservationID string) GenerationOpt } } +func withGenerationTraceContext(traceCtx *TraceContext) GenerationOption { + return func(g *generation) { + g.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithGenerationID sets on creation time func WithGenerationID(id string) GenerationOption { return func(g *generation) { @@ -185,73 +192,47 @@ func newGeneration(observer enqueue, opts ...GenerationOption) Generation { g.CompletionStartTime = g.StartTime } - genCreate := &api.IngestionEvent{IngestionEventFour: &api.IngestionEventFour{ - ID: newSpanID(), - Timestamp: getTimeRefString(g.StartTime), - Type: api.IngestionEventFourType(ingestionCreateGeneration).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(g.ObservationID), - TraceID: getStringRef(g.TraceID), - ParentObservationID: getStringRef(g.ParentObservationID), - Name: getStringRef(g.Name), - Metadata: g.Metadata, - Input: convertInput(g.Input, g.Tools), - Output: convertOutput(g.Output), - StartTime: g.StartTime, - EndTime: g.EndTime, - CompletionStartTime: g.CompletionStartTime, - Level: g.Level.ToLangfuse(), - StatusMessage: g.Status, - Version: g.Version, - Model: g.Model, - ModelParameters: g.ModelParameters.ToLangfuse(), - PromptName: g.PromptName, - PromptVersion: g.PromptVersion, - Usage: g.Usage.ToLangfuse(), - }, - }} - - g.observer.enqueue(genCreate) - return g } func (g *generation) End(opts ...GenerationOption) { - id := g.ObservationID - startTime := g.StartTime - g.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(g) - } - - // preserve the original observation ID and start time - g.ObservationID = id - g.StartTime = startTime - - genUpdate := &api.IngestionEvent{IngestionEventFive: &api.IngestionEventFive{ - ID: newSpanID(), - Timestamp: getTimeRefString(g.EndTime), - Type: api.IngestionEventFiveType(ingestionUpdateGeneration).Ptr(), - Body: &api.UpdateGenerationBody{ - ID: g.ObservationID, - Name: getStringRef(g.Name), - Metadata: g.Metadata, - Input: convertInput(g.Input, g.Tools), - Output: convertOutput(g.Output), + g.endOnce.Do(func() { + id := g.ObservationID + startTime := g.StartTime + g.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(g) + } + + // preserve the original observation ID and start time + g.ObservationID = id + g.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: g.TraceID, + ObservationID: g.ObservationID, + ParentObservationID: g.ParentObservationID, + Name: g.Name, + ObservationType: otelObservationTypeGeneration, + StartTime: g.StartTime, EndTime: g.EndTime, CompletionStartTime: g.CompletionStartTime, - Level: g.Level.ToLangfuse(), - StatusMessage: g.Status, + Level: g.Level, + Status: g.Status, Version: g.Version, + Input: convertInput(g.Input, g.Tools), + Output: convertOutput(g.Output), + Metadata: g.Metadata, Model: g.Model, - ModelParameters: g.ModelParameters.ToLangfuse(), + ModelParameters: g.ModelParameters, + Usage: g.Usage, PromptName: g.PromptName, PromptVersion: g.PromptVersion, - Usage: g.Usage.ToLangfuse(), - }, - }} + TraceCtx: g.traceCtx, + }) - g.observer.enqueue(genUpdate) + g.observer.enqueue(otelSpan) + }) } func (g *generation) String() string { @@ -263,10 +244,15 @@ func (g *generation) MarshalJSON() ([]byte, error) { } func (g *generation) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := g.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: g.TraceID, ObservationID: g.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: g.observer, } diff --git a/backend/pkg/observability/langfuse/guardrail.go b/backend/pkg/observability/langfuse/guardrail.go index 0e36877ef..d6f439c6c 100644 --- a/backend/pkg/observability/langfuse/guardrail.go +++ b/backend/pkg/observability/langfuse/guardrail.go @@ -4,10 +4,9 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - "pentagi/pkg/observability/langfuse/api" - "github.com/vxcontrol/langchaingo/llms" ) @@ -41,7 +40,9 @@ type guardrail struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type GuardrailOption func(*guardrail) @@ -58,6 +59,12 @@ func withGuardrailParentObservationID(parentObservationID string) GuardrailOptio } } +func withGuardrailTraceContext(traceCtx *TraceContext) GuardrailOption { + return func(g *guardrail) { + g.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithGuardrailID sets on creation time func WithGuardrailID(id string) GuardrailOption { return func(g *guardrail) { @@ -151,65 +158,43 @@ func newGuardrail(observer enqueue, opts ...GuardrailOption) Guardrail { opt(g) } - obsCreate := &api.IngestionEvent{IngestionEventSixteen: &api.IngestionEventSixteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(g.StartTime), - Type: api.IngestionEventSixteenType(ingestionCreateGuardrail).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(g.ObservationID), - TraceID: getStringRef(g.TraceID), - ParentObservationID: getStringRef(g.ParentObservationID), - Name: getStringRef(g.Name), - Metadata: g.Metadata, - Input: convertInput(g.Input, g.Tools), - Output: convertOutput(g.Output), - StartTime: g.StartTime, - EndTime: g.EndTime, - Level: g.Level.ToLangfuse(), - StatusMessage: g.Status, - Version: g.Version, - Model: g.Model, - ModelParameters: g.ModelParameters.ToLangfuse(), - }, - }} - - g.observer.enqueue(obsCreate) - return g } func (g *guardrail) End(opts ...GuardrailOption) { - id := g.ObservationID - startTime := g.StartTime - g.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(g) - } - - // preserve the original observation ID and start time - g.ObservationID = id - g.StartTime = startTime - - guardrailUpdate := &api.IngestionEvent{IngestionEventSixteen: &api.IngestionEventSixteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(g.EndTime), - Type: api.IngestionEventSixteenType(ingestionCreateGuardrail).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(g.ObservationID), - Name: getStringRef(g.Name), - Metadata: g.Metadata, - Input: convertInput(g.Input, g.Tools), - Output: convertOutput(g.Output), - EndTime: g.EndTime, - Level: g.Level.ToLangfuse(), - StatusMessage: g.Status, - Version: g.Version, - Model: g.Model, - ModelParameters: g.ModelParameters.ToLangfuse(), - }, - }} + g.endOnce.Do(func() { + id := g.ObservationID + startTime := g.StartTime + g.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(g) + } + + // preserve the original observation ID and start time + g.ObservationID = id + g.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: g.TraceID, + ObservationID: g.ObservationID, + ParentObservationID: g.ParentObservationID, + Name: g.Name, + ObservationType: otelObservationTypeGuardrail, + StartTime: g.StartTime, + EndTime: g.EndTime, + Level: g.Level, + Status: g.Status, + Version: g.Version, + Input: convertInput(g.Input, g.Tools), + Output: convertOutput(g.Output), + Metadata: g.Metadata, + Model: g.Model, + ModelParameters: g.ModelParameters, + TraceCtx: g.traceCtx, + }) - g.observer.enqueue(guardrailUpdate) + g.observer.enqueue(otelSpan) + }) } func (g *guardrail) String() string { @@ -221,10 +206,15 @@ func (g *guardrail) MarshalJSON() ([]byte, error) { } func (g *guardrail) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := g.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: g.TraceID, ObservationID: g.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: g.observer, } diff --git a/backend/pkg/observability/langfuse/helpers.go b/backend/pkg/observability/langfuse/helpers.go index 3d36f031d..2eb98cf20 100644 --- a/backend/pkg/observability/langfuse/helpers.go +++ b/backend/pkg/observability/langfuse/helpers.go @@ -6,6 +6,8 @@ import ( "encoding/json" "fmt" "maps" + "slices" + "strconv" "time" "pentagi/pkg/observability/langfuse/api" @@ -16,6 +18,41 @@ import ( const ( firstVersion = "v1" timeFormat8601 = "2006-01-02T15:04:05.000000Z" + + otelObservationTypeSpan = "span" + otelObservationTypeGeneration = "generation" + otelObservationTypeEvent = "event" + otelObservationTypeEmbedding = "embedding" + otelObservationTypeAgent = "agent" + otelObservationTypeTool = "tool" + otelObservationTypeChain = "chain" + otelObservationTypeRetriever = "retriever" + otelObservationTypeGuardrail = "guardrail" + otelObservationTypeEvaluator = "evaluator" + + otelAttrObservationType = "langfuse.observation.type" + otelAttrObservationLevel = "langfuse.observation.level" + otelAttrObservationStatusMessage = "langfuse.observation.status_message" + otelAttrObservationInput = "langfuse.observation.input" + otelAttrObservationOutput = "langfuse.observation.output" + otelAttrObservationMetadataPrefix = "langfuse.observation.metadata." + otelAttrObservationModelName = "langfuse.observation.model.name" + otelAttrObservationModelParameters = "langfuse.observation.model.parameters" + otelAttrObservationUsageDetails = "langfuse.observation.usage_details" + otelAttrObservationCostDetails = "langfuse.observation.cost_details" + otelAttrObservationPromptName = "langfuse.observation.prompt.name" + otelAttrObservationPromptVersion = "langfuse.observation.prompt.version" + otelAttrObservationCompletionStart = "langfuse.observation.completion_start_time" + + otelAttrTraceName = "langfuse.trace.name" + otelAttrTracePublic = "langfuse.trace.public" + otelAttrTraceTags = "langfuse.trace.tags" + otelAttrTraceMetadataPrefix = "langfuse.trace.metadata." + otelAttrUserID = "langfuse.user.id" + otelAttrSessionID = "langfuse.session.id" + otelAttrRelease = "langfuse.release" + otelAttrVersion = "langfuse.version" + otelAttrEnvironment = "langfuse.environment" ) var ( @@ -335,3 +372,246 @@ func getIntRef(i int) *int { func getBoolRef(b bool) *bool { return &b } + +type otelSpanParams struct { + TraceID string + ObservationID string + ParentObservationID string + Name string + ObservationType string + StartTime *time.Time + EndTime *time.Time + CompletionStartTime *time.Time + Level ObservationLevel + Status *string + Version *string + Input any + Output any + Metadata Metadata + Model *string + ModelParameters *ModelParameters + Usage *GenerationUsage + PromptName *string + PromptVersion *int + TraceCtx *TraceContext +} + +func otelStringAttr(key, val string) *api.OtelAttribute { + k := key + v := val + return &api.OtelAttribute{ + Key: &k, + Value: &api.OtelAttributeValue{StringValue: &v}, + } +} + +func otelIntAttr(key string, val int) *api.OtelAttribute { + k := key + v := val + return &api.OtelAttribute{ + Key: &k, + Value: &api.OtelAttributeValue{IntValue: &v}, + } +} + +func otelFloatAttr(key string, val float64) *api.OtelAttribute { + k := key + v := val + return &api.OtelAttribute{ + Key: &k, + Value: &api.OtelAttributeValue{DoubleValue: &v}, + } +} + +func otelBoolAttr(key string, val bool) *api.OtelAttribute { + k := key + v := val + return &api.OtelAttribute{ + Key: &k, + Value: &api.OtelAttributeValue{BoolValue: &v}, + } +} + +func otelJSONOrStringAttr(key string, val any) *api.OtelAttribute { + if val == nil { + return nil + } + if s, ok := val.(string); ok { + return otelStringAttr(key, s) + } + data, err := json.Marshal(val) + if err != nil { + return otelStringAttr(key, fmt.Sprintf("%v", val)) + } + return otelStringAttr(key, string(data)) +} + +func otelAnyAttr(key string, val any) *api.OtelAttribute { + if val == nil { + return nil + } + switch v := val.(type) { + case string: + return otelStringAttr(key, v) + case bool: + return otelBoolAttr(key, v) + case int: + return otelIntAttr(key, v) + case int64: + return otelIntAttr(key, int(v)) + case int32: + return otelIntAttr(key, int(v)) + case float64: + return otelFloatAttr(key, v) + case float32: + return otelFloatAttr(key, float64(v)) + default: + return otelJSONOrStringAttr(key, v) + } +} + +func buildOtelSpan(p otelSpanParams) *api.OtelSpan { + startTime := getCurrentTime() + if p.StartTime != nil { + startTime = p.StartTime.UTC() + } + endTime := startTime + if p.EndTime != nil { + endTime = p.EndTime.UTC() + } + + var attrs []*api.OtelAttribute + appendAttr := func(attr *api.OtelAttribute) { + if attr != nil { + attrs = append(attrs, attr) + } + } + + obsType := p.ObservationType + if obsType == "" { + obsType = otelObservationTypeSpan + } + appendAttr(otelStringAttr(otelAttrObservationType, obsType)) + + levelStr := string(*p.Level.ToLangfuse()) + appendAttr(otelStringAttr(otelAttrObservationLevel, levelStr)) + + if p.Status != nil { + appendAttr(otelStringAttr(otelAttrObservationStatusMessage, *p.Status)) + } + + inputVal := p.Input + if inputVal == nil && p.ParentObservationID == "" && p.TraceCtx != nil && p.TraceCtx.Input != nil { + inputVal = p.TraceCtx.Input + } + appendAttr(otelJSONOrStringAttr(otelAttrObservationInput, inputVal)) + + outputVal := p.Output + if outputVal == nil && p.ParentObservationID == "" && p.TraceCtx != nil && p.TraceCtx.Output != nil { + outputVal = p.TraceCtx.Output + } + appendAttr(otelJSONOrStringAttr(otelAttrObservationOutput, outputVal)) + + if len(p.Metadata) > 0 { + for _, k := range slices.Sorted(maps.Keys(p.Metadata)) { + appendAttr(otelAnyAttr(otelAttrObservationMetadataPrefix+k, p.Metadata[k])) + } + } + + if p.Model != nil && *p.Model != "" { + appendAttr(otelStringAttr(otelAttrObservationModelName, *p.Model)) + } + + if params := p.ModelParameters.ToLangfuse(); params != nil { + appendAttr(otelJSONOrStringAttr(otelAttrObservationModelParameters, params)) + } + + if usage := p.Usage.ToLangfuse(); usage != nil && usage.Usage != nil { + u := usage.Usage + usageDetails := map[string]int{ + "input": u.Input, + "output": u.Output, + "total": u.Total, + } + appendAttr(otelJSONOrStringAttr(otelAttrObservationUsageDetails, usageDetails)) + + if u.InputCost != nil || u.OutputCost != nil || u.TotalCost != nil { + costDetails := make(map[string]float64) + if u.InputCost != nil { + costDetails["input"] = *u.InputCost + } + if u.OutputCost != nil { + costDetails["output"] = *u.OutputCost + } + if u.TotalCost != nil { + costDetails["total"] = *u.TotalCost + } + appendAttr(otelJSONOrStringAttr(otelAttrObservationCostDetails, costDetails)) + } + } + + if p.PromptName != nil && *p.PromptName != "" { + appendAttr(otelStringAttr(otelAttrObservationPromptName, *p.PromptName)) + } + if p.PromptVersion != nil { + appendAttr(otelIntAttr(otelAttrObservationPromptVersion, *p.PromptVersion)) + } + if p.CompletionStartTime != nil { + appendAttr(otelStringAttr(otelAttrObservationCompletionStart, getTimeRefString(p.CompletionStartTime))) + } + + version := p.Version + if (version == nil || *version == "") && p.TraceCtx != nil && p.TraceCtx.Version != nil { + version = p.TraceCtx.Version + } + if version != nil && *version != "" { + appendAttr(otelStringAttr(otelAttrVersion, *version)) + } + + if p.TraceCtx != nil { + if p.TraceCtx.Name != nil && *p.TraceCtx.Name != "" { + appendAttr(otelStringAttr(otelAttrTraceName, *p.TraceCtx.Name)) + } + if p.TraceCtx.UserID != nil && *p.TraceCtx.UserID != "" { + appendAttr(otelStringAttr(otelAttrUserID, *p.TraceCtx.UserID)) + } + if p.TraceCtx.SessionID != nil && *p.TraceCtx.SessionID != "" { + appendAttr(otelStringAttr(otelAttrSessionID, *p.TraceCtx.SessionID)) + } + if p.TraceCtx.Public != nil { + appendAttr(otelBoolAttr(otelAttrTracePublic, *p.TraceCtx.Public)) + } + if len(p.TraceCtx.Tags) > 0 { + appendAttr(otelJSONOrStringAttr(otelAttrTraceTags, p.TraceCtx.Tags)) + } + if len(p.TraceCtx.Metadata) > 0 { + for _, k := range slices.Sorted(maps.Keys(p.TraceCtx.Metadata)) { + appendAttr(otelAnyAttr(otelAttrTraceMetadataPrefix+k, p.TraceCtx.Metadata[k])) + } + } + } + + span := &api.OtelSpan{ + TraceID: p.TraceID, + SpanID: p.ObservationID, + Name: getStringRef(p.Name), + Kind: getIntRef(1), + StartTimeUnixNano: strconv.FormatInt(startTime.UnixNano(), 10), + EndTimeUnixNano: strconv.FormatInt(endTime.UnixNano(), 10), + Attributes: attrs, + } + if p.ParentObservationID != "" { + span.ParentSpanID = p.ParentObservationID + } + if p.Level == ObservationLevelError { + status := map[string]any{"code": 2} + if p.Status != nil { + status["message"] = *p.Status + } + span.Status = status + } else if p.Status != nil { + span.Status = map[string]any{"code": 1, "message": *p.Status} + } + + return span +} diff --git a/backend/pkg/observability/langfuse/noop.go b/backend/pkg/observability/langfuse/noop.go index 0150c95a1..a1c7fa5bd 100644 --- a/backend/pkg/observability/langfuse/noop.go +++ b/backend/pkg/observability/langfuse/noop.go @@ -34,10 +34,21 @@ func (o *noopObserver) NewObservation( } } + var mergedTraceCtx *TraceContext + if parentObsCtxFound && parentObsCtx.TraceCtx != nil { + mergedTraceCtx = parentObsCtx.TraceCtx + } + if obsCtx.TraceCtx != nil { + mergedTraceCtx = mergeTraceContext(mergedTraceCtx, obsCtx.TraceCtx) + } else { + mergedTraceCtx = cloneTraceContext(mergedTraceCtx) + } + obs := &observation{ obsCtx: observationContext{ TraceID: obsCtx.TraceID, ObservationID: obsCtx.ObservationID, + TraceCtx: mergedTraceCtx, }, observer: o, } @@ -53,5 +64,8 @@ func (o *noopObserver) ForceFlush(ctx context.Context) error { return nil } -func (o *noopObserver) enqueue(event *api.IngestionEvent) { +func (o *noopObserver) enqueue(span *api.OtelSpan) { +} + +func (o *noopObserver) enqueueScore(score *api.CreateScoreRequest) { } diff --git a/backend/pkg/observability/langfuse/noop_test.go b/backend/pkg/observability/langfuse/noop_test.go index 8932a2727..025080fc7 100644 --- a/backend/pkg/observability/langfuse/noop_test.go +++ b/backend/pkg/observability/langfuse/noop_test.go @@ -20,7 +20,9 @@ func TestNoop_NoopObserver_LifecycleCallsDoNothing(t *testing.T) { assert.NoError(t, obs.ForceFlush(context.Background())) assert.NotPanics(t, func() { obs.enqueue(nil) - obs.enqueue(&api.IngestionEvent{}) + obs.enqueue(&api.OtelSpan{}) + obs.enqueueScore(nil) + obs.enqueueScore(&api.CreateScoreRequest{}) }) } diff --git a/backend/pkg/observability/langfuse/observation.go b/backend/pkg/observability/langfuse/observation.go index 1f08bb945..8e6bab1ef 100644 --- a/backend/pkg/observability/langfuse/observation.go +++ b/backend/pkg/observability/langfuse/observation.go @@ -4,8 +4,6 @@ import ( "context" "fmt" - "pentagi/pkg/observability/langfuse/api" - "github.com/sirupsen/logrus" ) @@ -52,14 +50,20 @@ func (o *observation) Log(ctx context.Context, message string) { "langfuse_log_id": logID, }).Info(message) - obsLog := &api.IngestionEvent{IngestionEventSeven: &api.IngestionEventSeven{ - ID: logID, - Timestamp: getCurrentTimeString(), - Type: api.IngestionEventSevenType(ingestionPutLog).Ptr(), - Body: &api.SdkLogBody{ - Log: message, - }, - }} + now := getCurrentTimeRef() + obsLog := buildOtelSpan(otelSpanParams{ + TraceID: o.obsCtx.TraceID, + ObservationID: logID, + ParentObservationID: o.obsCtx.ObservationID, + Name: "sdk-log", + ObservationType: otelObservationTypeEvent, + StartTime: now, + EndTime: now, + Level: ObservationLevelDefault, + Input: message, + Version: getStringRef(firstVersion), + TraceCtx: o.obsCtx.TraceCtx, + }) o.observer.enqueue(obsLog) } @@ -76,6 +80,7 @@ func (o *observation) Event(opts ...EventOption) Event { opts = append(opts, withEventTraceID(o.obsCtx.TraceID), withEventParentObservationID(o.obsCtx.ObservationID), + withEventTraceContext(o.obsCtx.TraceCtx), ) return newEvent(o.observer, opts...) } @@ -84,6 +89,7 @@ func (o *observation) Span(opts ...SpanOption) Span { opts = append(opts, withSpanTraceID(o.obsCtx.TraceID), withSpanParentObservationID(o.obsCtx.ObservationID), + withSpanTraceContext(o.obsCtx.TraceCtx), ) return newSpan(o.observer, opts...) } @@ -92,6 +98,7 @@ func (o *observation) Generation(opts ...GenerationOption) Generation { opts = append(opts, withGenerationTraceID(o.obsCtx.TraceID), withGenerationParentObservationID(o.obsCtx.ObservationID), + withGenerationTraceContext(o.obsCtx.TraceCtx), ) return newGeneration(o.observer, opts...) } @@ -100,6 +107,7 @@ func (o *observation) Agent(opts ...AgentOption) Agent { opts = append(opts, withAgentTraceID(o.obsCtx.TraceID), withAgentParentObservationID(o.obsCtx.ObservationID), + withAgentTraceContext(o.obsCtx.TraceCtx), ) return newAgent(o.observer, opts...) } @@ -108,6 +116,7 @@ func (o *observation) Tool(opts ...ToolOption) Tool { opts = append(opts, withToolTraceID(o.obsCtx.TraceID), withToolParentObservationID(o.obsCtx.ObservationID), + withToolTraceContext(o.obsCtx.TraceCtx), ) return newTool(o.observer, opts...) } @@ -116,6 +125,7 @@ func (o *observation) Chain(opts ...ChainOption) Chain { opts = append(opts, withChainTraceID(o.obsCtx.TraceID), withChainParentObservationID(o.obsCtx.ObservationID), + withChainTraceContext(o.obsCtx.TraceCtx), ) return newChain(o.observer, opts...) } @@ -124,6 +134,7 @@ func (o *observation) Retriever(opts ...RetrieverOption) Retriever { opts = append(opts, withRetrieverTraceID(o.obsCtx.TraceID), withRetrieverParentObservationID(o.obsCtx.ObservationID), + withRetrieverTraceContext(o.obsCtx.TraceCtx), ) return newRetriever(o.observer, opts...) } @@ -132,6 +143,7 @@ func (o *observation) Evaluator(opts ...EvaluatorOption) Evaluator { opts = append(opts, withEvaluatorTraceID(o.obsCtx.TraceID), withEvaluatorParentObservationID(o.obsCtx.ObservationID), + withEvaluatorTraceContext(o.obsCtx.TraceCtx), ) return newEvaluator(o.observer, opts...) } @@ -140,6 +152,7 @@ func (o *observation) Embedding(opts ...EmbeddingOption) Embedding { opts = append(opts, withEmbeddingTraceID(o.obsCtx.TraceID), withEmbeddingParentObservationID(o.obsCtx.ObservationID), + withEmbeddingTraceContext(o.obsCtx.TraceCtx), ) return newEmbedding(o.observer, opts...) } @@ -148,6 +161,7 @@ func (o *observation) Guardrail(opts ...GuardrailOption) Guardrail { opts = append(opts, withGuardrailTraceID(o.obsCtx.TraceID), withGuardrailParentObservationID(o.obsCtx.ObservationID), + withGuardrailTraceContext(o.obsCtx.TraceCtx), ) return newGuardrail(o.observer, opts...) } diff --git a/backend/pkg/observability/langfuse/observer.go b/backend/pkg/observability/langfuse/observer.go index bfd6d3e4e..ba5bdaf65 100644 --- a/backend/pkg/observability/langfuse/observer.go +++ b/backend/pkg/observability/langfuse/observer.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "errors" - "fmt" "net/http" "sync" "time" @@ -31,7 +30,8 @@ type Observer interface { } type enqueue interface { - enqueue(event *api.IngestionEvent) + enqueue(span *api.OtelSpan) + enqueueScore(score *api.CreateScoreRequest) } type observer struct { @@ -46,7 +46,7 @@ type observer struct { interval time.Duration timeout time.Duration queueSize int - queue chan *api.IngestionEvent + queue chan *api.OtelSpan flusher chan chan error } @@ -72,7 +72,7 @@ func NewObserver(client *Client, opts ...ObserverOption) Observer { if o.queueSize <= 0 || o.queueSize > 10000 { o.queueSize = defaultQueueSize } - o.queue = make(chan *api.IngestionEvent, o.queueSize) + o.queue = make(chan *api.OtelSpan, o.queueSize) o.wg.Add(1) go func() { @@ -105,14 +105,21 @@ func (o *observer) NewObservation( } } + var mergedTraceCtx *TraceContext + if parentObsCtxFound && parentObsCtx.TraceCtx != nil { + mergedTraceCtx = parentObsCtx.TraceCtx + } if obsCtx.TraceCtx != nil { - o.putTraceInfo(obsCtx) + mergedTraceCtx = mergeTraceContext(mergedTraceCtx, obsCtx.TraceCtx) + } else { + mergedTraceCtx = cloneTraceContext(mergedTraceCtx) } obs := &observation{ obsCtx: observationContext{ TraceID: obsCtx.TraceID, ObservationID: obsCtx.ObservationID, + TraceCtx: mergedTraceCtx, }, observer: o, } @@ -149,18 +156,44 @@ func (o *observer) ForceFlush(ctx context.Context) error { return <-ch } -func (o *observer) enqueue(event *api.IngestionEvent) { +func (o *observer) enqueue(span *api.OtelSpan) { o.mx.Lock() defer o.mx.Unlock() - if err := o.ctx.Err(); err != nil { + if err := o.ctx.Err(); err != nil || span == nil { return } - o.queue <- event + if o.release != "" { + span.Attributes = append(span.Attributes, otelStringAttr(otelAttrRelease, o.release)) + } + if o.environment != "" { + span.Attributes = append(span.Attributes, otelStringAttr(otelAttrEnvironment, o.environment)) + } + + o.queue <- span +} + +func (o *observer) enqueueScore(scoreReq *api.CreateScoreRequest) { + o.mx.Lock() + defer o.mx.Unlock() + + if err := o.ctx.Err(); err != nil || scoreReq == nil || o.client == nil || o.client.Score == nil { + return + } + + o.wg.Add(1) + go func() { + defer o.wg.Done() + ctx, cancel := context.WithTimeout(o.ctx, o.timeout) + defer cancel() + if _, err := o.client.Score.Create(ctx, scoreReq); err != nil && o.ctx.Err() == nil { + logrus.WithContext(o.ctx).WithError(err).Error("failed to send score to Langfuse") + } + }() } -func (o *observer) flush(ctx context.Context, batch []*api.IngestionEvent) error { +func (o *observer) flush(ctx context.Context, batch []*api.OtelSpan) error { ctx, cancel := context.WithTimeout(ctx, o.timeout) defer cancel() @@ -168,38 +201,40 @@ func (o *observer) flush(ctx context.Context, batch []*api.IngestionEvent) error return nil } - metadata := map[string]any{ - "batch_size": len(batch), - "app_release": o.release, - "sdk_integration": "langchain", - "sdk_name": "golang", - "sdk_version": InstrumentationVersion, - "public_key": o.client.PublicKey(), - "project_id": o.client.ProjectID(), + resourceAttrs := []*api.OtelAttribute{ + otelStringAttr("service.name", "pentagi"), + otelStringAttr("telemetry.sdk.language", "go"), + otelStringAttr("telemetry.sdk.name", "langfuse-observability-client-go"), + otelStringAttr("telemetry.sdk.version", InstrumentationVersion), } - - resp, err := o.client.Ingestion.Batch(ctx, &api.IngestionBatchRequest{ - Batch: batch, - Metadata: metadata, - }) - if err != nil { - return err + if o.release != "" { + resourceAttrs = append(resourceAttrs, otelStringAttr("service.version", o.release)) + } + if o.environment != "" { + resourceAttrs = append(resourceAttrs, otelStringAttr("deployment.environment", o.environment)) } - if len(resp.Errors) > 0 { - for _, event := range resp.Errors { - logrus.WithContext(ctx).WithFields(logrus.Fields{ - "event_id": event.ID, - "status": event.Status, - "message": event.Message, - "error": event.Error, - }).Errorf("failed to send event to Langfuse") - } - - return fmt.Errorf("failed to send %d events", len(resp.Errors)) + req := &api.OpentelemetryExportTracesRequest{ + ResourceSpans: []*api.OtelResourceSpan{ + { + Resource: &api.OtelResource{ + Attributes: resourceAttrs, + }, + ScopeSpans: []*api.OtelScopeSpan{ + { + Scope: &api.OtelScope{ + Name: getStringRef("langfuse-sdk"), + Version: getStringRef(InstrumentationVersion), + }, + Spans: batch, + }, + }, + }, + }, } - return nil + _, err := o.client.Opentelemetry.Exporttraces(ctx, req) + return err } // flushWithSplit flushes the batch and, if the request fails only because @@ -208,7 +243,7 @@ func (o *observer) flush(ctx context.Context, batch []*api.IngestionEvent) error // every event in the batch on a single oversized flush. Only an event that // still doesn't fit the limit on its own is ever dropped, and that is logged // with its approximate size so the loss is visible instead of silent. -func (o *observer) flushWithSplit(ctx context.Context, batch []*api.IngestionEvent) error { +func (o *observer) flushWithSplit(ctx context.Context, batch []*api.OtelSpan) error { if len(batch) == 0 { return nil } @@ -228,7 +263,7 @@ func (o *observer) flushWithSplit(ctx context.Context, batch []*api.IngestionEve if len(batch) == 1 { logrus.WithContext(ctx).WithFields(logrus.Fields{ - "event_size_bytes": approxIngestionEventSize(batch[0]), + "event_size_bytes": approxSpanSize(batch[0]), }).Error("dropping a single telemetry event that exceeds Langfuse's body size limit even alone") return nil } @@ -239,11 +274,11 @@ func (o *observer) flushWithSplit(ctx context.Context, batch []*api.IngestionEve return errors.Join(errFirst, errSecond) } -// approxIngestionEventSize returns the serialized size of a single event, or +// approxSpanSize returns the serialized size of a single span, or // -1 if it can't be marshaled - used only for the drop-log message above, so // a marshal failure there must not itself fail or panic. -func approxIngestionEventSize(event *api.IngestionEvent) int { - data, err := json.Marshal(event) +func approxSpanSize(span *api.OtelSpan) int { + data, err := json.Marshal(span) if err != nil { return -1 } @@ -251,7 +286,7 @@ func approxIngestionEventSize(event *api.IngestionEvent) int { } func (o *observer) sender() { - batch := make([]*api.IngestionEvent, 0, o.queueSize) + batch := make([]*api.OtelSpan, 0, o.queueSize) ticker := time.NewTicker(o.interval) defer ticker.Stop() @@ -269,8 +304,8 @@ func (o *observer) sender() { } batch = batch[:0] ticker.Reset(o.interval) - case event := <-o.queue: - batch = append(batch, event) + case span := <-o.queue: + batch = append(batch, span) if len(batch) >= o.queueSize { if err := o.flushWithSplit(o.ctx, batch); err != nil { logrus.WithContext(o.ctx).WithError(err).Error("failed to flush events by queue size") @@ -281,28 +316,3 @@ func (o *observer) sender() { } } } - -func (o *observer) putTraceInfo(obsCtx ObservationContext) { - traceCreate := &api.IngestionEvent{IngestionEventZero: &api.IngestionEventZero{ - ID: newSpanID(), - Timestamp: getCurrentTimeString(), - Type: api.IngestionEventZeroType(ingestionCreateTrace).Ptr(), - Body: &api.TraceBody{ - ID: getStringRef(obsCtx.TraceID), - Timestamp: obsCtx.TraceCtx.Timestamp, - Name: obsCtx.TraceCtx.Name, - UserID: obsCtx.TraceCtx.UserID, - Input: obsCtx.TraceCtx.Input, - Output: obsCtx.TraceCtx.Output, - SessionID: obsCtx.TraceCtx.SessionID, - Release: getStringRef(o.release), - Version: obsCtx.TraceCtx.Version, - Metadata: obsCtx.TraceCtx.Metadata, - Tags: obsCtx.TraceCtx.Tags, - Public: obsCtx.TraceCtx.Public, - Environment: getStringRef(o.environment), - }, - }} - - o.enqueue(traceCreate) -} diff --git a/backend/pkg/observability/langfuse/observer_test.go b/backend/pkg/observability/langfuse/observer_test.go index 46b595645..47b41f519 100644 --- a/backend/pkg/observability/langfuse/observer_test.go +++ b/backend/pkg/observability/langfuse/observer_test.go @@ -47,8 +47,12 @@ func observerNew(t *testing.T, url string) *observer { ).(*observer) } -func observerEvent(id string) *api.IngestionEvent { - return &api.IngestionEvent{IngestionEventZero: &api.IngestionEventZero{ID: id, Timestamp: getCurrentTimeString()}} +func observerEvent(id string) *api.OtelSpan { + return &api.OtelSpan{ + TraceID: newTraceID(), + SpanID: id, + Name: getStringRef("test"), + } } // observerWaitBatched waits until the sender has moved every queued event into its in-memory batch. @@ -69,9 +73,15 @@ func observerCountingSink(t *testing.T) (*httptest.Server, *atomic.Int32) { t.Helper() var hits atomic.Int32 - srv := observerSink(t, func(w http.ResponseWriter, _ *http.Request) { + srv := observerSink(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/public/otel/v1/traces" { + t.Errorf("unexpected ingestion path %q, want /api/public/otel/v1/traces", r.URL.Path) + } + if got := r.Header.Get(IngestionVersionHeader); got != IngestionVersion { + t.Errorf("header %s = %q, want %q", IngestionVersionHeader, got, IngestionVersion) + } hits.Add(1) - _, _ = w.Write([]byte(`{"successes":[],"errors":[]}`)) + _, _ = w.Write([]byte(`{}`)) }) return srv, &hits } @@ -124,14 +134,21 @@ func TestObserver_FlushWithSplit_SplitsA413InsteadOfDroppingTheBatch(t *testing. w.WriteHeader(http.StatusInternalServerError) return } - var req api.IngestionBatchRequest + var req api.OpentelemetryExportTracesRequest if err := json.Unmarshal(body, &req); err != nil { - t.Errorf("failed to unmarshal batch request: %v", err) + t.Errorf("failed to unmarshal OTLP traces request: %v", err) w.WriteHeader(http.StatusInternalServerError) return } - if len(req.Batch) > maxEventsPerRequest { + var spans []*api.OtelSpan + for _, rs := range req.ResourceSpans { + for _, ss := range rs.ScopeSpans { + spans = append(spans, ss.Spans...) + } + } + + if len(spans) > maxEventsPerRequest { rejectCount.Add(1) w.WriteHeader(http.StatusRequestEntityTooLarge) _, _ = w.Write([]byte("Body exceeded 4.5mb limit")) @@ -139,14 +156,14 @@ func TestObserver_FlushWithSplit_SplitsA413InsteadOfDroppingTheBatch(t *testing. } mu.Lock() - for _, event := range req.Batch { - if zero := event.GetIngestionEventZero(); zero != nil { - received[zero.ID] = true + for _, span := range spans { + if spanID, ok := span.SpanID.(string); ok { + received[spanID] = true } } mu.Unlock() - _, _ = w.Write([]byte(`{"successes":[],"errors":[]}`)) + _, _ = w.Write([]byte(`{}`)) }) o := observerNew(t, srv.URL) @@ -190,7 +207,7 @@ func TestObserver_FlushWithSplit_DropsAnOversizedSingleEvent(t *testing.T) { done := make(chan error, 1) go func() { - done <- o.flushWithSplit(context.Background(), []*api.IngestionEvent{observerEvent(newSpanID()), observerEvent(newSpanID())}) + done <- o.flushWithSplit(context.Background(), []*api.OtelSpan{observerEvent(newSpanID()), observerEvent(newSpanID())}) }() select { @@ -204,3 +221,325 @@ func TestObserver_FlushWithSplit_DropsAnOversizedSingleEvent(t *testing.T) { _ = o.Shutdown(context.Background()) } + +func TestObserver_V4Observations_SingleExportOnEndAndPropagatesTraceContext(t *testing.T) { + var ( + mu sync.Mutex + spans []*api.OtelSpan + ) + srv := observerSink(t, func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("failed to read request body: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + var req api.OpentelemetryExportTracesRequest + if err := json.Unmarshal(body, &req); err != nil { + t.Errorf("failed to unmarshal OTLP traces request: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + mu.Lock() + for _, rs := range req.ResourceSpans { + for _, ss := range rs.ScopeSpans { + spans = append(spans, ss.Spans...) + } + } + mu.Unlock() + _, _ = w.Write([]byte(`{}`)) + }) + + client, err := NewClient(WithBaseURL(srv.URL), WithPublicKey("pk"), WithSecretKey("sk"), WithProjectID("proj")) + if err != nil { + t.Fatalf("NewClient: %v", err) + } + o := NewObserver(client, + WithRelease("1.2.3"), + WithEnvironment("staging"), + WithSendInterval(10*time.Minute), + WithSendTimeout(2*time.Second), + WithQueueSize(100), + ).(*observer) + defer func() { _ = o.Shutdown(context.Background()) }() + + ctx, rootObs := o.NewObservation(context.Background(), + WithObservationTraceID("trace-v4"), + WithObservationTraceContext( + WithTraceName("flow-1"), + WithTraceUserID("alice"), + WithTraceSessionID("sess-1"), + WithTraceTags([]string{"controller", "flow"}), + WithTraceInput("initial user input"), + WithTraceMetadata(Metadata{"flow_id": 1}), + ), + ) + + rootSpan := rootObs.Span(WithSpanName("root-span")) + ctx, childObs := rootSpan.Observation(ctx) + gen := childObs.Generation( + WithGenerationName("llm-call"), + WithGenerationModel("gpt-4o"), + WithGenerationInput("prompt"), + ) + + // Before End() is called, nothing should be queued (immutable v4 single-export semantics). + if len(o.queue) != 0 { + t.Fatalf("open observations must not be queued before End(); got %d queued items", len(o.queue)) + } + + gen.End( + WithGenerationOutput("response"), + WithGenerationUsage(&GenerationUsage{Input: 10, Output: 5, Unit: GenerationUsageUnitTokens}), + ) + // Calling End() a second time must be a no-op. + gen.End(WithGenerationOutput("duplicate")) + + rootSpan.End(WithSpanOutput("done")) + observerWaitBatched(t, o) + if err := o.ForceFlush(context.Background()); err != nil { + t.Fatalf("ForceFlush: %v", err) + } + + mu.Lock() + defer mu.Unlock() + if len(spans) != 2 { + t.Fatalf("expected exactly 2 exported spans (1 generation, 1 root span), got %d", len(spans)) + } + + attrMap := func(s *api.OtelSpan) map[string]any { + m := make(map[string]any, len(s.Attributes)) + for _, a := range s.Attributes { + if a.Key == nil || a.Value == nil { + continue + } + switch { + case a.Value.StringValue != nil: + m[*a.Key] = *a.Value.StringValue + case a.Value.IntValue != nil: + m[*a.Key] = *a.Value.IntValue + case a.Value.DoubleValue != nil: + m[*a.Key] = *a.Value.DoubleValue + case a.Value.BoolValue != nil: + m[*a.Key] = *a.Value.BoolValue + } + } + return m + } + + genAttrs := attrMap(spans[0]) + rootAttrs := attrMap(spans[1]) + + // Both root and child generation must carry propagated trace context attributes. + for i, attrs := range []map[string]any{genAttrs, rootAttrs} { + if attrs[otelAttrTraceName] != "flow-1" { + t.Errorf("span[%d] %s = %v, want flow-1", i, otelAttrTraceName, attrs[otelAttrTraceName]) + } + if attrs[otelAttrUserID] != "alice" { + t.Errorf("span[%d] %s = %v, want alice", i, otelAttrUserID, attrs[otelAttrUserID]) + } + if attrs[otelAttrSessionID] != "sess-1" { + t.Errorf("span[%d] %s = %v, want sess-1", i, otelAttrSessionID, attrs[otelAttrSessionID]) + } + if attrs[otelAttrRelease] != "1.2.3" { + t.Errorf("span[%d] %s = %v, want 1.2.3", i, otelAttrRelease, attrs[otelAttrRelease]) + } + if attrs[otelAttrEnvironment] != "staging" { + t.Errorf("span[%d] %s = %v, want staging", i, otelAttrEnvironment, attrs[otelAttrEnvironment]) + } + if attrs[otelAttrTraceMetadataPrefix+"flow_id"] != 1 { + t.Errorf("span[%d] trace metadata flow_id = %v, want 1", i, attrs[otelAttrTraceMetadataPrefix+"flow_id"]) + } + } + + // Root span inherits trace input when its own input is unset. + if rootAttrs[otelAttrObservationInput] != "initial user input" { + t.Errorf("root span input = %v, want 'initial user input'", rootAttrs[otelAttrObservationInput]) + } + if genAttrs[otelAttrObservationType] != "generation" { + t.Errorf("generation observation type = %v, want generation", genAttrs[otelAttrObservationType]) + } + if genAttrs[otelAttrObservationModelName] != "gpt-4o" { + t.Errorf("generation model = %v, want gpt-4o", genAttrs[otelAttrObservationModelName]) + } +} + +func TestObserver_AllObservationTypesAndScoresV4(t *testing.T) { + var mu sync.Mutex + var spans []*api.OtelSpan + var scores []*api.CreateScoreRequest + + srv := observerSink(t, func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get(IngestionVersionHeader); got != IngestionVersion { + t.Errorf("header %s = %q, want %q", IngestionVersionHeader, got, IngestionVersion) + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + switch r.URL.Path { + case "/api/public/otel/v1/traces": + var req api.OpentelemetryExportTracesRequest + if err := json.Unmarshal(body, &req); err != nil { + t.Fatalf("unmarshal otel request: %v", err) + } + mu.Lock() + for _, rs := range req.ResourceSpans { + for _, ss := range rs.ScopeSpans { + spans = append(spans, ss.Spans...) + } + } + mu.Unlock() + _, _ = w.Write([]byte(`{}`)) + case "/api/public/scores": + var req api.CreateScoreRequest + if err := json.Unmarshal(body, &req); err != nil { + t.Fatalf("unmarshal score request: %v", err) + } + mu.Lock() + scores = append(scores, &req) + mu.Unlock() + _, _ = w.Write([]byte(`{"id":"score-1"}`)) + default: + t.Errorf("unexpected path %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } + }) + + client, err := NewClient(WithBaseURL(srv.URL), WithPublicKey("pk"), WithSecretKey("sk"), WithProjectID("proj")) + if err != nil { + t.Fatalf("NewClient: %v", err) + } + + obs := NewObserver(client, + WithRelease("1.2.3"), + WithEnvironment("test-env"), + WithSendInterval(10*time.Minute), + WithSendTimeout(2*time.Second), + WithQueueSize(100), + ).(*observer) + defer func() { _ = obs.Shutdown(context.Background()) }() + + traceID := newTraceID() + ctx, rootObs := obs.NewObservation(context.Background(), + WithObservationTraceID(traceID), + WithObservationTraceContext( + WithTraceName("v4-trace"), + WithTraceUserID("user-1"), + WithTraceSessionID("session-1"), + WithTraceTags([]string{"v4", "unit"}), + WithTraceInput(map[string]any{"task": "verify langfuse v4"}), + WithTraceMetadata(Metadata{"flow_id": 42, "tenant": "test"}), + WithTracePublic(), + ), + ) + + rootSpan := rootObs.Span( + WithSpanName("flow-root-span"), + WithSpanInput(map[string]any{"task": "verify langfuse v4"}), + WithSpanMetadata(Metadata{"component": "controller"}), + ) + ctx, spanObs := rootSpan.Observation(ctx) + + agent := spanObs.Agent( + WithAgentName("pentester-agent"), + WithAgentInput("run recon"), + WithAgentMetadata(Metadata{"agent_type": "pentester"}), + ) + ctx, agentObs := agent.Observation(ctx) + + temp := 0.2 + maxTok := 1024 + inCost := 0.0015 + outCost := 0.0030 + gen := agentObs.Generation( + WithGenerationName("openai-generation"), + WithGenerationModel("gpt-4o-mini"), + WithGenerationModelParameters(&ModelParameters{ + Temperature: &temp, + MaxTokens: &maxTok, + }), + WithGenerationInput([]map[string]string{{"role": "user", "content": "ping"}}), + WithGenerationPromptName("pentester-system"), + WithGenerationPromptVersion(3), + ) + gen.End( + WithGenerationOutput(map[string]string{"role": "assistant", "content": "pong"}), + WithGenerationUsage(&GenerationUsage{ + Input: 150, + Output: 50, + InputCost: &inCost, + OutputCost: &outCost, + Unit: GenerationUsageUnitTokens, + }), + WithGenerationStatus("ok"), + ) + + tool := agentObs.Tool(WithToolName("terminal-tool"), WithToolInput(map[string]string{"cmd": "whoami"})) + tool.End(WithToolOutput("root"), WithToolStatus("completed")) + + emb := agentObs.Embedding(WithEmbeddingName("memory-embedding"), WithEmbeddingModel("text-embedding-3-small"), WithEmbeddingInput("recon summary")) + emb.End(WithEmbeddingOutput([]float64{0.1, 0.2, 0.3}), WithEmbeddingStatus("embedded")) + + agentObs.Event(WithEventName("checkpoint-event"), WithEventInput("checkpoint reached"), WithEventOutput("saved"), WithEventMetadata(Metadata{"step": 1})) + + chain := agentObs.Chain(WithChainName("summarizer-chain"), WithChainInput("long conversation")) + chain.End(WithChainOutput("short summary"), WithChainStatus("summarized")) + + retriever := agentObs.Retriever(WithRetrieverName("vector-retriever"), WithRetrieverInput("query vector")) + retriever.End(WithRetrieverOutput([]string{"doc-1", "doc-2"}), WithRetrieverStatus("retrieved")) + + evaluator := agentObs.Evaluator(WithEvaluatorName("safety-evaluator"), WithEvaluatorInput("candidate output")) + evaluator.End(WithEvaluatorOutput(map[string]any{"passed": true}), WithEvaluatorStatus("evaluated")) + + guardrail := agentObs.Guardrail(WithGuardrailName("scope-guardrail"), WithGuardrailInput("target ip")) + guardrail.End(WithGuardrailOutput("allowed"), WithGuardrailLevel(ObservationLevelWarning), WithGuardrailStatus("warned")) + + agentObs.Score(WithScoreName("quality"), WithScoreFloatValue(0.95), WithScoreComment("score verification")) + agentObs.Score(WithScoreName("verdict"), WithScoreStringValue("pass"), WithScoreComment("categorical score verification")) + + agent.End(WithAgentOutput("recon complete"), WithAgentStatus("done")) + rootSpan.End(WithSpanOutput(map[string]any{"result": "success"}), WithSpanStatus("completed")) + + observerWaitBatched(t, obs) + if err := obs.ForceFlush(context.Background()); err != nil { + t.Fatalf("ForceFlush failed: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + if len(spans) != 10 { + t.Fatalf("expected 10 exported OTLP spans, got %d", len(spans)) + } + if len(scores) != 2 { + t.Fatalf("expected 2 exported scores, got %d", len(scores)) + } + + seenTypes := make(map[string]bool) + for _, s := range spans { + m := make(map[string]any, len(s.Attributes)) + for _, a := range s.Attributes { + if a.Key != nil && a.Value != nil && a.Value.StringValue != nil { + m[*a.Key] = *a.Value.StringValue + } + } + obsType, _ := m[otelAttrObservationType].(string) + seenTypes[obsType] = true + if m[otelAttrUserID] != "user-1" { + t.Errorf("span %s (%s) %s = %v, want user-1", s.SpanID, obsType, otelAttrUserID, m[otelAttrUserID]) + } + if m[otelAttrSessionID] != "session-1" { + t.Errorf("span %s (%s) %s = %v, want session-1", s.SpanID, obsType, otelAttrSessionID, m[otelAttrSessionID]) + } + if m[otelAttrEnvironment] != "test-env" { + t.Errorf("span %s (%s) %s = %v, want test-env", s.SpanID, obsType, otelAttrEnvironment, m[otelAttrEnvironment]) + } + } + + for _, wantType := range []string{"span", "agent", "generation", "tool", "embedding", "event", "chain", "retriever", "evaluator", "guardrail"} { + if !seenTypes[wantType] { + t.Errorf("missing observation type %q in exported spans", wantType) + } + } +} diff --git a/backend/pkg/observability/langfuse/retriever.go b/backend/pkg/observability/langfuse/retriever.go index 9bacdeac3..c80bdc402 100644 --- a/backend/pkg/observability/langfuse/retriever.go +++ b/backend/pkg/observability/langfuse/retriever.go @@ -4,10 +4,9 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - "pentagi/pkg/observability/langfuse/api" - "github.com/vxcontrol/langchaingo/llms" ) @@ -41,7 +40,9 @@ type retriever struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type RetrieverOption func(*retriever) @@ -58,6 +59,12 @@ func withRetrieverParentObservationID(parentObservationID string) RetrieverOptio } } +func withRetrieverTraceContext(traceCtx *TraceContext) RetrieverOption { + return func(r *retriever) { + r.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithRetrieverID sets on creation time func WithRetrieverID(id string) RetrieverOption { return func(r *retriever) { @@ -151,65 +158,43 @@ func newRetriever(observer enqueue, opts ...RetrieverOption) Retriever { opt(r) } - obsCreate := &api.IngestionEvent{IngestionEventThirteen: &api.IngestionEventThirteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(r.StartTime), - Type: api.IngestionEventThirteenType(ingestionCreateRetriever).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(r.ObservationID), - TraceID: getStringRef(r.TraceID), - ParentObservationID: getStringRef(r.ParentObservationID), - Name: getStringRef(r.Name), - Metadata: r.Metadata, - Input: convertInput(r.Input, r.Tools), - Output: convertOutput(r.Output), - StartTime: r.StartTime, - EndTime: r.EndTime, - Level: r.Level.ToLangfuse(), - StatusMessage: r.Status, - Version: r.Version, - Model: r.Model, - ModelParameters: r.ModelParameters.ToLangfuse(), - }, - }} - - r.observer.enqueue(obsCreate) - return r } func (r *retriever) End(opts ...RetrieverOption) { - id := r.ObservationID - startTime := r.StartTime - r.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(r) - } - - // preserve the original observation ID and start time - r.ObservationID = id - r.StartTime = startTime - - retrieverUpdate := &api.IngestionEvent{IngestionEventThirteen: &api.IngestionEventThirteen{ - ID: newSpanID(), - Timestamp: getTimeRefString(r.EndTime), - Type: api.IngestionEventThirteenType(ingestionCreateRetriever).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(r.ObservationID), - Name: getStringRef(r.Name), - Metadata: r.Metadata, - Input: convertInput(r.Input, r.Tools), - Output: convertOutput(r.Output), - EndTime: r.EndTime, - Level: r.Level.ToLangfuse(), - StatusMessage: r.Status, - Version: r.Version, - Model: r.Model, - ModelParameters: r.ModelParameters.ToLangfuse(), - }, - }} + r.endOnce.Do(func() { + id := r.ObservationID + startTime := r.StartTime + r.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(r) + } + + // preserve the original observation ID and start time + r.ObservationID = id + r.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: r.TraceID, + ObservationID: r.ObservationID, + ParentObservationID: r.ParentObservationID, + Name: r.Name, + ObservationType: otelObservationTypeRetriever, + StartTime: r.StartTime, + EndTime: r.EndTime, + Level: r.Level, + Status: r.Status, + Version: r.Version, + Input: convertInput(r.Input, r.Tools), + Output: convertOutput(r.Output), + Metadata: r.Metadata, + Model: r.Model, + ModelParameters: r.ModelParameters, + TraceCtx: r.traceCtx, + }) - r.observer.enqueue(retrieverUpdate) + r.observer.enqueue(otelSpan) + }) } func (r *retriever) String() string { @@ -221,10 +206,15 @@ func (r *retriever) MarshalJSON() ([]byte, error) { } func (r *retriever) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := r.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: r.TraceID, ObservationID: r.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: r.observer, } diff --git a/backend/pkg/observability/langfuse/score.go b/backend/pkg/observability/langfuse/score.go index c8da45680..b7152b083 100644 --- a/backend/pkg/observability/langfuse/score.go +++ b/backend/pkg/observability/langfuse/score.go @@ -113,23 +113,18 @@ func newScore(observer enqueue, opts ...ScoreOption) { opt(s) } - obsCreate := &api.IngestionEvent{IngestionEventOne: &api.IngestionEventOne{ - ID: newSpanID(), - Timestamp: getTimeRefString(s.StartTime), - Type: api.IngestionEventOneType(ingestionCreateScore).Ptr(), - Body: &api.ScoreBody{ - ID: getStringRef(s.ObservationID), - ObservationID: getStringRef(s.ParentObservationID), - TraceID: getStringRef(s.TraceID), - Name: s.Name, - Metadata: s.Metadata, - Value: s.Value, - DataType: s.DataType, - Comment: s.Comment, - ConfigID: s.ConfigID, - QueueID: s.QueueID, - }, - }} - - s.observer.enqueue(obsCreate) + scoreReq := &api.CreateScoreRequest{ + ID: getStringRef(s.ObservationID), + ObservationID: getStringRef(s.ParentObservationID), + TraceID: getStringRef(s.TraceID), + Name: s.Name, + Metadata: s.Metadata, + Value: s.Value, + DataType: s.DataType, + Comment: s.Comment, + ConfigID: s.ConfigID, + QueueID: s.QueueID, + } + + s.observer.enqueueScore(scoreReq) } diff --git a/backend/pkg/observability/langfuse/span.go b/backend/pkg/observability/langfuse/span.go index d146b33f8..8df7b6668 100644 --- a/backend/pkg/observability/langfuse/span.go +++ b/backend/pkg/observability/langfuse/span.go @@ -4,9 +4,8 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - - "pentagi/pkg/observability/langfuse/api" ) const ( @@ -36,7 +35,9 @@ type span struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type SpanOption func(*span) @@ -53,6 +54,12 @@ func withSpanParentObservationID(parentObservationID string) SpanOption { } } +func withSpanTraceContext(traceCtx *TraceContext) SpanOption { + return func(s *span) { + s.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithSpanID sets on creation time func WithSpanID(id string) SpanOption { return func(s *span) { @@ -128,61 +135,41 @@ func newSpan(observer enqueue, opts ...SpanOption) Span { opt(s) } - obsCreate := &api.IngestionEvent{IngestionEventTwo: &api.IngestionEventTwo{ - ID: newSpanID(), - Timestamp: getTimeRefString(s.StartTime), - Type: api.IngestionEventTwoType(ingestionCreateSpan).Ptr(), - Body: &api.CreateSpanBody{ - ID: getStringRef(s.ObservationID), - TraceID: getStringRef(s.TraceID), - ParentObservationID: getStringRef(s.ParentObservationID), - Name: getStringRef(s.Name), - Input: convertInput(s.Input, nil), - Output: convertOutput(s.Output), - StartTime: s.StartTime, - EndTime: s.EndTime, - Metadata: s.Metadata, - Level: s.Level.ToLangfuse(), - StatusMessage: s.Status, - Version: s.Version, - }, - }} - - s.observer.enqueue(obsCreate) - return s } func (s *span) End(opts ...SpanOption) { - id := s.ObservationID - startTime := s.StartTime - s.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(s) - } - - // preserve the original observation ID and start time - s.ObservationID = id - s.StartTime = startTime - - obsUpdate := &api.IngestionEvent{IngestionEventThree: &api.IngestionEventThree{ - ID: newSpanID(), - Timestamp: getTimeRefString(s.EndTime), - Type: api.IngestionEventThreeType(ingestionUpdateSpan).Ptr(), - Body: &api.UpdateSpanBody{ - ID: s.ObservationID, - Name: getStringRef(s.Name), - Metadata: s.Metadata, - Input: convertInput(s.Input, nil), - Output: convertOutput(s.Output), - EndTime: s.EndTime, - Level: s.Level.ToLangfuse(), - StatusMessage: s.Status, - Version: s.Version, - }, - }} + s.endOnce.Do(func() { + id := s.ObservationID + startTime := s.StartTime + s.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(s) + } + + // preserve the original observation ID and start time + s.ObservationID = id + s.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: s.TraceID, + ObservationID: s.ObservationID, + ParentObservationID: s.ParentObservationID, + Name: s.Name, + ObservationType: otelObservationTypeSpan, + StartTime: s.StartTime, + EndTime: s.EndTime, + Level: s.Level, + Status: s.Status, + Version: s.Version, + Input: convertInput(s.Input, nil), + Output: convertOutput(s.Output), + Metadata: s.Metadata, + TraceCtx: s.traceCtx, + }) - s.observer.enqueue(obsUpdate) + s.observer.enqueue(otelSpan) + }) } func (s *span) String() string { @@ -194,10 +181,15 @@ func (s *span) MarshalJSON() ([]byte, error) { } func (s *span) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := s.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: s.TraceID, ObservationID: s.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: s.observer, } diff --git a/backend/pkg/observability/langfuse/tool.go b/backend/pkg/observability/langfuse/tool.go index 5d3738517..fd0a1a340 100644 --- a/backend/pkg/observability/langfuse/tool.go +++ b/backend/pkg/observability/langfuse/tool.go @@ -4,9 +4,8 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" - - "pentagi/pkg/observability/langfuse/api" ) const ( @@ -36,7 +35,9 @@ type tool struct { ObservationID string `json:"observation_id"` ParentObservationID string `json:"parent_observation_id"` - observer enqueue `json:"-"` + traceCtx *TraceContext `json:"-"` + endOnce sync.Once `json:"-"` + observer enqueue `json:"-"` } type ToolOption func(*tool) @@ -53,6 +54,12 @@ func withToolParentObservationID(parentObservationID string) ToolOption { } } +func withToolTraceContext(traceCtx *TraceContext) ToolOption { + return func(t *tool) { + t.traceCtx = cloneTraceContext(traceCtx) + } +} + // WithToolID sets on creation time func WithToolID(id string) ToolOption { return func(t *tool) { @@ -128,61 +135,41 @@ func newTool(observer enqueue, opts ...ToolOption) Tool { opt(t) } - obsCreate := &api.IngestionEvent{IngestionEventEleven: &api.IngestionEventEleven{ - ID: newSpanID(), - Timestamp: getTimeRefString(t.StartTime), - Type: api.IngestionEventElevenType(ingestionCreateTool).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(t.ObservationID), - TraceID: getStringRef(t.TraceID), - ParentObservationID: getStringRef(t.ParentObservationID), - Name: getStringRef(t.Name), - Metadata: t.Metadata, - Input: convertInput(t.Input, nil), - Output: convertOutput(t.Output), - StartTime: t.StartTime, - EndTime: t.EndTime, - Level: t.Level.ToLangfuse(), - StatusMessage: t.Status, - Version: t.Version, - }, - }} - - t.observer.enqueue(obsCreate) - return t } func (t *tool) End(opts ...ToolOption) { - id := t.ObservationID - startTime := t.StartTime - t.EndTime = getCurrentTimeRef() - for _, opt := range opts { - opt(t) - } - - // preserve the original observation ID and start time - t.ObservationID = id - t.StartTime = startTime - - toolUpdate := &api.IngestionEvent{IngestionEventEleven: &api.IngestionEventEleven{ - ID: newSpanID(), - Timestamp: getTimeRefString(t.EndTime), - Type: api.IngestionEventElevenType(ingestionCreateTool).Ptr(), - Body: &api.CreateGenerationBody{ - ID: getStringRef(t.ObservationID), - Name: getStringRef(t.Name), - Metadata: t.Metadata, - Input: convertInput(t.Input, nil), - Output: convertOutput(t.Output), - EndTime: t.EndTime, - Level: t.Level.ToLangfuse(), - StatusMessage: t.Status, - Version: t.Version, - }, - }} + t.endOnce.Do(func() { + id := t.ObservationID + startTime := t.StartTime + t.EndTime = getCurrentTimeRef() + for _, opt := range opts { + opt(t) + } + + // preserve the original observation ID and start time + t.ObservationID = id + t.StartTime = startTime + + otelSpan := buildOtelSpan(otelSpanParams{ + TraceID: t.TraceID, + ObservationID: t.ObservationID, + ParentObservationID: t.ParentObservationID, + Name: t.Name, + ObservationType: otelObservationTypeTool, + StartTime: t.StartTime, + EndTime: t.EndTime, + Level: t.Level, + Status: t.Status, + Version: t.Version, + Input: convertInput(t.Input, nil), + Output: convertOutput(t.Output), + Metadata: t.Metadata, + TraceCtx: t.traceCtx, + }) - t.observer.enqueue(toolUpdate) + t.observer.enqueue(otelSpan) + }) } func (t *tool) String() string { @@ -194,10 +181,15 @@ func (t *tool) MarshalJSON() ([]byte, error) { } func (t *tool) Observation(ctx context.Context) (context.Context, Observation) { + traceCtx := t.traceCtx + if parent, ok := getObservationContext(ctx); ok && parent.TraceCtx != nil { + traceCtx = mergeTraceContext(parent.TraceCtx, traceCtx) + } obs := &observation{ obsCtx: observationContext{ TraceID: t.TraceID, ObservationID: t.ObservationID, + TraceCtx: cloneTraceContext(traceCtx), }, observer: t.observer, } diff --git a/backend/pkg/observability/langfuse/trace.go b/backend/pkg/observability/langfuse/trace.go index 71f131f2c..fdd3eaa8c 100644 --- a/backend/pkg/observability/langfuse/trace.go +++ b/backend/pkg/observability/langfuse/trace.go @@ -1,6 +1,10 @@ package langfuse -import "time" +import ( + "maps" + "slices" + "time" +) type TraceContext struct { Timestamp *time.Time `json:"timestamp,omitempty"` @@ -15,6 +19,61 @@ type TraceContext struct { Public *bool `json:"public,omitempty"` } +func cloneTraceContext(src *TraceContext) *TraceContext { + if src == nil { + return nil + } + cp := *src + if src.Metadata != nil { + cp.Metadata = maps.Clone(src.Metadata) + } + if src.Tags != nil { + cp.Tags = slices.Clone(src.Tags) + } + return &cp +} + +func mergeTraceContext(base, update *TraceContext) *TraceContext { + if update == nil { + return cloneTraceContext(base) + } + if base == nil { + return cloneTraceContext(update) + } + merged := cloneTraceContext(base) + if update.Timestamp != nil { + merged.Timestamp = update.Timestamp + } + if update.Name != nil { + merged.Name = update.Name + } + if update.UserID != nil { + merged.UserID = update.UserID + } + if update.Input != nil { + merged.Input = update.Input + } + if update.Output != nil { + merged.Output = update.Output + } + if update.SessionID != nil { + merged.SessionID = update.SessionID + } + if update.Version != nil { + merged.Version = update.Version + } + if update.Metadata != nil { + merged.Metadata = mergeMaps(merged.Metadata, update.Metadata) + } + if len(update.Tags) > 0 { + merged.Tags = slices.Clone(update.Tags) + } + if update.Public != nil { + merged.Public = update.Public + } + return merged +} + type TraceContextOption func(*TraceContext) func WithTraceTimestamp(timestamp time.Time) TraceContextOption { diff --git a/backend/pkg/tools/executor_test.go b/backend/pkg/tools/executor_test.go index b7448a9ce..4f9f70be6 100644 --- a/backend/pkg/tools/executor_test.go +++ b/backend/pkg/tools/executor_test.go @@ -114,6 +114,8 @@ type executorLangfuseEvent struct { Metadata map[string]any `json:"metadata"` Level string `json:"level"` StatusMessage string `json:"statusMessage"` + StartTimeUnixNano string `json:"startTimeUnixNano"` + EndTimeUnixNano string `json:"endTimeUnixNano"` } `json:"body"` } @@ -121,16 +123,80 @@ func (l *executorLangfuse) RoundTrip(r *http.Request) (*http.Response, error) { answer := `{"data":[{"id":"project","name":"executor"}]}` if r.Method == http.MethodPost { defer r.Body.Close() - var ingestion struct { - Batch []executorLangfuseEvent `json:"batch"` + var otlp struct { + ResourceSpans []struct { + ScopeSpans []struct { + Spans []struct { + TraceID string `json:"traceId"` + SpanID string `json:"spanId"` + ParentSpanID string `json:"parentSpanId"` + Name string `json:"name"` + StartTimeUnixNano string `json:"startTimeUnixNano"` + EndTimeUnixNano string `json:"endTimeUnixNano"` + Attributes []struct { + Key string `json:"key"` + Value struct { + StringValue *string `json:"stringValue"` + IntValue *float64 `json:"intValue"` + DoubleValue *float64 `json:"doubleValue"` + BoolValue *bool `json:"boolValue"` + } `json:"value"` + } `json:"attributes"` + } `json:"spans"` + } `json:"scopeSpans"` + } `json:"resourceSpans"` } - if err := json.NewDecoder(r.Body).Decode(&ingestion); err != nil { + if err := json.NewDecoder(r.Body).Decode(&otlp); err != nil { return nil, err } l.mu.Lock() - l.events = append(l.events, ingestion.Batch...) + for _, rs := range otlp.ResourceSpans { + for _, ss := range rs.ScopeSpans { + for _, s := range ss.Spans { + ev := executorLangfuseEvent{} + ev.Body.ID = s.SpanID + ev.Body.TraceID = s.TraceID + ev.Body.ParentObservationID = s.ParentSpanID + ev.Body.Name = s.Name + ev.Body.StartTimeUnixNano = s.StartTimeUnixNano + ev.Body.EndTimeUnixNano = s.EndTimeUnixNano + ev.Body.Metadata = make(map[string]any) + for _, attr := range s.Attributes { + var val any + switch { + case attr.Value.StringValue != nil: + val = *attr.Value.StringValue + case attr.Value.IntValue != nil: + val = *attr.Value.IntValue + case attr.Value.DoubleValue != nil: + val = *attr.Value.DoubleValue + case attr.Value.BoolValue != nil: + val = *attr.Value.BoolValue + } + switch { + case attr.Key == "langfuse.observation.type": + if str, ok := val.(string); ok { + ev.Type = str + "-create" + } + case attr.Key == "langfuse.observation.level": + if str, ok := val.(string); ok { + ev.Body.Level = str + } + case attr.Key == "langfuse.observation.status_message": + if str, ok := val.(string); ok { + ev.Body.StatusMessage = str + } + case strings.HasPrefix(attr.Key, "langfuse.observation.metadata."): + metaKey := strings.TrimPrefix(attr.Key, "langfuse.observation.metadata.") + ev.Body.Metadata[metaKey] = val + } + } + l.events = append(l.events, ev) + } + } + } l.mu.Unlock() - answer = `{"successes":[],"errors":[]}` + answer = `{}` } return &http.Response{ @@ -915,22 +981,26 @@ func TestExecutor_Execute_ObservesEachToolTypeUnderTheCallersTrace(t *testing.T) } } - // One goroutine sends events in the order Execute made them, so an observation's second event is its close. + // In Langfuse v4, each observation is exported once when completed with both start and end timestamps. var lifecycle []executorLangfuseEvent for _, event := range sent { if event.Body.ID == id { lifecycle = append(lifecycle, event) } } - if len(lifecycle) != 2 { - t.Errorf("call %d: observation %q was sent %d times, want once opened and once closed", + if len(lifecycle) != 1 { + t.Errorf("call %d: observation %q was sent %d times, want once on completion", i+1, id, len(lifecycle)) continue } - if closing := lifecycle[1].Body; closing.Level != want.level || closing.StatusMessage != want.status { + if closing := lifecycle[0].Body; closing.Level != want.level || closing.StatusMessage != want.status { t.Errorf("call %d: the observation was closed at level %q with %q, want %q with %q", i+1, closing.Level, closing.StatusMessage, want.level, want.status) } + if closing := lifecycle[0].Body; closing.StartTimeUnixNano == "" || closing.EndTimeUnixNano == "" { + t.Errorf("call %d: the observation must include both start and end timestamps, got start=%q end=%q", + i+1, closing.StartTimeUnixNano, closing.EndTimeUnixNano) + } } }) } diff --git a/docker-compose-langfuse.yml b/docker-compose-langfuse.yml index 06a18bb4b..775d3637c 100644 --- a/docker-compose-langfuse.yml +++ b/docker-compose-langfuse.yml @@ -20,7 +20,7 @@ networks: services: langfuse-worker: - image: ${LANGFUSE_WORKER_IMAGE:-langfuse/langfuse-worker:3} + image: ${LANGFUSE_WORKER_IMAGE:-langfuse/langfuse-worker:4} restart: unless-stopped container_name: langfuse-worker hostname: langfuse-worker @@ -36,6 +36,10 @@ services: expose: - 3030/tcp environment: &langfuse-worker-env + LANGFUSE_MIGRATION_V4_WRITE_MODE: ${LANGFUSE_MIGRATION_V4_WRITE_MODE:-events_only} + LANGFUSE_MIGRATION_V4_NATIVE_OTEL_BEHAVIOUR: ${LANGFUSE_MIGRATION_V4_NATIVE_OTEL_BEHAVIOUR:-direct} + LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN: ${LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN:-true} + LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL: ${LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL:-true} NEXTAUTH_URL: ${LANGFUSE_NEXTAUTH_URL:-http://localhost:${LANGFUSE_LISTEN_PORT:-4000}} DATABASE_URL: postgresql://${LANGFUSE_POSTGRES_USER:-postgres}:${LANGFUSE_POSTGRES_PASSWORD:-postgres}@langfuse-postgres:5432/${LANGFUSE_POSTGRES_DB:-langfuse} SALT: ${LANGFUSE_SALT:-myglobalsalt} # change this to a random string @@ -92,7 +96,7 @@ services: - langfuse-network langfuse-web: - image: ${LANGFUSE_WEB_IMAGE:-langfuse/langfuse:3} + image: ${LANGFUSE_WEB_IMAGE:-langfuse/langfuse:4} restart: unless-stopped container_name: langfuse-web hostname: langfuse-web @@ -147,7 +151,7 @@ services: - pentagi-network clickhouse: - image: ${CLICKHOUSE_IMAGE:-clickhouse/clickhouse-server:24} + image: ${CLICKHOUSE_IMAGE:-clickhouse/clickhouse-server:26.4} restart: unless-stopped user: "101:101" container_name: langfuse-clickhouse