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_client.tracing import trace, set_tags
3
from baml_py import Collector
4
5
@trace
6
async def run_with_tags():
7
    # Parent trace tags
8
    set_tags(parent_id="p123", run="xyz")
9
10
    collector = Collector(name="tags-collector")
11
12
    # Per-call tags via baml_options
13
    await b.TestOpenAIGPT4oMini(
14
        "hi",
15
        baml_options={"collector": collector, "tags": {"call_id": "first"}},
16
    )
17
18
    # Retrieve merged tags from the last log
19
    log = collector.last
20
    assert log is not None
21
    print(log.tags)  # {'parent_id': 'p123', 'run': 'xyz', 'call_id': 'first'}

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="stream-collector")
6
7
    # Use streaming
8
    stream = b.stream.ExtractResume("...", baml_options={"collector": collector})
9
    async for chunk in stream:
10
        print(chunk)  # Process streamed chunks
11
12
    # After streaming completes, access the raw SSE responses
13
    log = collector.last
14
    if log and log.selected_call:
15
        sse_responses = log.selected_call.sse_responses()
16
        if sse_responses:
17
            for sse in sse_responses:
18
                # Raw text of the SSE data field
19
                print(f"Raw: {sse.text}")
20
                # Parse as JSON if valid (returns None if not valid JSON)
21
                parsed = sse.json()
22
                if parsed:
23
                    print(f"JSON: {parsed}")

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}")