Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/routing/loggers/database.py: 25%

1"""Database-backed inference logger for LLM routing. 

2 

3Persists every ``InferenceLog`` entry to the ``inference_log`` table via 

4``DatabaseProviderProtocol``. Suitable for production deployments. 

5 

6All SQL errors are absorbed so that logging failures never interrupt 

7inference. 

8""" 

9 

10from __future__ import annotations 

11 

12from typing import TYPE_CHECKING 

13 

14from lexigram.ai.llm.routing.types import InferenceLog 

15from lexigram.logging import ( 

16 get_logger, 

17) 

18from lexigram.serialization import dumps_str, loads 

19 

20if TYPE_CHECKING: 

21 from lexigram.contracts.data import DatabaseProviderProtocol 

22 

23logger = get_logger(__name__) 

24 

25__all__ = ["DatabaseInferenceLogger"] 

26 

27_INSERT_SQL = """ 

28INSERT INTO inference_log ( 

29 routing_id, 

30 provider, 

31 model, 

32 content, 

33 prompt_tokens, 

34 completion_tokens, 

35 is_paid, 

36 succeeded, 

37 total_attempts, 

38 providers_tried, 

39 context, 

40 error_message, 

41 created_at 

42) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) 

43""" 

44 

45_GET_RECENT_SQL = """ 

46SELECT 

47 routing_id, 

48 provider, 

49 model, 

50 content, 

51 prompt_tokens, 

52 completion_tokens, 

53 is_paid, 

54 succeeded, 

55 total_attempts, 

56 providers_tried, 

57 context, 

58 error_message, 

59 created_at 

60FROM inference_log 

61ORDER BY created_at DESC 

62LIMIT $1 

63""" 

64 

65 

66class DatabaseInferenceLogger: 

67 """PostgreSQL-backed inference logger using ``DatabaseProviderProtocol``. 

68 

69 Every call to :meth:`log` performs a best-effort INSERT; failures are 

70 swallowed and only emitted as ``ERROR``-level log lines so that 

71 observability failures never break inference. 

72 

73 Example: 

74 >>> db_logger = DatabaseInferenceLogger(db=db_provider) 

75 >>> await db_logger.log(some_inference_log) 

76 >>> recent = await db_logger.get_recent(limit=20) 

77 """ 

78 

79 def __init__(self, db: DatabaseProviderProtocol) -> None: 

80 """Initialise the database inference logger. 

81 

82 Args: 

83 db: Framework database provider (injected from the DI container). 

84 """ 

85 self._db = db 

86 

87 async def log(self, entry: InferenceLog) -> None: 

88 """Persist one ``InferenceLog`` entry. 

89 

90 Args: 

91 entry: :class:`~lexigram.ai.llm.routing.types.InferenceLog` instance. 

92 """ 

93 try: 

94 provider = entry.result.provider if entry.result else None 

95 model = entry.result.model if entry.result else None 

96 content = entry.result.content if entry.result else None 

97 prompt_tokens = entry.result.prompt_tokens if entry.result else 0 

98 completion_tokens = entry.result.completion_tokens if entry.result else 0 

99 is_paid = entry.result.is_paid if entry.result else False 

100 error_message = entry.error.message if entry.error else None 

101 

102 async with self._db.scoped_context(): 

103 conn = await self._db.get_scoped_connection() 

104 await conn.execute( 

105 _INSERT_SQL, 

106 entry.routing_id, 

107 provider, 

108 model, 

109 content, 

110 prompt_tokens, 

111 completion_tokens, 

112 is_paid, 

113 entry.succeeded, 

114 entry.total_attempts, 

115 entry.providers_tried, 

116 dumps_str(entry.context), 

117 error_message, 

118 entry.created_at, 

119 ) 

120 except Exception as e: 

121 logger.exception( 

122 "llm.inference_log.db: failed to persist routing_id=%s", 

123 entry.routing_id, 

124 error=str(e), 

125 ) 

126 

127 async def get_recent(self, limit: int = 100) -> list[InferenceLog]: 

128 """Return the most recent *limit* log entries (newest-first). 

129 

130 Args: 

131 limit: Maximum number of entries to return. 

132 

133 Returns: 

134 List of :class:`~lexigram.ai.llm.routing.types.InferenceLog` instances, 

135 or an empty list on DB error. 

136 """ 

137 try: 

138 from lexigram.ai.llm.routing.types import ( 

139 InferenceError, 

140 InferenceResult, 

141 ) 

142 

143 async with self._db.scoped_context(): 

144 conn = await self._db.get_scoped_connection() 

145 rows = await conn.fetch(_GET_RECENT_SQL, limit) 

146 

147 entries: list[InferenceLog] = [] 

148 for row in rows: 

149 result: InferenceResult | None = None 

150 error: InferenceError | None = None 

151 

152 if row["succeeded"] and row["provider"]: 

153 result = InferenceResult( 

154 provider=row["provider"], 

155 model=row["model"] or "", 

156 content=row["content"] or "", 

157 prompt_tokens=row["prompt_tokens"] or 0, 

158 completion_tokens=row["completion_tokens"] or 0, 

159 is_paid=row["is_paid"] or False, 

160 ) 

161 elif row["error_message"]: 

162 error = InferenceError( 

163 message=row["error_message"], 

164 providers_tried=list(row["providers_tried"] or []), 

165 ) 

166 

167 ctx = row["context"] 

168 if isinstance(ctx, str): 

169 try: 

170 ctx = loads(ctx) 

171 except (ValueError, KeyError, TypeError): 

172 ctx = {} 

173 

174 entries.append( 

175 InferenceLog( 

176 routing_id=row["routing_id"], 

177 result=result, 

178 error=error, 

179 providers_tried=list(row["providers_tried"] or []), 

180 total_attempts=row["total_attempts"] or 0, 

181 context=ctx or {}, 

182 created_at=row["created_at"], 

183 ) 

184 ) 

185 return entries 

186 except Exception as e: 

187 logger.exception( 

188 "llm.inference_log.db: get_recent query failed", error=str(e) 

189 ) 

190 return []