Collector

1
from baml_client import b
2
from baml_py import Collector
3
4
# Create a collector with optional name
5
collector = Collector(name="my-collector")
6
7
# Use it with a function call
8
result = b.ExtractResume("...", baml_options={"collector": collector})
9
10
# Access logging information
11
print(collector.last.usage)  # Print usage metrics
12
print(collector.last.raw_llm_response)  # Print final response as string
13
# since there may be retries, print the last http response received
14
print(collector.last.calls[-1].http_response) 

1
from baml_client import b
2
from baml_py import Collector  # Import the Collector class
3
4
def run():
5
    # Create a collector instance with an optional name
6
    collector = Collector(name="my-collector")
7
    # collector will be modified by the function to include all internal state
8
    res = b.ExtractResume("...", baml_options={"collector": collector})
9
    # This will print the return type of the function
10
    print(res)
11
12
    # This is guaranteed to be set by the function
13
    assert collector.last is not None
14
15
    # This will print the id of the last request
16
    print(collector.last.id)
17
18
    # This will print the usage of the last request
19
    # (This aggregates usage from all retries if there was usage emitted)
20
    print(collector.last.usage)
21
22
    # This will print the raw response of the last request
23
    print(collector.last.calls[-1].http_response)
24
25
    # This will print the raw text we used to run the parser.
26
    print(collector.last.raw_llm_response)

1
from baml_client import b
2
from baml_py import Collector
3
4
def run():
5
    collector = Collector(name="reusable-collector")
6
    res = b.ExtractResume("...", baml_options={"collector": collector})
7
   
8
    # Reuse the same collector
9
    res = b.TestOpenAIGPT4oMini("Second call", baml_options={"collector": collector})

1
from baml_client import b
2
from baml_py import Collector
3
4
def run():
5
    # Create separate collectors for different parts of your application
6
    collector_a = Collector(name="collector-a")
7
    collector_b = Collector(name="collector-b")
8
    
9
    # Use both collectors for the same function call
10
    res = b.ExtractResume("...", baml_options={"collector": [collector_a, collector_b]})
11
    
12
    # Both collectors will have the same logs
13
    assert collector_a.last.usage.input_tokens == collector_b.last.usage.input_tokens
14
    
15
    # Use only collector_a for another call
16
    res2 = b.TestOpenAIGPT4oMini("another call", baml_options={"collector": collector_a})
17
    
18
    # collector_a will have 2 logs, collector_b will still have 1
19
    assert len(collector_a.logs) == 2
20
    assert len(collector_b.logs) == 1

1
from baml_client import b
2
from baml_py import Collector
3
4
def run():
5
    collector_a = Collector(name="collector-a")
6
    res = b.ExtractResume("...", baml_options={"collector": collector_a})
7
8
    collector_b = Collector(name="collector-b")
9
    res = b.ExtractResume("...", baml_options={"collector": collector_b})
10
11
    # The total usage of both logs is now available
12
    print(collector_a.usage)
13
    print(collector_b.usage)

1
from baml_client import b
2
from baml_py import Collector
3
4
async def run():
5
    collector = Collector(name="cache-tracker")
6
    
7
    # First call - content will be cached by the provider
8
    res = await b.TestCaching(large_content, "Question 1", baml_options={"collector": collector})
9
    
10
    # Second call with same content - should use cached tokens
11
    res2 = await b.TestCaching(large_content, "Question 2", baml_options={"collector": collector})
12
    
13
    # Access cached token counts
14
    first_log = collector.logs[0]
15
    second_log = collector.logs[1]
16
    
17
    print(f"First call cached tokens: {first_log.usage.cached_input_tokens}")
18
    print(f"Second call cached tokens: {second_log.usage.cached_input_tokens}")
19
    
20
    # Collector aggregates cached tokens across all calls
21
    print(f"Total cached tokens: {collector.usage.cached_input_tokens}")
22
    
23
    # You can also access cached tokens per LLM call (including retries)
24
    print(f"Per-call cached tokens: {first_log.calls[0].usage.cached_input_tokens}")