OnTick

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

1
from baml_client import b
2
from baml_py import baml_py
3
4
def progress_monitor(reason: str, log: baml_py.FunctionLog):
5
    tick_count = getattr(progress_monitor, 'count', 0)
6
    progress_monitor.count = tick_count + 1
7
    
8
    print(f"Progress tick #{progress_monitor.count}: {reason}")
9
    
10
    if log and log.calls:
11
        latest_call = log.calls[-1]
12
        print(f"Latest call to: {latest_call.client_name}")
13
14
result = await b.ExtractResume(
15
    resume_text, 
16
    baml_options={"on_tick": progress_monitor}
17
)

1
import json
2
from baml_client import b
3
from baml_py import baml_py
4
5
def extract_thinking(reason: str, log: baml_py.FunctionLog):
6
    thinking_content = ""
7
    
8
    if log and log.calls:
9
        last_call = log.calls[-1]
10
        
11
        # Check if it's a streaming call
12
        if hasattr(last_call, "sse_responses"):
13
            sse_responses = last_call.sse_responses()
14
            if sse_responses:
15
                for response in sse_responses:
16
                    try:
17
                        data = json.loads(response.text)
18
                        if "delta" in data and "thinking" in data["delta"]:
19
                            thinking_content += data["delta"]["thinking"]
20
                    except (json.JSONDecodeError, AttributeError):
21
                        pass
22
    
23
    if thinking_content:
24
        print(f"Thinking content: {thinking_content}")
25
26
# Use with streaming function
27
stream = b.stream.TestThinking(
28
    "Write a story about AI", 
29
    baml_options={"on_tick": extract_thinking}
30
)
31
32
async for msg in stream:
33
    pass
34
35
result = await stream.get_final_response()

1
from baml_client import b
2
from baml_py import baml_py
3
4
def debug_logger(reason: str, log: baml_py.FunctionLog):
5
    print(f"=== DEBUG TICK: {reason} ===")
6
    
7
    if log:
8
        print(f"Function: {log.function_name}")
9
        print(f"Log type: {log.log_type}")
10
        print(f"Number of calls: {len(log.calls)}")
11
        
12
        if log.usage:
13
            print(f"Input tokens: {log.usage.input_tokens}")
14
            print(f"Output tokens: {log.usage.output_tokens}")
15
        
16
        if log.calls:
17
            latest_call = log.calls[-1]
18
            print(f"Latest provider: {latest_call.provider}")
19
            print(f"Latest client: {latest_call.client_name}")
20
            
21
            if latest_call.usage:
22
                print(f"Call usage - Input: {latest_call.usage.input_tokens}, Output: {latest_call.usage.output_tokens}")
23
    
24
    print("=== END DEBUG ===\n")
25
26
result = await b.TestFunction("Debug this call", baml_options={"on_tick": debug_logger})

1
from baml_client import b
2
from baml_py import baml_py, Collector
3
4
def on_tick_with_collector(reason: str, log: baml_py.FunctionLog):
5
    print(f"OnTick fired: {reason}")
6
7
# Create a collector alongside onTick
8
collector = Collector("my-collector")
9
10
result = await b.TestFunction(
11
    "Hello world", 
12
    baml_options={
13
        "on_tick": on_tick_with_collector,
14
        "collector": collector
15
    }
16
)
17
18
# Access data through both mechanisms
19
print(f"Collector usage: {collector.last.usage}")

1
from baml_client import b
2
from baml_py import baml_py
3
4
def error_prone_tick(reason: str, log: baml_py.FunctionLog):
5
    # Simulate an error condition
6
    if hasattr(error_prone_tick, 'count'):
7
        error_prone_tick.count += 1
8
    else:
9
        error_prone_tick.count = 1
10
    
11
    if error_prone_tick.count == 5:
12
        raise ValueError("Intentional error in onTick")
13
    
14
    print(f"Tick #{error_prone_tick.count}: {reason}")
15
16
# Function will complete despite callback errors
17
result = await b.TestFunction("Hello world", baml_options={"on_tick": error_prone_tick})
18
print("Function completed successfully despite onTick error")

1
def on_tick(reason: str, log: baml_py.FunctionLog | None) -> None:
2
    """
3
    OnTick callback function
4
    
5
    Args:
6
        reason: The reason for the tick (currently always "Unknown")
7
        log: The current function log with call information
8
    """
9
    pass

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