Streaming

Streaming lets you observe a program's output as it is produced, rather than waiting for the final result. dspy4s streaming is synchronous: streamify turns a program into a function that returns a ClosableIterator[StreamEvent], which you consume with an ordinary loop.

Consuming a stream

Every stream is a sequence of StreamEvents. The type is sealed, so a single match handles every case: tokens as they arrive, status messages, the final prediction, and errors:

def consume(stream: ClosableIterator[StreamEvent]): Option[RawPrediction] =
  var finalPrediction: Option[RawPrediction] = None
  // The close stops the producer thread when a handler throws. After the end of the stream, it has no effect.
  try while stream.hasNext do
      stream.next() match
        case t: TokenEvent      => println(s"Output token of field ${t.fieldName}: ${t.chunk}")
        case s: StatusEvent     => println(s.message)
        case p: PredictionEvent => finalPrediction = Some(p.prediction)
        case e: ErrorEvent      => println(s"Error: ${e.error.message}")
  finally stream.close()
  finalPrediction

The finally closes the iterator, also when a handler throws an exception. Thus the producer thread does not stay parked. After the end of the stream, the close has no effect.

Streaming a program

Wrap a program with Streamify.streamify and declare which output fields to stream with a StreamListener. Calling the result returns the event iterator:

def streamAnswer(question: String)(using RuntimeContext): Option[RawPrediction] =
  val predict       = DynamicPredict(layout = Signature.fromString("question -> answer").layout)
  val streamPredict = Streamify.streamify(
    program = predict,
    streamListeners = Vector(StreamListener("answer"))
  )
  consume(streamPredict(DynamicValues.record("question" := question)))

The listener names the field (answer) you want streamed token by token. The same approach works on a composite program: add a listener per field, and disambiguate predictors that emit the same field name with predictName.

Notes

• The language model must support streaming. The bundled OpenAI provider does.
• Each model of the program streams, also the own model of a predictor.
• Streaming is synchronous, so there is no async iteration to manage; the consumer is a plain while loop over the iterator.
• A streamed call goes through the cache, the retry policy, and the history of ManagedLanguageModel, as a call does. Its token usage goes to the usage trackers and to the result, as the usage of a call does. When the cache has the answer, the stream gives no token: you get the status messages and the final prediction.
• A stream that stops before its end is an error (LmStreamAssemblyError). The retry policy can start a new attempt only before the first token.
• By default, a listener stops after one field. Pass allowReuse = true to get the field from each model call, for example next_thought from each step of ReAct.
• A listener on a Reasoning output gets the reasoning of the model, when the model gives it apart from its answer. Such tokens do not end the field, so the listener gets them from each model call. These tokens do not hide the final prediction, which holds the other outputs.
• Close the iterator if you stop before the end of the stream. The close stops the program and the model stream. If you stop and do not close, the producer thread stays parked on the full queue.

Next: Coming from DSPy.