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("dev.tramai:tramai-standalone:0.3.1")
   implementation("dev.tramai:tramai-anthropic:0.3.1") // 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 contains inputTokens, outputTokens when provider reports them.

   data class Error(val cause: TramaiException) : StreamChunk()
   // Terminal event: stream failed during execution. No further chunks.
}

Standalone Example

data class UsageMetrics(
   val inputTokens: Int? = null,
   val outputTokens: Int? = null,
   val thinkingTokens: Int? = null,
)

@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 Flow is cancelled, TramAI propagates cancellation to the underlying provider stream.
Patterns that cancel: take(N), first(), coroutine scope cancellation, or manual cancel().
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 uses the same Flow<StreamChunk> API. Since Java doesn't have Kotlin's Flow.collect natively, use a Reactor adapter or collect via runBlocking:

@AiService
public interface SummaryService {
   @Operation(prompt = "Summarize the article", model = "gpt-4o")
   Flow<StreamChunk> summarize(String text);
}

// In a Kotlin helper or using reactor-kotlin:
runBlocking(() -> {
   summaryService.summarize(longArticle).collect(chunk -> {
       if (chunk instanceof StreamChunk.Token token) {
           System.out.print(token.getText());
       } else if (chunk instanceof StreamChunk.Complete complete) {
           System.out.println("\nDone. Tokens: " + complete.getUsage().getOutputTokens());
       } else if (chunk instanceof StreamChunk.Error error) {
           System.err.println("Error: " + error.getCause().getMessage());
       }
   });
});

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