Spaces:
Sleeping
Sleeping
File size: 2,206 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | """Observability utilities for ACE.
Provides opt-in Logfire integration that auto-instruments all PydanticAI
agents (Agent, Reflector, SkillManager, RR).
Usage::
from ace.observability import configure_logfire
if configure_logfire():
print("Logfire active")
Or via runner::
ace = ACELiteLLM.from_model("gpt-4o-mini", logfire=True)
"""
from __future__ import annotations
import logging
logger = logging.getLogger(__name__)
_logfire_configured = False
def configure_logfire() -> bool:
"""Configure Logfire and instrument PydanticAI agents.
Reads ``LOGFIRE_TOKEN`` from the environment. Set
``LOGFIRE_SEND_TO_LOGFIRE=false`` to disable sending in CI/local dev.
Returns:
``True`` if Logfire was configured successfully, ``False`` if the
``logfire`` package is not installed.
Raises:
No exceptions — returns False on ImportError.
"""
global _logfire_configured
if _logfire_configured:
return True
try:
import logfire
def scrubbing_callback(m: logfire.ScrubMatch):
if m.path == ("attributes", "trace", "reasoning"):
return m.value
if m.path == ("attributes", "trace", "answer"):
return m.value
if "messages" in m.path and "content" in m.path:
return m.value
if "payment_id" in m.path:
return m.value
if "tool_arguments" in m.path:
return m.value
if "tool_response" in m.path:
return m.value
logfire.configure(
scrubbing=logfire.ScrubbingOptions(callback=scrubbing_callback)
)
logfire.instrument_pydantic_ai()
_logfire_configured = True
logger.info("Logfire configured — PydanticAI agents instrumented")
return True
except ImportError:
logger.debug("logfire not installed — skipping instrumentation")
return False
def is_configured() -> bool:
"""Return ``True`` if :func:`configure_logfire` has been called successfully."""
return _logfire_configured
|