Spring AI + Langfuse: Tracing Streaming with ObservationFilter

In the first part, we added prompt and response to Langfuse through CallAdvisor.

In many applications, it is necessary to stream the model’s response instead of using a simple .call(). For example, in chat applications, streaming displays the response gradually and makes the interaction with the model more interactive. However, the CallAdvisor approach is not suitable for this because responses returned by stream() do not pass through CallAdvisor.

A new approach

Spring AI already creates observations for both regular and streaming model calls. We only need to add the missing prompt and completion attributes using an ObservationFilter.

The flow is:

1. Spring AI creates an observation for the model call.
2. ObservationFilter adds the prompt and completion.
3. OpenTelemetry exports the trace to Langfuse.

ObservationFilter is used here instead of a streaming advisor because the goal is not to modify the streaming response itself, but to enrich the observation that Spring AI already creates. A streaming advisor works at the ChatClient level and is useful for intercepting or transforming the request and response stream. ObservationFilter, on the other hand, works directly with the observability context, allowing us to add attributes to the existing model span without creating a separate trace or changing the application logic.

Security note: Prompts and completions may contain personal data, secrets, document contents, or tool results. Spring AI does not export this content by default for security and privacy reasons. Enable this tracing only when necessary, and consider redaction, length limits, and environment-specific configuration before using it in production.

Controller

@GetMapping(value = "/ai", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
Flux<String> ai(@RequestParam(defaultValue = "Say hello in one short sentence") String message) {
    return this.chatClient.prompt()
            .user(message)
            .tools(this.weatherTools)
            .stream()
            .content();
}

ObservationFilter

Instead of CallAdvisor, we register an ObservationFilter, that enriches the tracing data. It works with ChatModelObservationContext and adds the prompt and completion to the existing Spring AI observation:

@Component
class LangfuseObservationFilter implements ObservationFilter {

    @Override
    public Observation.Context map(Observation.Context context) {
        if (!(context instanceof ChatModelObservationContext chatModelObservationContext)) {
            return context;
        }

        chatModelObservationContext.addHighCardinalityKeyValue(
                KeyValue.of("gen_ai.prompt", prompt(chatModelObservationContext)));

        chatModelObservationContext.addHighCardinalityKeyValue(
                KeyValue.of("gen_ai.completion", completion(chatModelObservationContext)));

        return chatModelObservationContext;
    }

    private String prompt(ChatModelObservationContext context) {
        return ofNullable(context.getRequest())
                .map(Prompt::getInstructions)
                .orElse(List.of())
                .stream()
                .map(Content::getText)
                .filter(StringUtils::hasText)
                .collect(Collectors.joining("\n"));
    }

    private String completion(ChatModelObservationContext context) {
        return ofNullable(context.getResponse())
                .map(ChatResponse::getResults)
                .orElse(List.of())
                .stream()
                .filter(generation -> generation.getOutput() != null)
                .map(generation -> generation.getOutput().getText())
                .filter(StringUtils::hasText)
                .collect(Collectors.joining("\n"));
    }
}

With automatic tool calling, a single user request may produce multiple model observations. The filter enriches each model span, not only the final response returned by the controller.