Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/pass_through_endpoints/llm_provider_handlers/gemini_passthrough_logging_handler.py: 23%
93 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import re
2from datetime import datetime
3from typing import TYPE_CHECKING, Any, Final
5import httpx
7import litellm
8from litellm._logging import verbose_proxy_logger
9from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
10from litellm.llms.gemini.videos.transformation import GeminiVideoConfig
11from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
12 ModelResponseIterator as GeminiModelResponseIterator,
13)
14from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
15from litellm.proxy.pass_through_endpoints.llm_provider_handlers.vertex_passthrough_logging_handler import (
16 VertexPassthroughLoggingHandler,
17)
18from litellm.types.utils import (
19 ModelResponse,
20 TextCompletionResponse,
21)
23if TYPE_CHECKING: 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true
24 from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
26 from ..success_handler import PassThroughEndpointLogging
27else:
28 PassThroughEndpointLogging = Any
29 EndpointType = Any
32class GeminiPassthroughLoggingHandler:
33 @staticmethod
34 def gemini_passthrough_handler(
35 httpx_response: httpx.Response,
36 response_body: dict,
37 logging_obj: LiteLLMLoggingObj,
38 url_route: str,
39 result: str,
40 start_time: datetime,
41 end_time: datetime,
42 cache_hit: bool,
43 request_body: dict,
44 **kwargs,
45 ) -> PassThroughEndpointLoggingTypedDict:
46 if VertexPassthroughLoggingHandler.is_interactions_route(url_route):
47 return VertexPassthroughLoggingHandler.interactions_passthrough_handler(
48 httpx_response=httpx_response,
49 request_body=request_body,
50 logging_obj=logging_obj,
51 kwargs=kwargs,
52 start_time=start_time,
53 end_time=end_time,
54 custom_llm_provider="gemini",
55 vertex_location=None,
56 )
57 if "predictLongRunning" in url_route:
58 model = GeminiPassthroughLoggingHandler.extract_model_from_url(url_route)
60 gemini_video_config: Final = GeminiVideoConfig()
61 litellm_video_response: Final = gemini_video_config.transform_video_create_response(
62 model=model,
63 raw_response=httpx_response,
64 logging_obj=logging_obj,
65 custom_llm_provider="gemini",
66 request_data=request_body,
67 )
68 logging_obj.model = model
69 logging_obj.model_call_details["model"] = model
70 logging_obj.model_call_details["custom_llm_provider"] = "gemini"
71 logging_obj.custom_llm_provider = "gemini"
73 response_cost: Final = litellm.completion_cost(
74 completion_response=litellm_video_response,
75 model=model,
76 custom_llm_provider="gemini",
77 call_type="create_video",
78 )
80 # Set response_cost in _hidden_params to prevent recalculation
81 if not hasattr(litellm_video_response, "_hidden_params"):
82 litellm_video_response._hidden_params = {}
83 litellm_video_response._hidden_params["response_cost"] = response_cost
85 kwargs["response_cost"] = response_cost
86 kwargs["model"] = model
87 kwargs["custom_llm_provider"] = "gemini"
88 logging_obj.model_call_details["response_cost"] = response_cost
89 return {
90 "result": litellm_video_response,
91 "kwargs": kwargs,
92 }
94 if "generateContent" in url_route:
95 model = GeminiPassthroughLoggingHandler.extract_model_from_url(url_route)
97 # Use Gemini config for transformation
98 instance_of_gemini_llm: Final = litellm.GoogleAIStudioGeminiConfig()
99 litellm_model_response: Final[ModelResponse] = instance_of_gemini_llm.transform_response(
100 model=model,
101 messages=[{"role": "user", "content": "no-message-pass-through-endpoint"}],
102 raw_response=httpx_response,
103 model_response=litellm.ModelResponse(),
104 logging_obj=logging_obj,
105 optional_params={},
106 litellm_params={},
107 api_key="",
108 request_data={},
109 encoding=getattr(litellm, "encoding", None),
110 )
111 kwargs = GeminiPassthroughLoggingHandler._create_gemini_response_logging_payload_for_generate_content(
112 litellm_model_response=litellm_model_response,
113 model=model,
114 kwargs=kwargs,
115 start_time=start_time,
116 end_time=end_time,
117 logging_obj=logging_obj,
118 custom_llm_provider="gemini",
119 )
121 return {
122 "result": litellm_model_response,
123 "kwargs": kwargs,
124 }
125 else:
126 return {
127 "result": None,
128 "kwargs": kwargs,
129 }
131 @staticmethod
132 def _handle_logging_gemini_collected_chunks(
133 litellm_logging_obj: LiteLLMLoggingObj,
134 passthrough_success_handler_obj: PassThroughEndpointLogging,
135 url_route: str,
136 request_body: dict,
137 endpoint_type: EndpointType,
138 start_time: datetime,
139 all_chunks: list[str],
140 model: str | None,
141 end_time: datetime,
142 ) -> PassThroughEndpointLoggingTypedDict:
143 """
144 Takes raw chunks from Gemini passthrough endpoint and logs them in litellm callbacks
146 - Builds complete response from chunks
147 - Creates standard logging object
148 - Logs in litellm callbacks
149 """
150 kwargs: dict[str, object] = {}
151 model = model or GeminiPassthroughLoggingHandler.extract_model_from_url(url_route)
152 complete_streaming_response: Final = GeminiPassthroughLoggingHandler._build_complete_streaming_response(
153 all_chunks=all_chunks,
154 litellm_logging_obj=litellm_logging_obj,
155 model=model,
156 url_route=url_route,
157 )
159 if complete_streaming_response is None:
160 verbose_proxy_logger.error(
161 "Unable to build complete streaming response for Gemini passthrough endpoint, not logging..."
162 )
163 return {
164 "result": None,
165 "kwargs": kwargs,
166 }
168 kwargs = GeminiPassthroughLoggingHandler._create_gemini_response_logging_payload_for_generate_content(
169 litellm_model_response=complete_streaming_response,
170 model=model,
171 kwargs=kwargs,
172 start_time=start_time,
173 end_time=end_time,
174 logging_obj=litellm_logging_obj,
175 custom_llm_provider="gemini",
176 )
178 return {
179 "result": complete_streaming_response,
180 "kwargs": kwargs,
181 }
183 @staticmethod
184 def _build_complete_streaming_response(
185 all_chunks: list[str],
186 litellm_logging_obj: LiteLLMLoggingObj,
187 model: str,
188 url_route: str,
189 ) -> ModelResponse | TextCompletionResponse | None:
190 parsed_chunks = []
191 if "generateContent" in url_route or "streamGenerateContent" in url_route:
192 gemini_iterator: Final[Any] = GeminiModelResponseIterator(
193 streaming_response=None,
194 sync_stream=False,
195 logging_obj=litellm_logging_obj,
196 )
197 chunk_parsing_logic: Final[Any] = gemini_iterator._common_chunk_parsing_logic
198 parsed_chunks = [chunk_parsing_logic(chunk) for chunk in all_chunks]
199 else:
200 return None
202 if len(parsed_chunks) == 0:
203 return None
205 all_openai_chunks: Final = []
206 for parsed_chunk in parsed_chunks:
207 if parsed_chunk is None:
208 continue
209 all_openai_chunks.append(parsed_chunk)
211 complete_streaming_response: Final = litellm.stream_chunk_builder(chunks=all_openai_chunks)
213 return complete_streaming_response
215 @staticmethod
216 def extract_model_from_url(url: str) -> str:
217 pattern: Final = r"/models/([^:]+)"
218 match: Final = re.search(pattern, url)
219 if match:
220 return match.group(1)
221 return "unknown"
223 @staticmethod
224 def _create_gemini_response_logging_payload_for_generate_content(
225 litellm_model_response: ModelResponse | TextCompletionResponse,
226 model: str,
227 kwargs: dict,
228 start_time: datetime,
229 end_time: datetime,
230 logging_obj: LiteLLMLoggingObj,
231 custom_llm_provider: str,
232 ):
233 """
234 Create the standard logging object for Gemini passthrough generateContent (streaming and non-streaming)
235 """
237 response_cost: Final = litellm.completion_cost(
238 completion_response=litellm_model_response,
239 model=model,
240 custom_llm_provider="gemini",
241 )
243 kwargs["response_cost"] = response_cost
244 kwargs["model"] = model
245 kwargs["custom_llm_provider"] = custom_llm_provider
247 # pretty print standard logging object
248 verbose_proxy_logger.debug("kwargs= %s", kwargs)
250 # set litellm_call_id to logging response object
251 litellm_model_response.id = logging_obj.litellm_call_id
252 logging_obj.model = litellm_model_response.model or model
253 logging_obj.model_call_details["model"] = logging_obj.model
254 logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider
255 logging_obj.model_call_details["response_cost"] = response_cost
256 return kwargs