feat(logger): support log message templates

Allow `emit_log()` and severity helpers to interpolate named parameters with
Python format strings. Preserve the original template and parameter values in
`braintrust.template` metadata so repeated messages remain queryable by their
stable structure.

Missing placeholders and malformed templates remain unchanged so logging does
not disrupt application code.

Author: AbhiPrasad

py/src/braintrust/logger.py

@@ -142,6 +142,13 @@
 }
 
 
+class _LogTemplateParameters(dict[str, object]):
+    """Preserve placeholders whose values were not provided."""
+
+    def __missing__(self, key: str) -> str:
+        return "{" + key + "}"
+
+
 @dataclasses.dataclass
 class Logs3OverflowInputRow:
     object_ids: dict[str, Any]
@@ -5947,24 +5954,48 @@ def emit_log(
         self,
         body: Any,
         level: LogLevel,
-        metadata: Metadata | None = None,
+        metadata: dict[str, Any] | None = None,
+        **parameters: object,
     ) -> str:
         """Capture a log record, associating it with the active span when one exists.
 
         The log is stored as an independent row. If a Braintrust or OpenTelemetry
         span is active, the row reuses its span and trace IDs for correlation.
         Otherwise, the row uses this logger's baseline trace ID.
 
-        :param body: The log body. May be any JSON-serializable value.
+        String bodies may contain ``str.format``-style placeholders. Keyword
+        parameters are interpolated into the body and retained in metadata along
+        with the original template. Missing parameters remain as placeholders.
+
+        :param body: The log body. May be any JSON-serializable value when no
+            template parameters are provided.
         :param level: The OpenTelemetry log severity: ``trace``, ``debug``,
             ``info``, ``warn``, ``error``, or ``fatal``.
         :param metadata: Optional JSON-serializable attributes for the log.
+        :param parameters: Values for named placeholders in a string body.
         :returns: The unique ID of the captured log row.
         """
         if level not in _OTEL_LOG_LEVELS:
             valid_levels = ", ".join(_OTEL_LOG_LEVELS)
             raise ValueError(f"Invalid log level {level!r}. Expected one of: {valid_levels}")
 
+        rendered_body = body
+        rendered_metadata = metadata
+        if parameters:
+            if not isinstance(body, str):
+                raise TypeError("Log body must be a string when template parameters are provided")
+            rendered_metadata = dict(metadata) if metadata is not None else {}
+            rendered_metadata.update(
+                {f"braintrust.template.parameter.{key}": value for key, value in parameters.items()}
+            )
+            rendered_metadata["braintrust.template"] = body
+            try:
+                rendered_body = body.format_map(_LogTemplateParameters(parameters))
+            except Exception:
+                # Logging should not disrupt the application because a template
+                # contains malformed braces or an unsupported format specifier.
+                rendered_body = body
+
         captured_at = time.time()
         span_info = self.state.context_manager.get_current_span_info()
         severity_number = _OTEL_LOG_LEVELS[level]
