BAML Basics

Streaming

BAML lets you stream in structured JSON output from LLMs as it comes in.

If you tried streaming in a JSON output from an LLM you’d see something like:

{"items": [{"name": "Appl
{"items": [{"name": "Apple", "quantity": 2, "price": 1.
{"items": [{"name": "Apple", "quantity": 2, "price": 1.50}], "total_cost":
{"items": [{"name": "Apple", "quantity": 2, "price": 1.50}], "total_cost": 3.00} # Completed

BAML gives you fine-grained control of how it fixes this partial JSON and transforms it into a series of semantically valid partial objects.

You can check out more examples (including streaming in FastAPI and NextJS) in the BAML Examples repo.

Let’s stream the output of this function function ExtractReceiptInfo(email: string) -> ReceiptInfo for our example:

class ReceiptItem {
  name string
  description string?
  quantity int
  price float
}
class ReceiptInfo {
    items ReceiptItem[]
    total_cost float?
}
function ExtractReceiptInfo(email: string) -> ReceiptInfo {
  client GPT4o
  prompt #"
    Given the receipt below:
    {{ email }}
    {{ ctx.output_format }}
  "#
}

The BAML code generator creates a set of types in the baml_client library in a module called partial_types in baml_client. These types are modified from your original types to support streaming.

By default, BAML will convert all Class fields into nullable fields, and fill those fields with non-null values as much as possible given the tokens received so far.

BAML will generate b.stream.ExtractReceiptInfo() for you, which you can use like so:

main.py
import asyncio
from baml_client import b, partial_types, types
# Using a stream:
def example1(receipt: str):
    stream = b.stream.ExtractReceiptInfo(receipt)
    # partial is a Partial type with all Optional fields
    for partial in stream:
        print(f"partial: parsed {len(partial.items)} items (object: {partial})")
    # final is the full, original, validated ReceiptInfo type
    final = stream.get_final_response()
    print(f"final: {len(final.items)} items (object: {final})")
# Using only get_final_response() of a stream
#
# In this case, you should just use b.ExtractReceiptInfo(receipt) instead,
# which is slightly faster and more efficient.
def example2(receipt: str):
    final = b.stream.ExtractReceiptInfo(receipt).get_final_response()
    print(f"final: {len(final.items)} items (object: {final})")
# Using the async client:
async def example3(receipt: str):
    # Note the import of the async client
    from baml_client.async_client import b
    stream = b.stream.ExtractReceiptInfo(receipt)
    async for partial in stream:
        print(f"partial: parsed {len(partial.items)} items (object: {partial})")
    final = await stream.get_final_response()
    print(f"final: {len(final.items)} items (object: {final})")
receipt = """
04/14/2024 1:05 pm
Ticket: 220000082489
Register: Shop Counter
Employee: Connor
Customer: Sam
Item	#	Price
Guide leash (1 Pair) uni UNI
1	$34.95
The Index Town Walls
1	$35.00
Boot Punch
3	$60.00
Subtotal	$129.95
Tax ($129.95 @ 9%)	$11.70
Total Tax	$11.70
Total	$141.65
"""
if __name__ == '__main__':
    #uncomment one at a time and run to see the difference
    example1(receipt)
    #example2(receipt)
    #asyncio.run(example3(receipt))

Number fields are always streamed in only when the LLM completes them. E.g. if the final number is 129.95, you’ll only see null or 129.95 instead of partial numbers like 1, 12, 129.9, etc.

Cancelling Streams

You can cancel ongoing streams using abort controllers, which is essential for responsive applications that allow users to stop generation or implement timeouts.

import { b } from './baml_client'
const controller = new AbortController()
const stream = b.stream.ExtractReceiptInfo(receipt, {
  abortController: controller
})
// Process stream with ability to cancel
let itemCount = 0
for await (const partial of stream) {
  itemCount = partial.items?.length || 0
  console.log(`Received ${itemCount} items so far`)
  // Cancel if we have enough items
  if (itemCount >= 5) {
    console.log('Stopping stream - got enough items')
    controller.abort()
    break
  }
}
// Or cancel after a timeout
setTimeout(() => {
  controller.abort()
  console.log('Stream cancelled due to timeout')
}, 5000)

Common Streaming Cancellation Patterns

User-Initiated Cancellation

Allow users to stop streaming generation with a “Stop” button:

function StreamingComponent() {
  const [controller, setController] = useState<AbortController | null>(null)
  const [isStreaming, setIsStreaming] = useState(false)
  const [result, setResult] = useState("")
  const startStreaming = async () => {
    const newController = new AbortController()
    setController(newController)
    setIsStreaming(true)
    try {
      const stream = b.stream.GenerateContent(prompt, {
        abortController: newController
      })
      let accumulated = ""
      for await (const partial of stream) {
        accumulated = partial.content || ""
        setResult(accumulated)
      }
    } catch (error) {
      if (error.name === 'BamlAbortError') {
        console.log('Stream cancelled by user')
      }
    } finally {
      setIsStreaming(false)
      setController(null)
    }
  }
  const stopStreaming = () => {
    controller?.abort()
  }
  return (
    <div>
      <button onClick={startStreaming} disabled={isStreaming}>
        Start Streaming
      </button>
      <button onClick={stopStreaming} disabled={!isStreaming}>
        Stop
      </button>
      <div>{result}</div>
    </div>
  )
}

For more examples and patterns, see the Abort Controllers guide.

Semantic Streaming

BAML provides powerful attributes to control how your data streams, ensuring that partial values always maintain semantic validity. Here are the three key streaming attributes:

@stream.done

This attribute ensures a type or field is only streamed when it’s completely finished. It’s useful when you need atomic, fully-formed values.

For example:

class ReceiptItem {
  name string
  quantity int
  price float
  // The entire ReceiptItem will only stream when complete
  @@stream.done
}
// Receipts is a list of ReceiptItems,
// each internal item will only stream when complete
type Receipts = ReceiptItem[]
class Person {
  // Name will only appear when fully complete,
  // until then it will be null
  name string @stream.done
  // Numbers (floats and ints) will only appear
  // when fully complete by default
  age int
  // Bio will stream token by token
  bio string
}

Atomic list items with union types

A common pattern is streaming a list of items where each item can be one of several types (e.g. tool calls and messages). You can use @stream.done on the list element type to ensure each item only appears once it’s fully complete:

class ToolCall {
  name string
  parameters string
}
class Message {
  role string
  content string
}
type OutputItem = ToolCall | Message
// Each list element appears only when fully complete.
// The list grows incrementally as items finish.
function Run(input: string) -> (OutputItem @stream.done)[] {
  client MyClient
  prompt #"
    {{ input }}
    {{ ctx.output_format }}
  "#
}

When @stream.done is applied to a union type, it propagates to all variants. This means you don’t need to add @@stream.done to each class individually — annotating the union is sufficient.

You can also achieve the same behavior by adding @@stream.done to each class in the union. The (T @stream.done)[] syntax is more concise when the classes are used in other contexts where you don’t want @@stream.done.

@stream.not_null

This attribute ensures a containing object is only streamed when this field has a value. It’s particularly useful for discriminator fields or required metadata.

For example:

class Message {
  // Message won't stream until type is known
  type "error" | "success" | "info" @stream.not_null
  // Timestamp will only appear when fully complete
  // until then it will be null
  timestamp string @stream.done
  // Content can stream token by token
  content string
}

@stream.with_state

This attribute adds metadata to track if a field has finished streaming. It’s perfect for showing loading states in UIs.

For example:

class BlogPost {
  // The blog post will only stream when title is known
  title string @stream.done @stream.not_null
  // The content will stream token by token, and include completion state
  content string @stream.with_state
}

This will generate the following code in the partial_types module:

class StreamState(BaseModel, Generic[T]):
  value: T,
  state: "incomplete" | "complete"
class BlogPost(BaseModel):
  title: str
  content: StreamState[str | None]

Type Transformation Summary

Here’s how these attributes affect your types in generated code:

BAML TypeGenerated Type (during streaming)Description
TPartial[T]?Default: Nullable and partial
T @stream.doneT?Nullable but always complete when present
T @stream.not_nullPartial[T]Always present but may be partial
T @stream.done @stream.not_nullTAlways present and always complete
T @stream.with_stateStreamState[Partial[T]?]Includes streaming state metadata

The return type of a function is not affected by streaming attributes!

Putting it all together

Let’s put all of these concepts together to design an application that streams a conversation containing stock recommendations, using semantic streaming to ensure that the streamed data obeys our domain’s invariants.

enum Stock {
  APPL
  MSFT
  GOOG
  BAML
}
// Make recommendations atomic - we do not want a recommendation to be
// modified by streaming additional messages.
class Recommendation {
  stock Stock
  amount float
  action "buy" | "sell"
  @@stream.done
}
class AssistantMessage {
  message_type "greeting" | "conversation" | "farewell" @stream.not_null
  message string @stream.with_state @stream.not_null
}
function Respond(
  history: (UserMessage | AssistantMessage | Recommendation)[]
) -> Message | Recommendation {
  client DeepseekR1
  prompt #"
    Make the message in the conversation, using a conversational
    message or a stock recommendation, based on this conversation history:
    {{ history }}.
    {{ ctx.output_format }}
  "#
}

The above BAML code will generate the following Python definitions in the partial_types module. The use of streaming attributes has several effects on the generated code:

  • Recommendation does not have any partial fields because it was marked @stream.done.
  • The Message.message string is wrapped in StreamState, allowing runtime checking of its completion status. This status could be used to render a spinner as the message streams in.
  • The Message.message_type field may not be null, because it was marked as @stream.not_null.
class StreamState(BaseModel, Generic[T]):
  value: T,
  state: Literal["Pending", "Incomplete", "Complete"]
class Stock(str, Enum):
    APPL = "APPL"
    MSFT = "MSFT"
    GOOG = "GOOG"
    BAML = "BAML"
class Recommendation(BaseClass):
    stock: Stock
    amount: float
    action: Literal["buy", "sell"]
class Message(BaseClass):
  message_type: Literal["gretting","conversation","farewell"]
  message: StreamState[string]