Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/pass_through_endpoints/llm_provider_handlers/base_passthrough_logging_handler.py: 38%

1import json 

2from datetime import datetime 

3from typing import TYPE_CHECKING, Any, Final 

4 

5import httpx 

6 

7import litellm 

8from litellm._logging import verbose_proxy_logger 

9from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj 

10from litellm.litellm_core_utils.litellm_logging import ( 

11 get_standard_logging_object_payload, 

12) 

13from litellm.llms.base_llm.chat.transformation import BaseConfig 

14from litellm.proxy._types import PassThroughEndpointLoggingTypedDict 

15from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body 

16from litellm.types.passthrough_endpoints.pass_through_endpoints import ( 

17 PassthroughStandardLoggingPayload, 

18) 

19from litellm.types.utils import LlmProviders, ModelResponse, TextCompletionResponse 

20 

21if TYPE_CHECKING: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 from ..success_handler import PassThroughEndpointLogging 

23 from ..types import EndpointType 

24else: 

25 PassThroughEndpointLogging = Any 

26 EndpointType = Any 

27 

28from abc import ABC, abstractmethod 

29 

30 

31class BasePassthroughLoggingHandler(ABC): 

32 @property 

33 @abstractmethod 

34 def llm_provider_name(self) -> LlmProviders: 

35 pass 

36 

37 @abstractmethod 

38 def get_provider_config(self, model: str) -> BaseConfig: 

39 pass 

40 

41 def passthrough_chat_handler( 

42 self, 

43 httpx_response: httpx.Response, 

44 response_body: dict, 

45 logging_obj: LiteLLMLoggingObj, 

46 url_route: str, 

47 result: str, 

48 start_time: datetime, 

49 end_time: datetime, 

50 cache_hit: bool, 

51 request_body: dict, 

52 **kwargs, 

53 ) -> PassThroughEndpointLoggingTypedDict: 

54 """ 

55 Transforms LLM response to OpenAI response, generates a standard logging object so downstream logging can be handled 

56 """ 

57 model: Final = request_body.get("model", response_body.get("model", "")) 

58 provider_config: Final = self.get_provider_config(model=model) 

59 litellm_model_response: Final[ModelResponse] = provider_config.transform_response( 

60 raw_response=httpx_response, 

61 model_response=litellm.ModelResponse(), 

62 model=model, 

63 messages=[], 

64 logging_obj=logging_obj, 

65 optional_params={}, 

66 api_key="", 

67 request_data={}, 

68 encoding=litellm.encoding, 

69 json_mode=False, 

70 litellm_params={}, 

71 ) 

72 

73 kwargs = self._create_response_logging_payload( 

74 litellm_model_response=litellm_model_response, 

75 model=model, 

76 kwargs=kwargs, 

77 start_time=start_time, 

78 end_time=end_time, 

79 logging_obj=logging_obj, 

80 ) 

81 

82 return { 

83 "result": litellm_model_response, 

84 "kwargs": kwargs, 

85 } 

86 

87 def _get_user_from_metadata( 

88 self, 

89 passthrough_logging_payload: PassthroughStandardLoggingPayload, 

90 ) -> str | None: 

91 request_body: Final = passthrough_logging_payload.get("request_body") 

92 if request_body: 

93 return get_end_user_id_from_request_body(request_body) 

94 return None 

95 

96 def _create_response_logging_payload( 

97 self, 

98 litellm_model_response: ModelResponse | TextCompletionResponse, 

99 model: str, 

100 kwargs: dict, 

101 start_time: datetime, 

102 end_time: datetime, 

103 logging_obj: LiteLLMLoggingObj, 

104 ) -> dict: 

105 """ 

106 Create the standard logging object for Generic LLM passthrough 

107 

108 handles streaming and non-streaming responses 

109 """ 

110 

111 try: 

112 response_cost: Final = litellm.completion_cost( 

113 completion_response=litellm_model_response, 

114 model=model, 

115 ) 

116 

