Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 44 additions & 2 deletions agentops/sdk/decorators/factory.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import inspect
import functools
import asyncio
import os
from typing import Any, Dict, Callable, Optional, Union


Expand All @@ -22,6 +23,46 @@
)


# Track whether we've warned about uninitialized tracer to avoid spamming
_uninitialized_warning_shown = False


def _ensure_tracer_initialized(operation_name: str, entity_kind: str) -> bool:
"""
Ensure the tracer is initialized before using decorators.

Returns True if tracer is initialized (or was auto-initialized), False otherwise.
"""
global _uninitialized_warning_shown

if tracer.initialized:
return True

# Check if auto-initialization is enabled via environment variable
auto_init = os.environ.get("AGENTOPS_AUTO_INIT", "").lower() in ("true", "1", "yes")

if auto_init:
try:
import agentops
agentops.init(auto_start_session=False, instrument_llm_calls=True)
logger.debug(f"AgentOps auto-initialized for decorator '{operation_name}'")
return True
except Exception as e:
logger.warning(f"Failed to auto-initialize AgentOps: {e}")
return False

# Show warning only once to avoid spam
if not _uninitialized_warning_shown:
logger.warning(
f"AgentOps decorator '@{entity_kind}' used on '{operation_name}' but tracer is not initialized. "
f"Spans will not be created. Call agentops.init() first, or set AGENTOPS_AUTO_INIT=true "
f"to auto-initialize when decorators are used."
)
_uninitialized_warning_shown = True

return False


def create_entity_decorator(entity_kind: str) -> Callable[..., Any]:
"""
Factory that creates decorators for instrumenting functions and classes.
Expand Down Expand Up @@ -101,10 +142,11 @@ async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
def wrapper(
wrapped_func: Callable[..., Any], instance: Optional[Any], args: tuple, kwargs: Dict[str, Any]
) -> Any:
if not tracer.initialized:
operation_name = name or wrapped_func.__name__

if not _ensure_tracer_initialized(operation_name, entity_kind):
return wrapped_func(*args, **kwargs)

operation_name = name or wrapped_func.__name__
is_async = asyncio.iscoroutinefunction(wrapped_func)
is_generator = inspect.isgeneratorfunction(wrapped_func)
is_async_generator = inspect.isasyncgenfunction(wrapped_func)
Expand Down
94 changes: 94 additions & 0 deletions tests/unit/sdk/test_decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -761,3 +761,97 @@ def no_cost_tool(self):
span for span in spans if span.attributes.get(SpanAttributes.AGENTOPS_SPAN_KIND) == SpanKind.TOOL
)
assert SpanAttributes.LLM_USAGE_TOOL_COST not in tool_span.attributes


class TestDecoratorInitializationWarning:
"""Tests for decorator behavior when tracer is not initialized."""

def test_decorator_warns_when_tracer_not_initialized(self, caplog):
"""Test that decorators warn when tracer is not initialized."""
import agentops.sdk.decorators.factory as factory_module

# Reset the warning flag to ensure we can test the warning
factory_module._uninitialized_warning_shown = False

# Create a mock tracer state where initialized is False
original_initialized = factory_module.tracer._initialized
factory_module.tracer._initialized = False

try:
@agent
def test_agent_func():
return "test result"

# Call the decorated function
result = test_agent_func()

# Function should still execute and return the result
assert result == "test result"

# Check that warning was logged
assert factory_module._uninitialized_warning_shown is True

finally:
# Restore original state
factory_module.tracer._initialized = original_initialized
factory_module._uninitialized_warning_shown = False

def test_decorator_warning_only_shown_once(self):
"""Test that the warning is only shown once to avoid spam."""
import agentops.sdk.decorators.factory as factory_module

# Reset the warning flag
factory_module._uninitialized_warning_shown = False

original_initialized = factory_module.tracer._initialized
factory_module.tracer._initialized = False

try:
@tool
def test_tool_1():
return "tool 1"

@tool
def test_tool_2():
return "tool 2"

# Call both tools
test_tool_1()
assert factory_module._uninitialized_warning_shown is True

# Reset to check that warning isn't shown again
# (the flag stays True, so second call shouldn't warn)
test_tool_2()
# The flag should still be True (not reset)
assert factory_module._uninitialized_warning_shown is True

finally:
factory_module.tracer._initialized = original_initialized
factory_module._uninitialized_warning_shown = False

def test_decorator_executes_function_when_tracer_not_initialized(self):
"""Test that decorated functions still execute even when tracer is not initialized."""
import agentops.sdk.decorators.factory as factory_module

factory_module._uninitialized_warning_shown = False
original_initialized = factory_module.tracer._initialized
factory_module.tracer._initialized = False

try:
call_count = 0

@operation
def counting_operation(value):
nonlocal call_count
call_count += 1
return value * 2

result = counting_operation(5)

# Function should execute
assert result == 10
assert call_count == 1

finally:
factory_module.tracer._initialized = original_initialized
factory_module._uninitialized_warning_shown = False