style: fix ruff format issues

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Matt Perpick

py/bench_compare.py

@@ -0,0 +1,102 @@
+"""Side-by-side benchmark. Run on main, then on the perf branch."""
+import os, sys, time
+os.environ["BRAINTRUST_DISABLE_ATEXIT_FLUSH"] = "true"
+sys.path.insert(0, "src")
+
+from braintrust.logger import (
+    BraintrustState, SpanImpl, _MemoryBackgroundLogger, SpanObjectTypeV3,
+    stringify_with_overflow_meta,
+)
+from braintrust.merge_row_batch import merge_row_batch
+from braintrust.bt_json import bt_safe_deep_copy
+from braintrust.util import LazyValue
+
+PID = LazyValue(lambda: "test", use_mutex=False)
+PID.get()
+
+MED = {
+    "input": {"messages": [{"role": "user", "content": "What is 2+2?"},
+                            {"role": "assistant", "content": "4"}]},
+    "output": {"result": "The answer is 4", "confidence": 0.95},
+    "scores": {"accuracy": 0.9, "relevance": 0.85},
+    "metadata": {"model": "gpt-4", "temperature": 0.7, "max_tokens": 100},
+}
+LARGE = {
+    "input": {"messages": [{"role": "user", "content": f"msg {i}" * 20} for i in range(10)]},
+    "output": {"result": "x" * 5000,
+               "details": {f"k{i}": f"v{i}" * 10 for i in range(50)}},
+    "scores": {f"s{i}": i / 100.0 for i in range(20)},
+    "metadata": {f"m{i}": f"v{i}" * 5 for i in range(30)},
+}
+
+def fresh():
+    s = BraintrustState()
+    ml = _MemoryBackgroundLogger()
+    s._override_bg_logger.logger = ml
+    return s, ml
+
+def bench(label, fn, N):
+    # warmup
+    fn()
+    t0 = time.perf_counter()
+    for _ in range(N):
+        fn()
+    us = (time.perf_counter() - t0) / N * 1e6
+    print(f"  {label:40s} {us:8.1f} us/op")
+    return us
+
+N = 5000
+print(f"N={N}\n")
+
+# -- user thread --
+print("User thread:")
+
+s, ml = fresh()
+bench("start_span (medium)", lambda: SpanImpl(
+    parent_object_type=SpanObjectTypeV3.PROJECT_LOGS, parent_object_id=PID,
+    parent_compute_object_metadata_args=None, parent_span_ids=None,
+    name="b", state=s, event=dict(MED), lookup_span_parent=False), N)
+
+s, ml = fresh()
+bench("start_span (large)", lambda: SpanImpl(
+    parent_object_type=SpanObjectTypeV3.PROJECT_LOGS, parent_object_id=PID,
+    parent_compute_object_metadata_args=None, parent_span_ids=None,
+    name="b", state=s, event=dict(LARGE), lookup_span_parent=False), N)
+
+s, ml = fresh()
+def _tree():
+    root = SpanImpl(parent_object_type=SpanObjectTypeV3.PROJECT_LOGS, parent_object_id=PID,
+        parent_compute_object_metadata_args=None, parent_span_ids=None,
+        name="root", state=s, event=dict(MED), lookup_span_parent=False)
+    c = root.start_span(name="child", input="x", output="y")
+    c.end(); root.end()
+bench("root + child + end (medium)", _tree, N)
+
+s, ml = fresh()
+span = SpanImpl(parent_object_type=SpanObjectTypeV3.PROJECT_LOGS, parent_object_id=PID,
+    parent_compute_object_metadata_args=None, parent_span_ids=None,
+    name="b", state=s, lookup_span_parent=False)
+bench("span.log (medium)", lambda: span.log(**MED), N)
+
+# -- deep copy --
+print("\nDeep copy:")
+bench("bt_safe_deep_copy (medium)", lambda: bt_safe_deep_copy(MED), N)
+bench("bt_safe_deep_copy (large)", lambda: bt_safe_deep_copy(LARGE), N)
+
+# -- flush thread --
+print("\nFlush thread:")
+for count in (1000, 5000):
+    s2, ml2 = fresh()
+    for _ in range(count):
+        SpanImpl(parent_object_type=SpanObjectTypeV3.PROJECT_LOGS, parent_object_id=PID,
+            parent_compute_object_metadata_args=None, parent_span_ids=None,
+            name="b", state=s2, event=dict(MED), lookup_span_parent=False)
+    items = ml2.logs[:]
+    def _flush(items=items):
+        unwrapped = [it.get() for it in items]
+        merged = merge_row_batch(unwrapped)
+        _ = [stringify_with_overflow_meta(m) for m in merged]
+    t0 = time.perf_counter()
+    _flush()
+    elapsed = time.perf_counter() - t0
+    print(f"  flush {count} items (medium)             {elapsed/count*1e6:8.1f} us/item")

py/bench_e2e.py