117 kwargs["response_cost"] = response_cost 

118 kwargs["model"] = model 

119 # the pass-through success path reads spend from 

120 # model_call_details["response_cost"], not from kwargs 

121 logging_obj.model_call_details["response_cost"] = response_cost 

122 passthrough_logging_payload: Final[PassthroughStandardLoggingPayload | None] = kwargs.get( 

123 "passthrough_logging_payload" 

124 ) 

125 if passthrough_logging_payload: 

126 user: Final = self._get_user_from_metadata( 

127 passthrough_logging_payload=passthrough_logging_payload, 

128 ) 

129 if user: 

130 kwargs.setdefault("litellm_params", {}) 

131 kwargs["litellm_params"].update({"proxy_server_request": {"body": {"user": user}}}) 

132 

133 # Make standard logging object for Anthropic 

134 standard_logging_object: Final = get_standard_logging_object_payload( 

135 kwargs=kwargs, 

136 init_response_obj=litellm_model_response, 

137 start_time=start_time, 

138 end_time=end_time, 

139 logging_obj=logging_obj, 

140 status="success", 

141 ) 

142 

143 # pretty print standard logging object 

144 verbose_proxy_logger.debug( 

145 "standard_logging_object= %s", 

146 json.dumps(standard_logging_object, indent=4), 

147 ) 

148 kwargs["standard_logging_object"] = standard_logging_object 

149 

150 # set litellm_call_id to logging response object 

151 litellm_model_response.id = logging_obj.litellm_call_id 

152 litellm_model_response.model = model 

153 logging_obj.model_call_details["model"] = model 

154 return kwargs 

155 except Exception as e: 

156 verbose_proxy_logger.exception("Error creating LLM passthrough response logging payload: %s", e) 

157 return kwargs 

158 

159 @abstractmethod 

160 def _build_complete_streaming_response( 

161 self, 

162 all_chunks: list[str], 

163 litellm_logging_obj: LiteLLMLoggingObj, 

164 model: str, 

165 ) -> ModelResponse | TextCompletionResponse | None: 

166 """ 

167 Builds complete response from raw chunks 

168 

169 - Converts str chunks to generic chunks 

170 - Converts generic chunks to litellm chunks (OpenAI format) 

171 - Builds complete response from litellm chunks 

172 """ 

173 

174 def _handle_logging_llm_collected_chunks( 

175 self, 

176 litellm_logging_obj: LiteLLMLoggingObj, 

177 passthrough_success_handler_obj: PassThroughEndpointLogging, 

178 url_route: str, 

179 request_body: dict, 

180 endpoint_type: EndpointType, 

181 start_time: datetime, 

182 all_chunks: list[str], 

183 end_time: datetime, 

184 ) -> PassThroughEndpointLoggingTypedDict: 

185 """ 

186 Takes raw chunks from Anthropic passthrough endpoint and logs them in litellm callbacks 

187 

188 - Builds complete response from chunks 

189 - Creates standard logging object 

190 - Logs in litellm callbacks 

191 """ 

192 

193 model: Final = request_body.get("model", "") 

194 complete_streaming_response: Final = self._build_complete_streaming_response( 

195 all_chunks=all_chunks, 

196 litellm_logging_obj=litellm_logging_obj, 

197 model=model, 

198 ) 

199 if complete_streaming_response is None: 

200 verbose_proxy_logger.error( 

201 "Unable to build complete streaming response for Anthropic passthrough endpoint, not logging..." 

202 ) 

203 return { 

204 "result": None, 

205 "kwargs": {}, 

206 } 

207 kwargs: Final = self._create_response_logging_payload( 

208 litellm_model_response=complete_streaming_response, 

209 model=model, 

210 kwargs={}, 

211 start_time=start_time, 

212 end_time=end_time, 

213 logging_obj=litellm_logging_obj, 

214 ) 

215 

216 return { 

217 "result": complete_streaming_response, 

218 "kwargs": kwargs, 

219 }