@@ -5976,9 +6007,13 @@ def emit_log(
             span_id=span_info.span_id if span_info else None,
             root_span_id=span_info.trace_id if span_info else self._baseline_trace_id,
             lookup_span_parent=False,
-            output=body,
-            error=body if severity_number >= _OTEL_LOG_LEVELS["error"] and isinstance(body, str) else None,
-            metadata=metadata,
+            output=rendered_body,
+            error=(
+                rendered_body
+                if severity_number >= _OTEL_LOG_LEVELS["error"] and isinstance(rendered_body, str)
+                else None
+            ),
+            metadata=rendered_metadata,
             context={
                 "otel": {
                     "signal": "logs",
@@ -5997,29 +6032,29 @@ def emit_log(
 
         return span.id
 
-    def trace(self, body: Any, metadata: Metadata | None = None) -> str:
+    def trace(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
         """Capture a log at OpenTelemetry TRACE severity."""
-        return self.emit_log(body=body, level="trace", metadata=metadata)
+        return self.emit_log(body=body, level="trace", metadata=metadata, **parameters)
 
-    def debug(self, body: Any, metadata: Metadata | None = None) -> str:
+    def debug(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
         """Capture a log at OpenTelemetry DEBUG severity."""
-        return self.emit_log(body=body, level="debug", metadata=metadata)
+        return self.emit_log(body=body, level="debug", metadata=metadata, **parameters)
 
-    def info(self, body: Any, metadata: Metadata | None = None) -> str:
+    def info(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
         """Capture a log at OpenTelemetry INFO severity."""
-        return self.emit_log(body=body, level="info", metadata=metadata)
+        return self.emit_log(body=body, level="info", metadata=metadata, **parameters)
 
-    def warn(self, body: Any, metadata: Metadata | None = None) -> str:
+    def warn(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
         """Capture a log at OpenTelemetry WARN severity."""
-        return self.emit_log(body=body, level="warn", metadata=metadata)
+        return self.emit_log(body=body, level="warn", metadata=metadata, **parameters)
 
-    def error(self, body: Any, metadata: Metadata | None = None) -> str:
+    def error(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
         """Capture a log at OpenTelemetry ERROR severity."""
-        return self.emit_log(body=body, level="error", metadata=metadata)
+        return self.emit_log(body=body, level="error", metadata=metadata, **parameters)
 
-    def fatal(self, body: Any, metadata: Metadata | None = None) -> str:
+    def fatal(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
         """Capture a log at OpenTelemetry FATAL severity."""
-        return self.emit_log(body=body, level="fatal", metadata=metadata)
+        return self.emit_log(body=body, level="fatal", metadata=metadata, **parameters)
 
     def log_feedback(
         self,

py/src/braintrust/test_logger.py

@@ -1488,6 +1488,57 @@ def test_logger_log_level_helpers(with_memory_logger, method_name, severity_numb
     assert row["context"]["otel"]["log"]["severity_number"] == severity_number
 
 
+def test_logger_log_helpers_render_template_parameters(with_memory_logger):
+    test_logger = init_test_logger(__name__)
+
+    log_id = test_logger.info(
+        "User {user_id} paid {amount:.2f} with {method}",
+        metadata={"source": "checkout"},
+        user_id="user-123",
+        amount=12.5,
+    )
+
+    [row] = with_memory_logger.pop()
+    assert row["id"] == log_id
+    assert row["output"] == "User user-123 paid 12.50 with {method}"
+    assert row["metadata"] == {
+        "source": "checkout",
+        "braintrust.template.parameter.user_id": "user-123",
+        "braintrust.template.parameter.amount": 12.5,
+        "braintrust.template": "User {user_id} paid {amount:.2f} with {method}",
+    }
+
+
+def test_logger_error_uses_rendered_template_as_error(with_memory_logger):
+    test_logger = init_test_logger(__name__)
+
+    test_logger.error("Payment {payment_id} failed", payment_id="pay-123")
+
+    [row] = with_memory_logger.pop()
+    assert row["output"] == "Payment pay-123 failed"
+    assert row["error"] == "Payment pay-123 failed"
+
+
+def test_logger_log_helpers_do_not_format_without_parameters(with_memory_logger):
+    test_logger = init_test_logger(__name__)
+
+    test_logger.info('{"key": "{value}"}')
+
+    [row] = with_memory_logger.pop()
+    assert row["output"] == '{"key": "{value}"}'
+    assert not row.get("metadata")
+
+
+def test_logger_log_template_parameters_are_safely_serialized(with_memory_logger):
+    test_logger = init_test_logger(__name__)
+
+    test_logger.warn("Request failed: {error}", error=ValueError("bad request"))
+
+    [row] = with_memory_logger.pop()
+    assert row["output"] == "Request failed: bad request"
+    assert row["metadata"]["braintrust.template.parameter.error"] == "bad request"
+
+
 def test_logger_emit_log_rejects_invalid_level(with_memory_logger):
     test_logger = init_test_logger(__name__)
 

py/src/braintrust/type_tests/test_metadata_types.py

@@ -18,20 +18,27 @@ def dict(self, *, exclude_none: bool = False) -> Mapping[str, Any]:
 
 def accepts_logger_metadata(logger: Logger) -> None:
     mapping_metadata: Mapping[str, Any] = {"user_id": "user-1"}
+    dict_metadata: dict[str, Any] = {"user_id": "user-1"}
 
     logger.log(metadata=mapping_metadata)
     logger.log(metadata=PydanticV2Metadata())
     logger.log(metadata=PydanticV1Metadata())
 
-    logger.emit_log(body="message", level="info", metadata=mapping_metadata)
-    logger.emit_log(body="message", level="info", metadata=PydanticV2Metadata())
-    logger.emit_log(body="message", level="info", metadata=PydanticV1Metadata())
-    logger.trace("message", metadata=mapping_metadata)
-    logger.debug("message", metadata=PydanticV2Metadata())
-    logger.info("message", metadata=PydanticV1Metadata())
-    logger.warn("message", metadata=mapping_metadata)
-    logger.error("message", metadata=PydanticV2Metadata())
-    logger.fatal("message", metadata=PydanticV1Metadata())
+    logger.emit_log(body="message", level="info", metadata=dict_metadata)
+    logger.trace("message", metadata=dict_metadata)
+    logger.debug("message", metadata=dict_metadata)
+    logger.info("message", metadata=dict_metadata)
+    logger.warn("message", metadata=dict_metadata)
+    logger.error("message", metadata=dict_metadata)
+    logger.fatal("message", metadata=dict_metadata)
+
+    logger.emit_log("User {user_id}", "info", metadata=dict_metadata, user_id="user-1")
+    logger.trace("User {user_id}", metadata=dict_metadata, user_id="user-1")
+    logger.debug("User {user_id}", metadata=dict_metadata, user_id="user-1")
+    logger.info("User {user_id}", metadata=dict_metadata, user_id="user-1")
+    logger.warn("User {user_id}", metadata=dict_metadata, user_id="user-1")
+    logger.error("User {user_id}", metadata=dict_metadata, user_id="user-1")
+    logger.fatal("User {user_id}", metadata=dict_metadata, user_id="user-1")
 
     logger.log_feedback(id="event-id", metadata=mapping_metadata)
     logger.log_feedback(id="event-id", metadata=PydanticV2Metadata())