Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-agents/src/lexigram/ai/agents/observability/tracer.py: 28%

1"""Agent distributed tracing for observability. 

2 

3Integrates with ``lexigram-monitor`` (if available) to provide: 

4- Distributed tracing spans for the reasoning loop 

5- Per-tool call spans 

6- Per-LLM call spans 

7""" 

8 

9from __future__ import annotations 

10 

11from contextlib import asynccontextmanager 

12from typing import TYPE_CHECKING, Any 

13 

14from lexigram.ai.agents.constants import ( 

15 SPAN_AGENT_EXECUTE, 

16 SPAN_AGENT_LLM, 

17 SPAN_AGENT_TOOL, 

18) 

19from lexigram.logging import ( 

20 get_logger, 

21) 

22 

23logger = get_logger(__name__) 

24 

25if TYPE_CHECKING: 

26 from collections.abc import AsyncIterator 

27 

28 from lexigram.contracts.observability.tracing import TracerProtocol 

29 

30 

31class AgentTracer: 

32 """Distributed tracing for agent execution. 

33 

34 Delegates to ``TracerProtocol`` from ``lexigram-monitor`` if 

35 available, otherwise operates as a no-op. 

36 """ 

37 

38 def __init__(self, tracer: TracerProtocol | None = None) -> None: 

39 self._tracer = tracer 

40 

41 @asynccontextmanager 

42 async def trace_execution( 

43 self, 

44 agent_name: str, 

45 message: str, 

46 session_id: str | None = None, 

47 ) -> AsyncIterator[Any]: 

48 """Create a span for the entire agent execution.""" 

49 if not self._tracer: 

50 yield None 

51 return 

52 

53 span = self._tracer.start_span( 

54 f"{SPAN_AGENT_EXECUTE}.{agent_name}", 

55 attributes={ 

56 "agent.name": agent_name, 

57 "agent.session_id": session_id or "", 

58 "agent.message_length": len(message), 

59 }, 

60 ) 

61 

62 try: 

63 yield span 

64 except Exception as e: # tracing instrumentation layer re-raises all exceptions 

65 if hasattr(span, "record_exception"): 

66 span.record_exception(e) 

67 if hasattr(span, "set_status"): 

68 span.set_status("ERROR") 

69 raise 

70 finally: 

71 if hasattr(span, "end"): 

72 span.end() 

73 

74 @asynccontextmanager 

75 async def trace_tool_call( 

76 self, 

77 agent_name: str, 

78 tool_name: str, 

79 ) -> AsyncIterator[Any]: 

80 """Create a span for a tool call.""" 

81 if not self._tracer: 

82 yield None 

83 return 

84 

85 span = self._tracer.start_span( 

86 f"{SPAN_AGENT_TOOL}.{tool_name}", 

87 attributes={ 

88 "agent.name": agent_name, 

89 "tool.name": tool_name, 

90 }, 

91 ) 

92 

93 try: 

94 yield span 

95 except Exception as e: # tracing instrumentation layer re-raises all exceptions 

96 if hasattr(span, "record_exception"): 

97 span.record_exception(e) 

98 raise 

99 finally: 

100 if hasattr(span, "end"): 

101 span.end() 

102 

103 @asynccontextmanager 

104 async def trace_llm_call( 

105 self, 

106 agent_name: str, 

107 iteration: int, 

108 ) -> AsyncIterator[Any]: 

109 """Create a span for an LLM reasoning call.""" 

110 if not self._tracer: 

111 yield None 

112 return 

113 

114 span = self._tracer.start_span( 

115 f"{SPAN_AGENT_LLM}.{agent_name}", 

116 attributes={ 

117 "agent.name": agent_name, 

118 "agent.iteration": iteration, 

119 }, 

120 ) 

121 

122 try: 

123 yield span 

124 except Exception as e: # tracing instrumentation layer re-raises all exceptions 

125 if hasattr(span, "record_exception"): 

126 span.record_exception(e) 

127 raise 

128 finally: 

129 if hasattr(span, "end"): 

130 span.end() 

131 

132 

133__all__ = ["AgentTracer"]