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

1"""Agent metrics collection for observability. 

2 

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

4- Per-agent execution metrics (duration, token usage, tool calls) 

5- Per-tool call metrics (duration, success/failure) 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING 

11 

12from lexigram.ai.agents.constants import ( 

13 METRIC_AGENT_EXECUTION_DURATION_MS, 

14 METRIC_AGENT_EXECUTION_STEPS, 

15 METRIC_AGENT_EXECUTION_TOKENS, 

16 METRIC_AGENT_EXECUTION_TOOL_CALLS, 

17 METRIC_AGENT_EXECUTIONS_ERRORS, 

18 METRIC_AGENT_EXECUTIONS_TOTAL, 

19 METRIC_AGENT_GOVERNANCE_DENIED, 

20 METRIC_AGENT_TOOL_CALL_DURATION_MS, 

21 METRIC_AGENT_TOOL_CALLS_FAILED, 

22 METRIC_AGENT_TOOL_CALLS_FAILURE, 

23 METRIC_AGENT_TOOL_CALLS_SUCCESS, 

24 METRIC_AGENT_TOOL_CALLS_TOTAL, 

25) 

26from lexigram.logging import ( 

27 get_logger, 

28) 

29 

30logger = get_logger(__name__) 

31 

32if TYPE_CHECKING: 

33 from lexigram.ai.agents.types import ToolExecutionRecord 

34 from lexigram.contracts.ai.agents import AgentResponse 

35 from lexigram.contracts.observability.metrics import MetricsRecorderProtocol 

36 

37 

38class AgentMetrics: 

39 """Collects metrics for agent execution. 

40 

41 Delegates to ``MetricsRecorderProtocol`` from ``lexigram-monitor`` if 

42 available, otherwise operates as a no-op. 

43 """ 

44 

45 def __init__(self, recorder: MetricsRecorderProtocol | None = None) -> None: 

46 self._recorder = recorder 

47 

48 def record_execution( 

49 self, 

50 agent_name: str, 

51 response: AgentResponse, 

52 ) -> None: 

53 """Record metrics for a completed agent execution.""" 

54 if not self._recorder: 

55 return 

56 

57 tags = {"agent": agent_name} 

58 

59 self._recorder.increment(METRIC_AGENT_EXECUTIONS_TOTAL, tags=tags) 

60 self._recorder.histogram( 

61 METRIC_AGENT_EXECUTION_DURATION_MS, 

62 response.duration_ms, 

63 tags=tags, 

64 ) 

65 self._recorder.histogram( 

66 METRIC_AGENT_EXECUTION_TOKENS, 

67 float(response.total_tokens), 

68 tags=tags, 

69 ) 

70 self._recorder.histogram( 

71 METRIC_AGENT_EXECUTION_STEPS, 

72 float(response.step_count), 

73 tags=tags, 

74 ) 

75 self._recorder.histogram( 

76 METRIC_AGENT_EXECUTION_TOOL_CALLS, 

77 float(response.tool_call_count), 

78 tags=tags, 

79 ) 

80 

81 if response.failed_tool_calls: 

82 self._recorder.increment( 

83 METRIC_AGENT_TOOL_CALLS_FAILED, 

84 value=float(len(response.failed_tool_calls)), 

85 tags=tags, 

86 ) 

87 

88 def record_tool_call( 

89 self, 

90 agent_name: str, 

91 tool_call: ToolExecutionRecord, 

92 ) -> None: 

93 """Record metrics for a single tool call.""" 

94 if not self._recorder: 

95 return 

96 

97 tags = {"agent": agent_name, "tool": tool_call.tool_name} 

98 

99 self._recorder.increment(METRIC_AGENT_TOOL_CALLS_TOTAL, tags=tags) 

100 self._recorder.histogram( 

101 METRIC_AGENT_TOOL_CALL_DURATION_MS, 

102 tool_call.duration_ms, 

103 tags=tags, 

104 ) 

105 

106 if tool_call.succeeded: 

107 self._recorder.increment(METRIC_AGENT_TOOL_CALLS_SUCCESS, tags=tags) 

108 else: 

109 self._recorder.increment(METRIC_AGENT_TOOL_CALLS_FAILURE, tags=tags) 

110 

111 def record_error( 

112 self, 

113 agent_name: str, 

114 error_type: str, 

115 ) -> None: 

116 """Record an agent execution error.""" 

117 if not self._recorder: 

118 return 

119 

120 self._recorder.increment( 

121 METRIC_AGENT_EXECUTIONS_ERRORS, 

122 tags={"agent": agent_name, "error_type": error_type}, 

123 ) 

124 

125 def record_governance_denied(self, agent_name: str) -> None: 

126 """Record a governance denial.""" 

127 if not self._recorder: 

128 return 

129 

130 self._recorder.increment( 

131 METRIC_AGENT_GOVERNANCE_DENIED, 

132 tags={"agent": agent_name}, 

133 ) 

134 

135 

136__all__ = ["AgentMetrics"]