Streaming and Async

Why Stream?

A normal agent.run(...) blocks until the whole answer is ready. For long answers that feels slow. Streaming sends the answer in small pieces (tokens) so you can print text as it arrives.

Koog streams output as a Kotlin Flow<StreamFrame>. Because it is a Flow, you get backpressure, cancellation, and composition for free.

Diagram: Streaming sends many small TextDelta frames, then a final End frame.

Loading diagram...

StreamFrame Types

Koog defines a sealed StreamFrame. The important ones:

• StreamFrame.TextDelta(text) — one piece of assistant text (show this live)
• StreamFrame.TextComplete(text) — the full text, after all deltas
• StreamFrame.ToolCallComplete(name, content) — a finished tool call
• StreamFrame.End(finishReason) — the stream is over

Listen to Streaming Frames in an Agent

Add the EventHandler feature and react to each frame. This is the simplest way to stream in an agent:

val agent = AIAgent(
    promptExecutor = simpleOpenAIExecutor(apiKey),
    llmModel = OpenAIModels.Chat.GPT4o
) {
    handleEvents {
        onLLMStreamingFrameReceived { ctx ->
            when (val frame = ctx.streamFrame) {
                is StreamFrame.TextDelta -> print(frame.text)
                is StreamFrame.ToolCallComplete ->
                    println("\n[tool call: ${frame.name}]")
                is StreamFrame.End -> println("\n[stream done]")
                else -> {}
            }
        }
        onLLMStreamingFailed { ctx ->
            println("Stream error: ${ctx.error}")
        }
    }
}

Expected behavior as the answer is produced:

Hel|lo!| How| can| I| help?
[stream done]

The | marks where each TextDelta arrives. The user sees text appear word by word.

Derive Plain Text

If you only want text, Koog gives helpers on the stream:

frames.filterTextOnly().collect { chunk -> print(chunk) }   // live
val full = frames.collectText()                              // after End

Concurrent Tool Calls

Tools run inside coroutines, so independent calls can run at the same time. Use async and awaitAll:

import kotlinx.coroutines.*

suspend fun fetchAll(cities: List<String>): List<String> = coroutineScope {
    cities.map { city ->
        async { agent.run("What is the weather in $city?") }
    }.awaitAll()
}

Three cities queried in parallel take about as long as one.

Cancellation and Timeout

Coroutines give you structured cancellation. If the user stops waiting, you cancel the work. Use withTimeout to set a hard limit:

import kotlinx.coroutines.*

suspend fun askWithTimeout(question: String): String = runCatching {
    withTimeout(15.seconds) {
        agent.run(question)
    }
}.getOrElse { "Timed out or cancelled." }

When the timeout fires, Koog's Flow is cancelled and the in-flight LLM request is stopped. This prevents one slow call from hanging your app.

Rules of Thumb

• Stream for any answer shown to a user (chat UI)
• Use withTimeout for every external LLM call
• Run independent tools in parallel with async/awaitAll
• Keep one CoroutineScope per request so cancellation cleans up everything

Practice

1. Run an agent in streaming mode and print each TextDelta without a newline.
2. Wrap agent.run(...) in withTimeout(10.seconds) and trigger the timeout with a very long prompt.
3. Query the weather for five cities in parallel and print the total elapsed time versus doing them one by one.