OnTick

The onTick feature allows you to receive real-time callbacks during BAML function execution, providing access to internal state, streaming responses, and progress updates. This is particularly useful for monitoring function progress, debugging, and accessing intermediate data like “thinking” content from streaming LLM responses.

Quick Start

from baml_client import b
from baml_py import baml_py
def on_tick(reason: str, log: baml_py.FunctionLog):
print(f"Tick received: {reason}")
print(f"Function calls: {len(log.calls) if log else 0}")
# Use with async function
result = await b.TestFunction("Hello world", baml_options={"on_tick": on_tick})

Common Use Cases

Progress Monitoring

Track the progress of long-running BAML function calls:

from baml_client import b
from baml_py import baml_py
def progress_monitor(reason: str, log: baml_py.FunctionLog):
tick_count = getattr(progress_monitor, 'count', 0)
progress_monitor.count = tick_count + 1
print(f"Progress tick #{progress_monitor.count}: {reason}")
if log and log.calls:
latest_call = log.calls[-1]
print(f"Latest call to: {latest_call.client_name}")
result = await b.ExtractResume(
resume_text,
baml_options={"on_tick": progress_monitor}
)

Accessing Streaming “Thinking” Content

Extract intermediate “thinking” content from streaming LLM responses:

import json
from baml_client import b
from baml_py import baml_py
def extract_thinking(reason: str, log: baml_py.FunctionLog):
thinking_content = ""
if log and log.calls:
last_call = log.calls[-1]
# Check if it's a streaming call
if hasattr(last_call, "sse_responses"):
sse_responses = last_call.sse_responses()
if sse_responses:
for response in sse_responses:
try:
data = json.loads(response.text)
if "delta" in data and "thinking" in data["delta"]:
thinking_content += data["delta"]["thinking"]
except (json.JSONDecodeError, AttributeError):
pass
if thinking_content:
print(f"Thinking content: {thinking_content}")
# Use with streaming function
stream = b.stream.TestThinking(
"Write a story about AI",
baml_options={"on_tick": extract_thinking}
)
async for msg in stream:
pass
result = await stream.get_final_response()

Debugging and Logging

Use onTick for comprehensive debugging and logging:

from baml_client import b
from baml_py import baml_py
def debug_logger(reason: str, log: baml_py.FunctionLog):
print(f"=== DEBUG TICK: {reason} ===")
if log:
print(f"Function: {log.function_name}")
print(f"Log type: {log.log_type}")
print(f"Number of calls: {len(log.calls)}")
if log.usage:
print(f"Input tokens: {log.usage.input_tokens}")
print(f"Output tokens: {log.usage.output_tokens}")
if log.calls:
latest_call = log.calls[-1]
print(f"Latest provider: {latest_call.provider}")
print(f"Latest client: {latest_call.client_name}")
if latest_call.usage:
print(f"Call usage - Input: {latest_call.usage.input_tokens}, Output: {latest_call.usage.output_tokens}")
print("=== END DEBUG ===\n")
result = await b.TestFunction("Debug this call", baml_options={"on_tick": debug_logger})

Using with Collectors

OnTick can be used alongside Collectors for comprehensive logging:

from baml_client import b
from baml_py import baml_py, Collector
def on_tick_with_collector(reason: str, log: baml_py.FunctionLog):
print(f"OnTick fired: {reason}")
# Create a collector alongside onTick
collector = Collector("my-collector")
result = await b.TestFunction(
"Hello world",
baml_options={
"on_tick": on_tick_with_collector,
"collector": collector
}
)
# Access data through both mechanisms
print(f"Collector usage: {collector.last.usage}")

Error Handling

OnTick callbacks should handle errors gracefully. If an onTick callback throws an error, the function execution will continue:

from baml_client import b
from baml_py import baml_py
def error_prone_tick(reason: str, log: baml_py.FunctionLog):
# Simulate an error condition
if hasattr(error_prone_tick, 'count'):
error_prone_tick.count += 1
else:
error_prone_tick.count = 1
if error_prone_tick.count == 5:
raise ValueError("Intentional error in onTick")
print(f"Tick #{error_prone_tick.count}: {reason}")
# Function will complete despite callback errors
result = await b.TestFunction("Hello world", baml_options={"on_tick": error_prone_tick})
print("Function completed successfully despite onTick error")

Limitations

Keep these limitations in mind when using onTick:

  1. Synchronous Functions: OnTick is not supported for synchronous function calls. Attempting to use onTick with sync functions will throw an error.

  2. Error Isolation: Errors in onTick callbacks do not stop function execution, but they may not be explicitly surfaced.

API Reference

OnTick Callback Signature

def on_tick(reason: str, log: baml_py.FunctionLog | None) -> None:
"""
OnTick callback function
Args:
reason: The reason for the tick (currently always "Unknown")
log: The current function log with call information
"""
pass

Integration with Function Calls

OnTick is passed via the baml_options parameter (Python) or options object (TypeScript/Go):

# Async function call
result = await b.FunctionName(input, baml_options={"on_tick": callback})
# Streaming function call
stream = b.stream.FunctionName(input, baml_options={"on_tick": callback})

Best Practices

  1. Keep Callbacks Light: OnTick callbacks should be fast and non-blocking
  2. Handle Errors Gracefully: Always include error handling in your callbacks
  3. Use with Collectors: Combine onTick with Collectors for comprehensive logging
  4. Monitor Performance: Test the performance impact for your specific use case
  5. Async Only: Remember that onTick only works with async function calls, not sync calls