Skip to content

observability_registry

Class info

Classes

Name Children Inherits
AgentOpsProviderConfig
llmling_agent_config.observability
Configuration for AgentOps provider.
    • BaseObservabilityProviderConfig
    ArizePhoenixProviderConfig
    llmling_agent_config.observability
    Configuration for Arize Phoenix provider.
      • BaseObservabilityProviderConfig
      BraintrustProviderConfig
      llmling_agent_config.observability
      Configuration for Braintrust provider.
        • BaseObservabilityProviderConfig
        LaminarProviderConfig
        llmling_agent_config.observability
        Configuration for Laminar provider.
          • BaseObservabilityProviderConfig
          LangsmithProviderConfig
          llmling_agent_config.observability
          Configuration for Langsmith provider.
            • BaseObservabilityProviderConfig
            LogfireProviderConfig
            llmling_agent_config.observability
            Configuration for Logfire provider.
              • BaseObservabilityProviderConfig
              MlFlowProviderConfig
              llmling_agent_config.observability
              Configuration for MlFlow provider.
                • BaseObservabilityProviderConfig
                ObservabilityRegistry
                llmling_agent.observability.observability_registry
                Registry for pending decorations and provider configuration.
                  TraceloopProviderConfig
                  llmling_agent_config.observability
                  Configuration for Traceloop provider.
                    • BaseObservabilityProviderConfig
                    TrackedDecoration
                    llmling_agent.observability.observability_registry
                    Registration of an item (callable / class) that needs observability tracking.

                      🛈 DocStrings

                      Observability registry for tracking items and providers.

                      ObservabilityRegistry

                      Registry for pending decorations and provider configuration.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                       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
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
                      class ObservabilityRegistry:
    """Registry for pending decorations and provider configuration."""

    def __init__(self):
        self._registered_tools: dict[str, TrackedDecoration] = {}
        self._registered_actions: dict[str, TrackedDecoration] = {}
        self.providers: list[ObservabilityProvider] = []
        # to prevent double registration
        self._registered_provider_classes: set[type[ObservabilityProvider]] = set()

    def register_tool(
        self,
        name: str,
        target: Callable,
        **kwargs: Any,
    ):
        """Register a function for tool tracking."""
        self._registered_tools[name] = TrackedDecoration(target, name, kwargs=kwargs)
        logger.debug("Registered tool %r for observability tracking", name)

    def register_action(
        self,
        name: str,
        target: Callable,
        **kwargs: Any,
    ):
        """Register a function for action tracking."""
        self._registered_actions[name] = TrackedDecoration(target, name, kwargs=kwargs)
        msg = "Registered action %r for observability tracking with args %r"
        logger.debug(msg, name, kwargs)

    def configure_provider(self, provider: ObservabilityProvider):
        """Configure a new provider and apply tracking to all registered items.

        When a new provider is configured, it will:
        1. Get added to the list of active providers
        2. Apply its tracking to all previously registered functions/tools/agents
        3. Be available for immediate tracking of new registrations

        The registry maintains a permanent list of what needs tracking,
        collected through decorators at import time. Each provider uses
        these registrations to know what to track.
        """
        msg = "Configuring provider: %s, Current pending actions: %s"
        actions = list(self._registered_actions.keys())
        logger.info(msg, provider.__class__.__name__, actions)
        self.providers.append(provider)

        for pending in self._registered_tools.values():
            try:
                pending.target = provider.wrap_tool(
                    pending.target,
                    pending.name,
                    **pending.kwargs,
                )
                logger.debug("Applied tool tracking to %r", pending.name)
            except Exception:
                msg = "Failed to apply tool tracking to %r"
                logger.exception(msg, pending.name)

        for pending in self._registered_actions.values():
            try:
                pending.target = provider.wrap_action(
                    pending.target,
                    msg_template=pending.name,
                    **pending.kwargs,
                )
                msg = "Applied action tracking to %r with args %r"
                logger.debug(msg, pending.name, pending.kwargs)
            except Exception:
                msg = "Failed to apply action tracking to %r"
                logger.exception(msg, pending.name)

    def register_providers(self, observability_config: ObservabilityConfig):
        """Register and configure all observability providers.

        Args:
            observability_config: Configuration for observability providers
        """
        if not observability_config.enabled:
            return
        for library in observability_config.instrument_libraries or []:
            match library:
                case "pydantic_ai":
                    import pydantic_ai  # noqa

        # Configure each provider
        for provider_config in observability_config.providers:
            provider_cls = get_provider_cls(provider_config)
            if provider_cls not in self._registered_provider_classes:
                provider = provider_cls(provider_config)  # type: ignore
                logger.debug("Registering %s", provider_cls.__name__)
                self._registered_provider_classes.add(provider_cls)
                self.configure_provider(provider)

                      configure_provider 

                      configure_provider(provider: ObservabilityProvider)

                      Configure a new provider and apply tracking to all registered items.

                      When a new provider is configured, it will: 1. Get added to the list of active providers 2. Apply its tracking to all previously registered functions/tools/agents 3. Be available for immediate tracking of new registrations

                      The registry maintains a permanent list of what needs tracking, collected through decorators at import time. Each provider uses these registrations to know what to track.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                       82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
                      def configure_provider(self, provider: ObservabilityProvider):
    """Configure a new provider and apply tracking to all registered items.

    When a new provider is configured, it will:
    1. Get added to the list of active providers
    2. Apply its tracking to all previously registered functions/tools/agents
    3. Be available for immediate tracking of new registrations

    The registry maintains a permanent list of what needs tracking,
    collected through decorators at import time. Each provider uses
    these registrations to know what to track.
    """
    msg = "Configuring provider: %s, Current pending actions: %s"
    actions = list(self._registered_actions.keys())
    logger.info(msg, provider.__class__.__name__, actions)
    self.providers.append(provider)

    for pending in self._registered_tools.values():
        try:
            pending.target = provider.wrap_tool(
                pending.target,
                pending.name,
                **pending.kwargs,
            )
            logger.debug("Applied tool tracking to %r", pending.name)
        except Exception:
            msg = "Failed to apply tool tracking to %r"
            logger.exception(msg, pending.name)

    for pending in self._registered_actions.values():
        try:
            pending.target = provider.wrap_action(
                pending.target,
                msg_template=pending.name,
                **pending.kwargs,
            )
            msg = "Applied action tracking to %r with args %r"
            logger.debug(msg, pending.name, pending.kwargs)
        except Exception:
            msg = "Failed to apply action tracking to %r"
            logger.exception(msg, pending.name)

                      register_action 

                      register_action(name: str, target: Callable, **kwargs: Any)

                      Register a function for action tracking.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                      71
