Class AiServiceStreamingEventPublisher

java.lang.Object
dev.langchain4j.service.AiServiceStreamingEventPublisher
All Implemented Interfaces:
Flow.Publisher<AiServiceStreamingEvent>
public class AiServiceStreamingEventPublisher extends Object implements Flow.Publisher<AiServiceStreamingEvent>
A cold, non-blocking reactive stream for an AI Service method that returns a Flow.Publisher of AiServiceStreamingEvents.

It consumes the model's reactive StreamingChatModel.chat(ChatRequest) publisher round by round, mapping each low-level ChatModelStreamingEvent to the corresponding high-level AiServiceStreamingEvent as it arrives (AiServiceStreamingEvent.PartialThinkingEvent, AiServiceStreamingEvent.PartialResponseEvent, AiServiceStreamingEvent.PartialToolCallEvent, AiServiceStreamingEvent.CompleteToolCallEvent). When a round requests tools, each tool is started eagerly as soon as its AiServiceStreamingEvent.CompleteToolCallEvent is emitted - executed without blocking via ToolExecutor.executeAsync(ToolExecutionRequest, InvocationContext) (emitting AiServiceStreamingEvent.BeforeToolExecutionEvent and AiServiceStreamingEvent.AfterToolExecutionEvent as they happen) - and the round's ChatResponse is emitted as an AiServiceStreamingEvent.IntermediateResponseEvent once the round's stream completes, after which the next round's publisher is subscribed to. Exactly one AiServiceStreamingEvent.FinalResponseEvent carrying the final answer is emitted last, followed by onComplete. If RAG content was retrieved, a single AiServiceStreamingEvent.RetrievedContentsEvent is emitted first.

Output guardrails. When the AI Service method has output guardrails configured, the final answer may be rewritten by a guardrail, so the streamed partial chunks are not authoritative. In that case no AiServiceStreamingEvent.PartialResponseEvent is emitted for the final answer; only the (possibly rewritten) AiServiceStreamingEvent.FinalResponseEvent carries it. Consumers must therefore rely on AiServiceStreamingEvent.FinalResponseEvent for the answer whenever output guardrails may be present, rather than concatenating AiServiceStreamingEvent.PartialResponseEvents.

Event ordering. Because tools start eagerly on their AiServiceStreamingEvent.CompleteToolCallEvent (a latency optimization: a tool overlaps the streaming of the round's remaining tokens and of any later tool calls), a round's AiServiceStreamingEvent.BeforeToolExecutionEvent / AiServiceStreamingEvent.AfterToolExecutionEvent may be emitted before that round's AiServiceStreamingEvent.IntermediateResponseEvent. Treat AiServiceStreamingEvent.IntermediateResponseEvent as a per-round marker carrying the assembled model ChatResponse, not as a barrier that all of the round's tool-execution events follow. This is the one ordering difference from the handler-based AiServiceTokenStream/TokenStream, whose intermediate-response callback always precedes that round's tool callbacks. Otherwise the semantics mirror that handler-based stream: events from every round are surfaced in order, and no thread is ever blocked or pinned while a model response or a tool result is in flight.

Back-pressure. The model's streaming response is consumed with unbounded demand and its events are relayed to the subscriber through a bounded buffer (16384 entries by default, overridable per AI Service via AiServices.streamingBufferSize(int)). The model stream could be throttled — its subscription demand maps to TCP / HTTP-2 flow control (see onSubscribe) — but we deliberately request unbounded: throttling the socket cannot slow token generation (already produced and billed server-side) and only risks an idle-timeout reset of the in-flight response, so the sole thing it would protect is heap — which the bounded buffer already guards. Consequently a subscriber that consumes slower than the model produces and overflows the buffer terminates with an IllegalStateException rather than dropping events (which would corrupt the assembled response) or buffering unbounded (which risks OutOfMemoryError). Consumers must therefore not block in onNext; offload heavy per-event work. If a slow-but-correct consumer trips the buffer on a long response, raise it via AiServices.streamingBufferSize(int) (or set it to Integer.MAX_VALUE for an effectively unbounded buffer, accepting the OOM risk).

Cancellation. Cancelling the Flow.Subscription stops the interaction: the in-flight model call is cancelled (for providers whose reactive stream supports it, this aborts the underlying HTTP request), no further round is started, and no more events — including the terminal AiServiceStreamingEvent.FinalResponseEvent and onComplete/onError — are emitted. A tool execution that has already started is not interrupted: it runs to completion and its result is discarded (Java cannot safely interrupt arbitrary tool code; this is a deliberate best-effort contract, consistent with the CompletableFuture path).

Since:
1.20.0

  • Field Details

    • DEFAULT_BUFFER_SIZE

      public static final int DEFAULT_BUFFER_SIZE
      Default size of the bounded back-pressure buffer (see the class-level "Back-pressure" note), used when an AI Service does not override it via AiServices.streamingBufferSize(int).
      See Also:

  • Constructor Details

  • Method Details

    • subscribe

      public void subscribe(Flow.Subscriber<? super AiServiceStreamingEvent> subscriber)
      Specified by:
      subscribe in interface Flow.Publisher<AiServiceStreamingEvent>

    • toTextPublisher

      public static Flow.Publisher<String> toTextPublisher(Flow.Publisher<AiServiceStreamingEvent> events, boolean hasOutputGuardrails, int bufferSize)
      Maps a rich AiServiceStreamingEvent stream to a text-only String stream. Used to satisfy AI Service methods that return a Publisher<String> (or a Flux<String>/Multi<String> via a PublisherAdapter). The concatenation of all emitted strings equals the final answer.

      When there are no output guardrails, the text of each AiServiceStreamingEvent.PartialResponseEvent is emitted as it streams (true token-by-token streaming) and every other event is dropped.

      When there are output guardrails, an output guardrail may rewrite the final answer, so the individual partial chunks are no longer authoritative. In that case only the (possibly rewritten) text of the single AiServiceStreamingEvent.FinalResponseEvent is emitted. This loses no streaming granularity in practice: with output guardrails the rich event stream does not emit AiServiceStreamingEvent.PartialResponseEvents at all (they cannot be reconciled with a possibly-rewritten final answer), so the whole answer would arrive at the very end regardless.

      The event stream is drained with unbounded demand into a bounded buffer rather than mapped with demand-passing operators (such as mutiny-zero's Select/Transform). The last event of a round-trip is always a AiServiceStreamingEvent.FinalResponseEvent, which contributes no string when there are no output guardrails: a demand-passing filter would only replenish upstream demand upon receiving that event, so once the subscriber has taken the last string and stops requesting, the trailing event - and the onComplete behind it - would never be delivered and the stream would hang.

      Parameters:
      events - the rich event stream to adapt
      hasOutputGuardrails - whether the AI Service method has output guardrails configured
      bufferSize - the back-pressure buffer size