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
      BaseObservabilityProviderConfig
      llmling_agent_config.observability
      Base configuration for observability providers.
      • LogfireProviderConfig
      • AgentOpsProviderConfig
      • LangsmithProviderConfig
      • ArizePhoenixProviderConfig
      • MlFlowProviderConfig
      • BraintrustProviderConfig
      • LaminarProviderConfig
      • TraceloopProviderConfig
      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
                ObservabilityConfig
                llmling_agent_config.observability
                Global observability configuration.
                  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 
                         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
166
167
168
                        class ObservabilityRegistry:
    """Registry for pending decorations and provider configuration."""

    def __init__(self):
        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,
    ):
        """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,
    ):
        """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)

        # 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):
        """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)

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

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

                        Register a function for action tracking.

                        Source code in src/llmling_agent/observability/observability_registry.py 
                        80
81
82
83
84
85
86
87
88
89
                        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_agent 

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

                        Register a class for agent tracking.

                        Source code in src/llmling_agent/observability/observability_registry.py 
                        60
61
62
63
64
65
66
67
68
                        def register_agent(
    self,
    name: str,
    target: type[Any],
    **kwargs: Any,
):
    """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)

                        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 
                        146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
                        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
            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)

                        Register a function for tool tracking.

                        Source code in src/llmling_agent/observability/observability_registry.py 
                        70
71
72
73
74
75
76
77
78
                        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 
                        32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
                        @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]