From bd7445dfd3ab11bcb2129b0a3629c02435726974 Mon Sep 17 00:00:00 2001 From: fcy <3181088318@qq.com> Date: Mon, 31 Aug 2026 10:09:38 +0800 Subject: [PATCH] fix(tracing): report streaming errors to generation spans --- pkg/tracing/traced_llm.go | 41 ++++++-- pkg/tracing/traced_llm_error_test.go | 142 +++++++++++++++++++++++++++ 2 files changed, 174 insertions(+), 9 deletions(-) create mode 100644 pkg/tracing/traced_llm_error_test.go diff --git a/pkg/tracing/traced_llm.go b/pkg/tracing/traced_llm.go index ad87573..9ddfac3 100644 --- a/pkg/tracing/traced_llm.go +++ b/pkg/tracing/traced_llm.go @@ -287,12 +287,20 @@ func (m *TracedLLM) GenerateWithToolsStream(ctx context.Context, prompt string, // Start a goroutine to proxy events and handle span completion go func() { defer close(wrappedChan) + + var streamErr error + defer func() { // When streaming is complete, create tool call spans and end main span endTime := time.Now() duration := endTime.Sub(startTime) span.SetAttribute("duration_ms", duration.Milliseconds()) + // Record streaming error on span if one was captured (#328) + if streamErr != nil { + span.RecordError(streamErr) + } + // Get tool calls from context and create spans using TraceGeneration if any exist toolCalls := GetToolCallsFromContext(ctx) @@ -306,25 +314,40 @@ func (m *TracedLLM) GenerateWithToolsStream(ctx context.Context, prompt string, model = streamingLLM.Name() } + // Build metadata, including error if the stream failed mid-flight (#328) + metadata := map[string]any{ + "streaming": true, + "tools": len(tools), + } + if streamErr != nil { + metadata["error"] = streamErr.Error() + } + + responseText := "streaming_response" + if streamErr != nil { + if m.shouldIncludeContent() { + responseText = "" + } else { + responseText = "" + } + } + // Create spans using TraceGeneration which handles tool calls correctly if adapter, ok := m.tracer.(*OTELTracerAdapter); ok { - _, _ = adapter.otelTracer.TraceGeneration(ctx, model, prompt, "streaming_response", startTime, endTime, map[string]any{ //nolint:gosec - "streaming": true, - "tools": len(tools), - }) + _, _ = adapter.otelTracer.TraceGeneration(ctx, model, prompt, responseText, startTime, endTime, metadata) //nolint:gosec } else if tracer, ok := m.tracer.(*OTELLangfuseTracer); ok { - _, _ = tracer.TraceGeneration(ctx, model, prompt, "streaming_response", startTime, endTime, map[string]any{ //nolint:gosec - "streaming": true, - "tools": len(tools), - }) + _, _ = tracer.TraceGeneration(ctx, model, prompt, responseText, startTime, endTime, metadata) //nolint:gosec } } span.End() }() - // Proxy all events from the original channel + // Proxy all events from the original channel, capturing any error event (#328) for event := range originalChan { + if event.Type == interfaces.StreamEventError && event.Error != nil { + streamErr = event.Error + } wrappedChan <- event } }() diff --git a/pkg/tracing/traced_llm_error_test.go b/pkg/tracing/traced_llm_error_test.go new file mode 100644 index 0000000..e2eb801 --- /dev/null +++ b/pkg/tracing/traced_llm_error_test.go @@ -0,0 +1,142 @@ +package tracing + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/Ingenimax/agent-sdk-go/pkg/interfaces" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" +) + +type errorStreamLLM struct { + streamErr error +} + +func (*errorStreamLLM) Generate(context.Context, string, ...interfaces.GenerateOption) (string, error) { + return "", nil +} + +func (*errorStreamLLM) GenerateWithTools(context.Context, string, []interfaces.Tool, ...interfaces.GenerateOption) (string, error) { + return "", nil +} + +func (*errorStreamLLM) GenerateDetailed(context.Context, string, ...interfaces.GenerateOption) (*interfaces.LLMResponse, error) { + return &interfaces.LLMResponse{}, nil +} + +func (*errorStreamLLM) GenerateWithToolsDetailed(context.Context, string, []interfaces.Tool, ...interfaces.GenerateOption) (*interfaces.LLMResponse, error) { + return &interfaces.LLMResponse{}, nil +} + +func (*errorStreamLLM) GenerateStream(context.Context, string, ...interfaces.GenerateOption) (<-chan interfaces.StreamEvent, error) { + return nil, nil +} + +func (m *errorStreamLLM) GenerateWithToolsStream(ctx context.Context, _ string, _ []interfaces.Tool, _ ...interfaces.GenerateOption) (<-chan interfaces.StreamEvent, error) { + AddToolCallToContext(ctx, ToolCall{Name: "test_tool", Timestamp: time.Now().Format(time.RFC3339)}) + ch := make(chan interfaces.StreamEvent, 1) + if m.streamErr != nil { + ch <- interfaces.StreamEvent{Type: interfaces.StreamEventError, Error: m.streamErr} + } else { + ch <- interfaces.StreamEvent{Type: interfaces.StreamEventContentComplete, Content: "complete"} + } + close(ch) + return ch, nil +} + +func (*errorStreamLLM) Name() string { return "test" } +func (*errorStreamLLM) SupportsStreaming() bool { return true } + +func TestGenerateWithToolsStreamGenerationSpan(t *testing.T) { + for _, tc := range []struct { + name string + includeContent bool + streamErr error + wantError string + wantResponse string + }{ + { + name: "error with content included", + includeContent: true, + streamErr: errors.New("stream interrupted"), + wantError: "stream interrupted", + wantResponse: "", + }, + { + name: "error with content redacted", + includeContent: false, + streamErr: errors.New("stream interrupted"), + wantError: "stream interrupted", + wantResponse: "", + }, + { + name: "successful stream preserves response placeholder", + includeContent: false, + wantResponse: "streaming_response", + }, + } { + t.Run(tc.name, func(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + provider := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exporter)) + t.Cleanup(func() { _ = provider.Shutdown(context.Background()) }) + + otelTracer := &OTELLangfuseTracer{ + tracerProvider: provider, + tracer: provider.Tracer("test"), + enabled: true, + IncludeContent: tc.includeContent, + } + traced := NewTracedLLM(&errorStreamLLM{streamErr: tc.streamErr}, NewOTELTracerAdapter(otelTracer)).(*TracedLLM) + + stream, err := traced.GenerateWithToolsStream(context.Background(), "secret prompt", nil) + if err != nil { + t.Fatalf("GenerateWithToolsStream() error = %v", err) + } + for range stream { + } + + var generation sdktrace.ReadOnlySpan + for _, span := range exporter.GetSpans() { + for _, attr := range span.Attributes { + if string(attr.Key) == "langfuse.observation.type" && attr.Value.AsString() == "generation" { + generation = span.Snapshot() + } + } + } + if generation == nil { + t.Fatal("generation span was not exported") + } + + attributes := make(map[string]string) + for _, attr := range generation.Attributes() { + attributes[string(attr.Key)] = attr.Value.AsString() + } + if got := attributes["langfuse.observation.metadata.error"]; got != tc.wantError { + t.Errorf("error metadata = %q, want %q", got, tc.wantError) + } + if got := attributes["gen_ai.completion.0.content"]; got != tc.wantResponse { + t.Errorf("generation response = %q, want %q", got, tc.wantResponse) + } + + if tc.streamErr != nil { + var foundErrorEvent bool + for _, span := range exporter.GetSpans() { + if span.Name != "llm.generate_with_tools_stream" { + continue + } + for _, event := range span.Events { + if event.Name == "exception" { + foundErrorEvent = true + } + } + } + if !foundErrorEvent { + t.Error("main streaming span did not record the stream error") + } + } + }) + } +}