Skip to content

observability

Class info

Classes

Name Children Inherits
ObservabilityRegistry
llmling_agent.observability.observability_registry
Registry for pending decorations and provider configuration.

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

    track_action 

    track_action(msg_template: str | None = None, **kwargs: Any) -> Callable[[F], F]

    Register a function for action tracking.

    Source code in src/llmling_agent/observability/decorators.py 
    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
    def track_action(msg_template: str | None = None, **kwargs: Any) -> Callable[[F], F]:
    """Register a function for action tracking."""

    def decorator(func: F) -> F:
        from llmling_agent.observability import registry

        # If we have an active provider, decorate immediately
        logger.info(
            "Decorating function %s.%s with template %s. Current providers: %s",
            func.__module__,
            func.__qualname__,
            msg_template,
            [p.__class__.__name__ for p in registry.providers],
        )
        wrapped = func
        for provider in registry.providers:
            wrapped = provider.wrap_action(wrapped, msg_template=msg_template, **kwargs)  # type: ignore

        # Otherwise queue for later
        action_name = msg_template or func.__name__
        msg = "Queuing action %r with template %r and args %r for later decoration"
        logger.debug(msg, func.__name__, action_name, kwargs)
        registry.register_action(action_name, func, **kwargs)
        return wrapped  # type: ignore

    return decorator

    track_agent 

    track_agent(name: str | None = None, **kwargs: Any) -> Callable[[T], T]

    Register a class for agent tracking.

    Source code in src/llmling_agent/observability/decorators.py 
    15
16
17
18
19
20
21
22
23
24
25
26
    def track_agent(name: str | None = None, **kwargs: Any) -> Callable[[T], T]:
    """Register a class for agent tracking."""

    def decorator(cls: T) -> T:
        from llmling_agent.observability import registry

        agent_name = name or cls.__name__
        logger.debug("Registering agent class %r as %r", cls.__name__, agent_name)
        registry.register_agent(agent_name, cls, **kwargs)
        return cls

    return decorator

    track_tool 

    track_tool(name: str | None = None, **kwargs: Any) -> Callable[[F], F]

    Register a function for tool tracking.

    Source code in src/llmling_agent/observability/decorators.py 
    29
30
31
32
33
34
35
36
37
38
39
40
    def track_tool(name: str | None = None, **kwargs: Any) -> Callable[[F], F]:
    """Register a function for tool tracking."""

    def decorator(func: F) -> F:
        from llmling_agent.observability import registry

        tool_name = name or func.__name__
        logger.debug("Registering tool function %r as %r", func.__name__, tool_name)
        registry.register_tool(tool_name, func, **kwargs)
        return func

    return decorator