calute.logging.structured

calute.logging.structured#

Structured logging configuration for Calute.

This module provides an advanced logging configuration system for the Calute framework with support for: - Structured logging via structlog (when available) - OpenTelemetry distributed tracing integration - Prometheus metrics for observability - JSON logging format support - Rotating file handlers for log persistence

The module gracefully degrades when optional dependencies are not installed, providing fallback implementations where necessary.

Example

>>> from calute.logging.structured import get_logger, configure_logging
>>> configure_logging(level="DEBUG", enable_json=True)
>>> logger = get_logger()
>>> logger.log_function_call("agent1", "search", {"query": "test"})
class calute.logging.structured.CaluteLogger(name: str = 'calute', level: str = 'INFO', log_file: pathlib.Path | None = None, enable_json: bool = True, enable_tracing: bool = False, trace_endpoint: str | None = None)[source]#

Bases: object

Enhanced logger with structured logging and tracing.

Provides a comprehensive logging solution that integrates with structlog for structured logging, OpenTelemetry for distributed tracing, and Prometheus for metrics collection.

name#

Logger name identifier.

level#

Numeric logging level.

log_file#

Optional path to log file for persistence.

enable_json#

Whether JSON output format is enabled.

enable_tracing#

Whether OpenTelemetry tracing is enabled.

logger#

The underlying logger instance (structlog or standard).

Note

Falls back to standard Python logging if structlog is not installed.

get_metrics() → bytes[source]#

Get Prometheus metrics in text format.

Returns

Prometheus metrics formatted for HTTP exposition.

log_agent_switch(from_agent: str, to_agent: str, reason: str | None = None)[source]#

Log an agent switch event with metrics.

Records when control transfers from one agent to another, updating the agent switch counter for monitoring.

Parameters

  • from_agent – Name of the agent relinquishing control.

  • to_agent – Name of the agent receiving control.

  • reason – Optional reason for the switch.

log_function_call(agent_id: str, function_name: str, arguments: dict[str, Any], result: Any = None, error: Exception | None = None, duration: float = 0.0)[source]#

Log a function call with metrics and structured data.

Records function execution details including arguments, results, and timing. Updates Prometheus metrics for monitoring.

Parameters

  • agent_id – Identifier of the agent executing the function.

  • function_name – Name of the function being called.

  • arguments – Dictionary of function arguments.

  • result – Optional function result (truncated to 200 chars in logs).

  • error – Optional exception if the function failed.

  • duration – Execution duration in seconds.

log_llm_request(provider: str, model: str, prompt_tokens: int, completion_tokens: int, duration: float, error: Exception | None = None)[source]#

Log an LLM API request with token and timing metrics.

Records LLM request details for monitoring and cost tracking, updating Prometheus counters for requests and tokens.

Parameters

  • provider – LLM provider name (e.g., ‘openai’, ‘anthropic’).

  • model – Model identifier (e.g., ‘gpt-4’, ‘claude-3’).

  • prompt_tokens – Number of tokens in the prompt.

  • completion_tokens – Number of tokens in the completion.

  • duration – Request duration in seconds.

  • error – Optional exception if the request failed.

log_memory_operation(operation: str, memory_type: str, agent_id: str, entry_count: int = 1, error: Exception | None = None)[source]#

Log a memory store operation with metrics.

Records memory operations (add, remove) and updates the memory usage gauge for monitoring.

Parameters

  • operation – Type of operation (‘add’ or ‘remove’).

  • memory_type – Type of memory (e.g., ‘short_term’, ‘long_term’).

  • agent_id – Identifier of the agent performing the operation.

  • entry_count – Number of entries affected by the operation.

  • error – Optional exception if the operation failed.

span(name: str, **attributes)[source]#

Create a tracing span context manager.

Creates an OpenTelemetry span for distributed tracing. The span automatically records exceptions and sets error status on failure.

Parameters

  • name – Name for the span.

  • **attributes – Additional attributes to attach to the span.

Yields

The created span object, or None if tracing is disabled.

Example

>>> with logger.span("process_request", user_id="123"):
...     process_data()
calute.logging.structured.Counter(*args, **kwargs)#
class calute.logging.structured.DummyMetric[source]#

Bases: object

No-op metric implementation for when prometheus_client is not installed.

Provides the same interface as Prometheus Counter, Gauge, and Histogram metrics but performs no actual operations, allowing code to function without the prometheus_client dependency.

All methods are intentionally no-ops and return self where appropriate to support fluent method chaining (e.g., metric.labels(foo="bar").inc()).

None. This class carries no state.
dec(amount=1)[source]#

No-op decrement operation.

Mimics the Prometheus Gauge.dec method.

Parameters

amount – The amount to decrement by (ignored). Defaults to 1.

inc(amount=1)[source]#

No-op increment operation.

Mimics the Prometheus Counter.inc / Gauge.inc method.

Parameters

amount – The amount to increment by (ignored). Defaults to 1.

labels(**kwargs)[source]#

Return self for method chaining compatibility.

Mimics the Prometheus labels() method which returns a child metric filtered by the given label values.

Parameters

**kwargs – Label key-value pairs (ignored).

Returns

self, allowing subsequent calls to inc, dec, or observe to succeed silently.

observe(value)[source]#

No-op observation operation.

Mimics the Prometheus Histogram.observe method.

Parameters

value – The observed value (ignored).

calute.logging.structured.Gauge(*args, **kwargs)#
calute.logging.structured.Histogram(*args, **kwargs)#
calute.logging.structured.configure_logging(**kwargs)[source]#

Configure global logging settings with custom parameters.

Creates a new CaluteLogger with the specified settings and sets it as the global logger instance.

Parameters

**kwargs – Configuration options passed to CaluteLogger: - name: Logger name identifier. - level: Log level string. - log_file: Optional Path for log file. - enable_json: Whether to use JSON format. - enable_tracing: Whether to enable tracing. - trace_endpoint: OTLP endpoint for traces.

Example

>>> configure_logging(level="DEBUG", enable_json=False)
calute.logging.structured.generate_latest()[source]#

Return empty bytes when prometheus_client is not installed.

This is a fallback replacement for prometheus_client.generate_latest that returns an empty byte string when the Prometheus client library is unavailable.

Returns

An empty bytes object (b"").

calute.logging.structured.get_logger(name: str | None = None, **kwargs) → CaluteLogger[source]#

Get or create the global CaluteLogger instance.

Lazily initializes the logger using configuration from the global Calute config. Subsequent calls return the same instance.

Parameters

  • name – Optional logger name (used only on first initialization).

  • **kwargs – Additional keyword arguments (currently unused).

Returns

The global CaluteLogger instance.

Example

>>> logger = get_logger()
>>> logger.log_function_call("agent1", "search", {"q": "test"})
Contents