@@ -0,0 +1,104 @@
+"""End-to-end CPU time benchmark simulating real usage patterns.
+
+Measures total CPU cost of a realistic workload:
+- Init a logger
+- Create spans with nested children
+- Log inputs/outputs/scores
+- Flush everything
+
+Run: cd py && python bench_e2e.py
+"""
+import os, sys, time
+os.environ["BRAINTRUST_DISABLE_ATEXIT_FLUSH"] = "true"
+sys.path.insert(0, "src")
+
+from braintrust.logger import (
+    BraintrustState, SpanImpl, _MemoryBackgroundLogger, SpanObjectTypeV3,
+    stringify_with_overflow_meta,
+)
+from braintrust.merge_row_batch import merge_row_batch
+from braintrust.util import LazyValue
+
+
+def run_workload(num_requests):
+    """Simulate num_requests LLM calls, each with a root span + child span + logging."""
+    state = BraintrustState()
+    ml = _MemoryBackgroundLogger()
+    state._override_bg_logger.logger = ml
+    pid = LazyValue(lambda: "proj-abc123", use_mutex=False)
+    pid.get()
+
+    # Phase 1: create spans and log (user thread work)
+    t_start = time.perf_counter()
+
+    for i in range(num_requests):
+        # Root span: the overall request
+        root = SpanImpl(
+            parent_object_type=SpanObjectTypeV3.PROJECT_LOGS,
+            parent_object_id=pid,
+            parent_compute_object_metadata_args=None,
+            parent_span_ids=None,
+            name="handle_request",
+            state=state,
+            event={
+                "input": {
+                    "messages": [
+                        {"role": "system", "content": "You are a helpful assistant."},
+                        {"role": "user", "content": f"Question {i}: What is {i} + {i}?"},
+                    ]
+                },
+                "metadata": {"user_id": f"user_{i % 100}", "session_id": "sess_abc"},
+            },
+            lookup_span_parent=False,
+        )
+
+        # Child span: LLM call
+        child = root.start_span(
+            name="llm_call",
+            input={"model": "gpt-4", "temperature": 0.7, "max_tokens": 500},
+        )
+
+        # Log the LLM response
+        child.log(
+            output={
+                "choices": [{"message": {"role": "assistant", "content": f"The answer is {i*2}."}}],
+                "usage": {"prompt_tokens": 50, "completion_tokens": 20, "total_tokens": 70},
+            },
+            metrics={"latency": 0.234, "tokens_per_second": 85.5},
+        )
+        child.end()
+
+        # Log scores on root
+        root.log(
+            output=f"The answer is {i*2}.",
+            scores={"accuracy": 0.95, "relevance": 0.88},
+        )
+        root.end()
+
+    t_user = time.perf_counter() - t_start
+
+    # Phase 2: flush (background thread work)
+    t_flush_start = time.perf_counter()
+    items = ml.logs[:]
+    unwrapped = [it.get() for it in items]
+    merged = merge_row_batch(unwrapped)
+    stringified = [stringify_with_overflow_meta(m) for m in merged]
+    t_flush = time.perf_counter() - t_flush_start
+
+    return t_user, t_flush, len(items), len(merged)
+
+
+# Warmup
+run_workload(10)
+
+print("End-to-end CPU time benchmark")
+print("=" * 60)
+
+for num_requests in [100, 500, 1000, 5000]:
+    t_user, t_flush, num_items, num_merged = run_workload(num_requests)
+    t_total = t_user + t_flush
+
+    print(f"\n{num_requests} requests ({num_items} log items -> {num_merged} merged):")
+    print(f"  User thread:  {t_user*1000:8.1f} ms  ({t_user/num_requests*1e6:6.1f} us/req)")
+    print(f"  Flush:        {t_flush*1000:8.1f} ms  ({t_flush/num_merged*1e6:6.1f} us/item)")
+    print(f"  Total CPU:    {t_total*1000:8.1f} ms  ({t_total/num_requests*1e6:6.1f} us/req)")

py/src/braintrust/logger.py

@@ -4341,7 +4341,7 @@ def log_internal(self, event: dict[str, Any] | None = None, internal_data: dict[
         # Cache the object_id_fields result since it's constant per span.
         # Uses a LazyValue to defer evaluation until flush time (when
         # parent_object_id may still be resolving).
-        if not hasattr(self, '_cached_object_id_fields'):
+        if not hasattr(self, "_cached_object_id_fields"):
             parent_object_type = self.parent_object_type
             parent_object_id = self.parent_object_id
             exporter = _get_exporter()
@@ -4355,13 +4355,15 @@ def log_internal(self, event: dict[str, Any] | None = None, internal_data: dict[
         cached_fields = self._cached_object_id_fields
 
         if lazy_partial_record:
+
             def compute_record() -> dict[str, Any]:
                 return dict(
                     **serializable_partial_record,
                     **{k: v.get() for k, v in lazy_partial_record.items()},
                     **cached_fields.get(),
                 )
         else:
+
             def compute_record() -> dict[str, Any]:
                 return dict(
                     **serializable_partial_record,
@@ -4649,8 +4651,10 @@ def split_logging_data(
             serializable_partial_record: dict[str, Any] = {}
             for k2, v2 in sanitized_and_internal_data.items():
                 if isinstance(v2, BraintrustStream):
+
                     def make_final_value_callback(v2):
                         return LazyValue(lambda: v2.copy().final_value(), use_mutex=False)
+
                     lazy_partial_record[k2] = make_final_value_callback(v2)
                 else:
                     serializable_partial_record[k2] = v2