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.

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

1
class ReceiptItem {
2
  name string
3
  description string?
4
  quantity int
5
  price float
6
}
7
8
class ReceiptInfo {
9
    items ReceiptItem[]
10
    total_cost float?
11
}
12
13
function ExtractReceiptInfo(email: string) -> ReceiptInfo {
14
  client GPT4o
15
  prompt #"
16
    Given the receipt below:
17
18
    {{ email }}
19
20
    {{ ctx.output_format }}
21
  "#
22
}

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.

1
import asyncio
2
from baml_client import b, partial_types, types
3
4
# Using a stream:
5
def example1(receipt: str):
6
    stream = b.stream.ExtractReceiptInfo(receipt)
7
8
    # partial is a Partial type with all Optional fields
9
    for partial in stream:
10
        print(f"partial: parsed {len(partial.items)} items (object: {partial})")
11
12
    # final is the full, original, validated ReceiptInfo type
13
    final = stream.get_final_response()
14
    print(f"final: {len(final.items)} items (object: {final})")
15
16
# Using only get_final_response() of a stream
17
#
18
# In this case, you should just use b.ExtractReceiptInfo(receipt) instead,
19
# which is slightly faster and more efficient.
20
def example2(receipt: str):
21
    final = b.stream.ExtractReceiptInfo(receipt).get_final_response()
22
    print(f"final: {len(final.items)} items (object: {final})")
23
24
# Using the async client:
25
async def example3(receipt: str):
26
    # Note the import of the async client
27
    from baml_client.async_client import b
28
    stream = b.stream.ExtractReceiptInfo(receipt)
29
    async for partial in stream:
30
        print(f"partial: parsed {len(partial.items)} items (object: {partial})")
31
32
    final = await stream.get_final_response()
33
    print(f"final: {len(final.items)} items (object: {final})")
34
35
receipt = """
36
04/14/2024 1:05 pm
37
38
Ticket: 220000082489
39
Register: Shop Counter
40
Employee: Connor
41
Customer: Sam
42
Item	#	Price
43
Guide leash (1 Pair) uni UNI
44
1	$34.95
45
The Index Town Walls
46
1	$35.00
47
Boot Punch
48
3	$60.00
49
Subtotal	$129.95
50
Tax ($129.95 @ 9%)	$11.70
51
Total Tax	$11.70
52
Total	$141.65
53
"""
54
55
if __name__ == '__main__':
56
    #uncomment one at a time and run to see the difference
57
    example1(receipt)
58
    #example2(receipt)
59
    #asyncio.run(example3(receipt))

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.

1
import { b } from './baml_client'
2
3
const controller = new AbortController()
4
5
const stream = b.stream.ExtractReceiptInfo(receipt, {
6
  abortController: controller
7
})
8
9
// Process stream with ability to cancel
10
let itemCount = 0
11
for await (const partial of stream) {
12
  itemCount = partial.items?.length || 0
13
  console.log(`Received ${itemCount} items so far`)
14
15
  // Cancel if we have enough items
16
  if (itemCount >= 5) {
17
    console.log('Stopping stream - got enough items')
18
    controller.abort()
19
    break
20
  }
21
}
22
23
// Or cancel after a timeout
24
setTimeout(() => {
25
  controller.abort()
26
  console.log('Stream cancelled due to timeout')
27
}, 5000)

Common Streaming Cancellation Patterns

User-Initiated Cancellation

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

1
function StreamingComponent() {
2
  const [controller, setController] = useState<AbortController | null>(null)
3
  const [isStreaming, setIsStreaming] = useState(false)
4
  const [result, setResult] = useState("")
5
6
  const startStreaming = async () => {
7
    const newController = new AbortController()
8
    setController(newController)
9
    setIsStreaming(true)
10
11
    try {
12
      const stream = b.stream.GenerateContent(prompt, {
13
        abortController: newController
14
      })
15
16
      let accumulated = ""
17
      for await (const partial of stream) {
18
        accumulated = partial.content || ""
19
        setResult(accumulated)
20
      }
21
    } catch (error) {
22
      if (error.name === 'BamlAbortError') {
23
        console.log('Stream cancelled by user')
24
      }
25
    } finally {
26
      setIsStreaming(false)
27
      setController(null)
28
    }
29
  }
30
31
  const stopStreaming = () => {
32
    controller?.abort()
33
  }
34
35
  return (
36
    <div>
37
      <button onClick={startStreaming} disabled={isStreaming}>
38
        Start Streaming
39
      </button>
40
      <button onClick={stopStreaming} disabled={!isStreaming}>
41
        Stop
42
      </button>
43
      <div>{result}</div>
44
    </div>
45
  )
46
}

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:

1
class ReceiptItem {
2
  name string
3
  quantity int
4
  price float
5
6
  // The entire ReceiptItem will only stream when complete
7
  @@stream.done
8
}
9
10
// Receipts is a list of ReceiptItems,
11
// each internal item will only stream when complete
12
type Receipts = ReceiptItem[]
13
14
class Person {
15
  // Name will only appear when fully complete,
16
  // until then it will be null
17
  name string @stream.done
18
  // Numbers (floats and ints) will only appear
19
  // when fully complete by default
20
  age int
21
  // Bio will stream token by token
22
  bio string
23
}

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:

1
class ToolCall {
2
  name string
3
  parameters string
4
}
5
6
class Message {
7
  role string
8
  content string
9
}
10
11
type OutputItem = ToolCall | Message
12
13
// Each list element appears only when fully complete.
14
// The list grows incrementally as items finish.
15
function Run(input: string) -> (OutputItem @stream.done)[] {
16
  client MyClient
17
  prompt #"
18
    {{ input }}
19
    {{ ctx.output_format }}
20
  "#
21
}

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.

@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:

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

@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:

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

This will generate the following code in the partial_types module:

1
class StreamState(BaseModel, Generic[T]):
2
  value: T,
3
  state: "incomplete" | "complete"
4
5
class BlogPost(BaseModel):
6
  title: str
7
  content: StreamState[str | None]

Type Transformation Summary

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

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.

1
enum Stock {
2
  APPL
3
  MSFT
4
  GOOG
5
  BAML
6
}
7
8
// Make recommendations atomic - we do not want a recommendation to be
9
// modified by streaming additional messages.
10
class Recommendation {
11
  stock Stock
12
  amount float
13
  action "buy" | "sell"
14
  @@stream.done
15
}
16
17
class AssistantMessage {
18
  message_type "greeting" | "conversation" | "farewell" @stream.not_null
19
  message string @stream.with_state @stream.not_null
20
}
21
22
function Respond(
23
  history: (UserMessage | AssistantMessage | Recommendation)[]
24
) -> Message | Recommendation {
25
  client DeepseekR1
26
  prompt #"
27
    Make the message in the conversation, using a conversational
28
    message or a stock recommendation, based on this conversation history:
29
    {{ history }}.
30
31
    {{ ctx.output_format }}
32
  "#
33
}

1
class StreamState(BaseModel, Generic[T]):
2
  value: T,
3
  state: Literal["Pending", "Incomplete", "Complete"]
4
5
class Stock(str, Enum):
6
    APPL = "APPL"
7
    MSFT = "MSFT"
8
    GOOG = "GOOG"
9
    BAML = "BAML"
10
11
class Recommendation(BaseClass):
12
    stock: Stock
13
    amount: float
14
    action: Literal["buy", "sell"]
15
16
class Message(BaseClass):
17
  message_type: Literal["gretting","conversation","farewell"]
18
  message: StreamState[string]