Skip to content

observability_registry

Class info

Classes

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

                      🛈 DocStrings

                      ObservabilityRegistry

                      Registry for pending decorations and provider configuration.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                       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
 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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
                      class ObservabilityRegistry:
    """Registry for pending decorations and provider configuration."""

    def __init__(self) -> None:
        self._registered_agents: dict[str, TrackedDecoration] = {}
        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_agent(
        self,
        name: str,
        target: type[Any],
        **kwargs: Any,
    ) -> None:
        """Register a class for agent tracking."""
        self._registered_agents[name] = TrackedDecoration(target, name, kwargs=kwargs)
        logger.debug("Registered agent %r for observability tracking", name)

    def register_tool(
        self,
        name: str,
        target: Callable,
        **kwargs: Any,
    ) -> None:
        """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,
    ) -> None:
        """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) -> None:
        """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)

        # Apply decorations for each type
        for pending in self._registered_agents.values():
            try:
                pending.target = provider.wrap_agent(
                    pending.target,
                    pending.name,
                    **pending.kwargs,
                )
                logger.debug("Applied agent tracking to %r", pending.name)
            except Exception:
                msg = "Failed to apply agent tracking to %r"
                logger.exception(msg, pending.name)

        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) -> None:
        """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
                case "litellm":
                    import litellm  # 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) -> None

                      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 
                       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
                      def configure_provider(self, provider: ObservabilityProvider) -> None:
    """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)

    # Apply decorations for each type
    for pending in self._registered_agents.values():
        try:
            pending.target = provider.wrap_agent(
                pending.target,
                pending.name,
                **pending.kwargs,
            )
            logger.debug("Applied agent tracking to %r", pending.name)
        except Exception:
            msg = "Failed to apply agent tracking to %r"
            logger.exception(msg, pending.name)

    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) -> None

                      Register a function for action tracking.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                      77
78
79
80
81
82
83
84
85
86
                      def register_action(
    self,
    name: str,
    target: Callable,
    **kwargs: Any,
) -> None:
    """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_agent 

                      register_agent(name: str, target: type[Any], **kwargs: Any) -> None

                      Register a class for agent tracking.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                      57
58
59
60
61
62
63
64
65
                      def register_agent(
    self,
    name: str,
    target: type[Any],
    **kwargs: Any,
) -> None:
    """Register a class for agent tracking."""
    self._registered_agents[name] = TrackedDecoration(target, name, kwargs=kwargs)
    logger.debug("Registered agent %r for observability tracking", name)

                      register_providers 

                      register_providers(observability_config: ObservabilityConfig) -> None

                      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 
                      143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
                      def register_providers(self, observability_config: ObservabilityConfig) -> None:
    """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
            case "litellm":
                import litellm  # 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) -> None

                      Register a function for tool tracking.

                      Source code in src/llmling_agent/observability/observability_registry.py 
                      67
68
69
70
71
72
73
74
75
                      def register_tool(
    self,
    name: str,
    target: Callable,
    **kwargs: Any,
) -> None:
    """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 
                      29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
                      @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]