Spring AI + Langfuse: Tracing Streaming with ObservationFilter

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

This works well for a simple .call(). But once we switch to streaming, this approach becomes less suitable.

The problem

The problem is that responses from stream() do not pass through CallAdvisor.

Spring AI does have StreamAdvisor, but it is not always the best option for tracing. Advisors are part of the execution flow. They can depend on order, overlap with tool calling, and add unnecessary noise to traces.

For observability, it is better not to interfere with request execution, but to enrich the existing observations instead.

A new approach

Spring AI already creates observations for model calls.

We do not need to create or mark spans manually. We only need to add the missing attributes through ObservationFilter.

The idea is:

1. Spring AI creates an observation.
2. ObservationFilter adds the prompt and completion.
3. OpenTelemetry sends the trace to Langfuse.

Controller with stream()

The endpoint can use stream() internally while still returning a complete string:

@GetMapping("/ai")
String ai(@RequestParam(defaultValue = "Say hello in one short sentence") String message) {
    var response = this.chatClient.prompt()
            .user(message)
            .tools(this.weatherTools)
            .stream()
            .content()
            .reduce(new StringBuilder(), StringBuilder::append)
            .map(StringBuilder::toString)
            .block();

    return response != null ? response : "";
}

ChatClient no longer needs a custom advisor.

ObservationFilter

Instead of CallAdvisor, we add an ObservationFilter.

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"));
    }
}

This filter:

• does not call the model;
• does not change the response;
• does not affect streaming;
• does not depend on advisor order;
• only enriches the tracing data.

Tool calling

With tool calling, one user request can produce several internal model calls:

1. User prompt.
2. The model decides to call a tool.
3. Tool execution.
4. Final model response.

So having several model observations in one trace is normal. They reflect the real execution flow.

This keeps tracing separate from execution logic and fits the Spring AI model better.