72
73
74
75
76
77
78
79
80
                      def register_action(
    self,
    name: str,
    target: Callable,
    **kwargs: Any,
):
    """Register a function for action tracking."""
    self._registered_actions[name] = TrackedDecoration(target, name, kwargs=kwargs)
    msg = "Registered action %r for observability tracking with args %r"
    logger.debug(msg, name, kwargs)

                      register_providers 

                      register_providers(observability_config: ObservabilityConfig)

                      Register and configure all observability providers.

                      Parameters:

                      Name Type Description Default
                      observability_config ObservabilityConfig

                      Configuration for observability providers

                      		 required
                      Source code in src/llmling_agent/observability/observability_registry.py 
                      124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
                      def register_providers(self, observability_config: ObservabilityConfig):
    """Register and configure all observability providers.

    Args:
        observability_config: Configuration for observability providers
    """
    if not observability_config.enabled:
        return
    for library in observability_config.instrument_libraries or []:
        match library:
            case "pydantic_ai":
                import pydantic_ai  # noqa

    # Configure each provider
    for provider_config in observability_config.providers:
        provider_cls = get_provider_cls(provider_config)
        if provider_cls not in self._registered_provider_classes:
            provider = provider_cls(provider_config)  # type: ignore
            logger.debug("Registering %s", provider_cls.__name__)
            self._registered_provider_classes.add(provider_cls)
            self.configure_provider(provider)

                      register_tool 

                      register_tool(name: str, target: Callable, **kwargs: Any)

                      Register a function for tool tracking.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                      61
62
63
64
65
66
67
68
69
                      def register_tool(
    self,
    name: str,
    target: Callable,
    **kwargs: Any,
):
    """Register a function for tool tracking."""
    self._registered_tools[name] = TrackedDecoration(target, name, kwargs=kwargs)
    logger.debug("Registered tool %r for observability tracking", name)

                      TrackedDecoration dataclass

                      Registration of an item (callable / class) that needs observability tracking.

                      These registrations are permanent and collected at import time through decorators. Each new provider will use these registrations to apply its tracking.

                      The system is a bit more "sophisticated" because everything is working lazily in order to not having to import the observability platforms (and thus, also the AI libraries) on library loading.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                      34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
                      @dataclass
class TrackedDecoration:
    """Registration of an item (callable / class) that needs observability tracking.

    These registrations are permanent and collected at import time through decorators.
    Each new provider will use these registrations to apply its tracking.

    The system is a bit more "sophisticated" because everything is working lazily in
    order to not having to import the observability platforms (and thus, also the AI
    libraries) on library loading.
    """

    target: Any  # class or function to be decorated
    name: str
    kwargs: dict[str, Any]