Streaming
What it is: TramAI supports raw-text streaming through Kotlin Flow<StreamChunk>. Each chunk is either an incremental text token, a successful terminal event, or a terminal error.
When to use it: For low-latency text delivery — displaying model output as it's generated, powering chat UIs, or processing incremental results. Not for structured output (use the standard request/response path instead).
Minimum Setup
Add the dependency:
dependencies {
implementation(platform("dev.tramai:tramai-bom:0.6.0"))
implementation("dev.tramai:tramai-standalone")
implementation("dev.tramai:tramai-anthropic") // or your provider
}
Define a streaming service method returning Flow<StreamChunk>:
@AiService
interface StreamingService {
@Operation(
prompt = "Stream a detailed response to the user's question",
model = "claude-sonnet-4-20250514",
)
fun stream(question: String): Flow<StreamChunk>
}
Wire it up:
val tramai = Tramai {
provider(
AnthropicProvider(System.getenv("ANTHROPIC_API_KEY")),
name = "anthropic",
default = true,
)
model("claude-sonnet-4-20250514", "anthropic")
}
val service = tramai.create<StreamingService>()
Consume the stream:
runBlocking {
service.stream("Explain quantum computing").collect { chunk ->
when (chunk) {
is StreamChunk.Token -> print(chunk.text)
is StreamChunk.Complete -> println("\n--- done ---")
is StreamChunk.Error -> println("\nerror: ${chunk.cause.message}")
}
}
}
StreamChunk Types
sealed class StreamChunk {
data class Token(val text: String) : StreamChunk()
// One incremental text fragment — may be a partial word
data class Complete(
val fullText: String,
val usage: UsageMetrics = UsageMetrics(),
) : StreamChunk()
// Terminal event: stream completed successfully.
// usage carries inputTokens, outputTokens, thinkingTokens, imageCount and
// imageTokensEstimate when the provider reports them.
data class Error(val cause: TramaiException) : StreamChunk()
// Terminal event: stream failed during execution. No further chunks.
}
Standalone Example
import dev.tramai.core.model.StreamChunk
import dev.tramai.core.model.UsageMetrics
@AiService
interface ChatService {
@Operation(
prompt = "You are a helpful assistant.",
model = "gpt-4o",
)
fun chat(message: String): Flow<StreamChunk>
}
val tramai = Tramai {
provider(OpenAiProvider(System.getenv("OPENAI_API_KEY")), name = "openai", default = true)
model("gpt-4o", "openai")
}
val service = tramai.create<ChatService>()
runBlocking {
service.chat("Tell me a story").collect { chunk ->
when (chunk) {
is StreamChunk.Token -> print(chunk.text)
is StreamChunk.Complete -> {
println("\n[Tokens: ${chunk.usage.inputTokens} in / ${chunk.usage.outputTokens} out]")
}
is StreamChunk.Error -> System.err.println("\nFailed: ${chunk.cause.message}")
}
}
}
Cancellation Semantics
- Consumer-driven cancellation: If the coroutine collecting the
Flowis cancelled, TramAI propagates cancellation to the underlying provider stream. - Patterns that cancel:
take(N),first(), coroutine scope cancellation, or manualcancel(). - Provider support: When the provider implementation supports stream cancellation, provider work stops promptly. Otherwise the engine discards further chunks.
// Take only the first 5 tokens (cancels the provider stream after 5)
service.chat("Explain FPGAs").take(5).collect { chunk ->
if (chunk is StreamChunk.Token) print(chunk.text)
}
Terminal Error Handling
| Scenario | Behavior |
|---|---|
| Provider unavailable before first token | If a fallback route exists, engine may retry or fail over before the first emitted token |
| Provider fails before first token, no fallback | First emitted chunk is StreamChunk.Error |
| Provider fails after first token emitted | Engine emits StreamChunk.Error — no mid-stream failover |
| Usage exceeds token budget | Engine emits StreamChunk.Error after already-emitted tokens |
| Provider does not support streaming | Engine fails with ProviderCapabilityException at service creation time |
Important: Once a stream has emitted a visible StreamChunk.Token, TramAI does NOT attempt to splice together partial output from another provider. Cross-provider stream stitching would invent correctness guarantees the API cannot support. Instead it terminates with StreamChunk.Error.
Kotlin + Java Examples
Kotlin
@AiService
interface SummaryService {
@Operation(prompt = "Summarize the article", model = "gpt-4o")
fun summarize(text: String): Flow<StreamChunk>
}
runBlocking {
val allText = StringBuilder()
summaryService.summarize(longArticle).collect { chunk ->
when (chunk) {
is StreamChunk.Token -> allText.append(chunk.text)
is StreamChunk.Complete ->
println("Full summary (${allText.length} chars, ${chunk.usage.outputTokens} tokens)")
is StreamChunk.Error -> throw chunk.cause
}
}
}
Java
Java declares the same Flow<StreamChunk> return type, but Java has no first-class Flow collector:
@AiService
public interface SummaryService {
@Operation(prompt = "Summarize the article", model = "gpt-4o")
Flow<StreamChunk> summarize(String text);
}
Consume the stream from a Kotlin boundary rather than from Java. In Spring the simplest form is a Kotlin controller method that returns the flow directly:
@PostMapping("/summary/stream", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
fun streamSummary(@RequestBody request: SummaryInput) = facade.streamSummary(request.text)
Adding a Reactor bridge your application owns (kotlinx-coroutines-reactor) is the other option. A Java-native streaming surface is not part of the 0.6.0 contract.
Guarantees
| Guarantee | Detail |
|---|---|
| Startup failover | May retry/fallback before first token if a fallback route is configured |
| No mid-stream failover | Once a token is emitted, failure is terminal |
| Cancellation propagation | Consumer cancellation stops provider work |
| Terminal usage | Arrives on StreamChunk.Complete when the provider reports it |
| Thread safety | Flow is cold — each collector gets its own stream |
Limitations
- Raw text only: Streaming structured partials are not supported. For typed structured output, use the standard request/response path.
- No tool calling during streaming: Not part of the current public contract.
- Java-specific streaming wrappers: Not yet a first-class surface. Java users should use Kotlin coroutine bridges or Reactor adapters.
- No provider stitching: Mid-stream failures are terminal. The engine does not splice partial output from alternate providers.
Next Steps
- Structured Output — For typed results, use the non-streaming path
- Tool Calling — Combine tool execution with standard (non-streaming) operations
- Observability — Monitor streaming latency and error rates
