From 9a7a1d7c2fdbc8078215e6a65d739bcd1cc12b1b Mon Sep 17 00:00:00 2001 From: snaily Date: Sun, 20 Apr 2025 12:02:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E6=97=A5=E5=BF=97):=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BA=93=E6=97=A5=E5=BF=97=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E5=B9=B6=E5=A2=9E=E5=BC=BAAPI=E9=87=8D=E8=AF=95/=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 为 Gemini 聊天（流式/非流式）、OpenAI 图像聊天（流式/非流式）和 embedding 服务的 API 调用实现全面的数据库日志记录。日志包括请求详情、成功/失败状态、状态码、延迟和错误消息。 - 重构 Gemini 流式聊天服务 (`stream_generate_content`) 以整合使用 `KeyManager` 的重试逻辑，与非流式实现保持一致，包括失败时的 API 密钥切换。 - 增强重试处理器 (`RetryHandler`) 的日志记录，以提高密钥切换和失败场景下的清晰度。 - 确保 `api_key` 正确传递给 OpenAI 图像聊天完成。 - 改进 embedding 服务中的错误处理，区分 `APIStatusError` 和通用异常，并将错误记录到数据库。 - 为 embedding 服务日志添加请求负载截断。 - 修复 Gemini `_build_payload` 中使用正确的 `model` 变量获取 `THINKING_BUDGET_MAP` 的错误。 - 移除 `ImageCreateService` 中未使用的 `paid_key` 类变量。 --- app/handler/retry_handler.py | 13 +- app/router/gemini_routes.py | 2 + app/router/openai_routes.py | 2 +- app/service/chat/gemini_chat_service.py | 180 +++++------ app/service/chat/openai_chat_service.py | 348 +++++++++++++-------- app/service/embedding/embedding_service.py | 65 +++- app/service/image/image_create_service.py | 3 +- 7 files changed, 373 insertions(+), 240 deletions(-) diff --git a/app/handler/retry_handler.py b/app/handler/retry_handler.py index 116f604..66cdcfd 100644 --- a/app/handler/retry_handler.py +++ b/app/handler/retry_handler.py @@ -23,21 +23,26 @@ class RetryHandler: last_exception = None for attempt in range(self.max_retries): + retries = attempt + 1 try: return await func(*args, **kwargs) except Exception as e: last_exception = e logger.warning( - f"API call failed with error: {str(e)}. Attempt {attempt + 1} of {self.max_retries}" + f"API call failed with error: {str(e)}. Attempt {retries} of {self.max_retries}" ) # 从函数参数中获取 key_manager key_manager = kwargs.get("key_manager") if key_manager: old_key = kwargs.get(self.key_arg) - new_key = await key_manager.handle_api_failure(old_key, attempt) - kwargs[self.key_arg] = new_key - logger.info(f"Switched to new API key: {new_key}") + new_key = await key_manager.handle_api_failure(old_key, retries) + if new_key: + kwargs[self.key_arg] = new_key + logger.info(f"Switched to new API key: {new_key}") + else: + logger.error(f"No valid API key available after {retries} retries.") + break logger.error( f"All retry attempts failed, raising final exception: {str(last_exception)}" diff --git a/app/router/gemini_routes.py b/app/router/gemini_routes.py index 3140111..0b6b3c5 100644 --- a/app/router/gemini_routes.py +++ b/app/router/gemini_routes.py @@ -109,6 +109,7 @@ async def generate_content( request: GeminiRequest, _=Depends(security_service.verify_key_or_goog_api_key), api_key: str = Depends(get_next_working_key), + key_manager: KeyManager = Depends(get_key_manager), chat_service: GeminiChatService = Depends(get_chat_service) ): """非流式生成内容""" @@ -140,6 +141,7 @@ async def stream_generate_content( request: GeminiRequest, _=Depends(security_service.verify_key_or_goog_api_key), api_key: str = Depends(get_next_working_key), + key_manager: KeyManager = Depends(get_key_manager), chat_service: GeminiChatService = Depends(get_chat_service) ): """流式生成内容""" diff --git a/app/router/openai_routes.py b/app/router/openai_routes.py index 4757468..e52a29b 100644 --- a/app/router/openai_routes.py +++ b/app/router/openai_routes.py @@ -86,7 +86,7 @@ async def chat_completion( try: # 如果model是imagen3,使用paid_key if request.model == f"{settings.CREATE_IMAGE_MODEL}-chat": - response = await chat_service.create_image_chat_completion(request=request) + response = await chat_service.create_image_chat_completion(request, api_key) else: response = await chat_service.create_chat_completion(request, api_key) # 处理流式响应 diff --git a/app/service/chat/gemini_chat_service.py b/app/service/chat/gemini_chat_service.py index 5d1b42d..bb3c874 100644 --- a/app/service/chat/gemini_chat_service.py +++ b/app/service/chat/gemini_chat_service.py @@ -112,7 +112,7 @@ def _build_payload(model: str, request: GeminiRequest) -> Dict[str, Any]: if model.endswith("-non-thinking"): payload["generationConfig"]["thinkingConfig"] = {"thinkingBudget": 0} if model in settings.THINKING_BUDGET_MAP: - payload["generationConfig"]["thinkingConfig"] = {"thinkingBudget": settings.THINKING_BUDGET_MAP.get(request.model,1000)} + payload["generationConfig"]["thinkingConfig"] = {"thinkingBudget": settings.THINKING_BUDGET_MAP.get(model,1000)} return payload @@ -162,10 +162,6 @@ class GeminiChatService: try: response = await self.api_client.generate_content(payload, model, api_key) - # Assuming success if no exception is raised and response is received - # The actual status code might be within the response structure or headers, - # but api_client doesn't seem to expose it directly here. - # We'll assume 200 for success if no exception. is_success = True status_code = 200 # Assume 200 on success return self.response_handler.handle_response(response, model, stream=False) @@ -184,7 +180,7 @@ class GeminiChatService: await add_error_log( gemini_key=api_key, model_name=model, - error_type="gemini_chat_service", + error_type="gemini-chat-non-stream", error_log=error_log_msg, error_code=status_code, request_msg=payload @@ -210,96 +206,90 @@ class GeminiChatService: retries = 0 max_retries = settings.MAX_RETRIES payload = _build_payload(model, request) - start_time = time.perf_counter() # Record start time before loop - request_datetime = datetime.datetime.now() is_success = False status_code = None - final_api_key = api_key # Store the initial key + final_api_key = api_key - try: - while retries < max_retries: - current_attempt_key = api_key # Key used for this attempt - final_api_key = current_attempt_key # Update final key used - try: - async for line in self.api_client.stream_generate_content( - payload, model, current_attempt_key - ): - # print(line) - if line.startswith("data:"): - line = line[6:] - response_data = self.response_handler.handle_response( - json.loads(line), model, stream=True - ) - text = self._extract_text_from_response(response_data) - # 如果有文本内容，且开启了流式输出优化器，则使用流式输出优化器处理 - if text and settings.STREAM_OPTIMIZER_ENABLED: - # 使用流式输出优化器处理文本输出 - async for ( - optimized_chunk - ) in gemini_optimizer.optimize_stream_output( - text, - lambda t: self._create_char_response(response_data, t), - lambda c: "data: " + json.dumps(c) + "\n\n", - ): - yield optimized_chunk - else: - # 如果没有文本内容（如工具调用等），整块输出 - yield "data: " + json.dumps(response_data) + "\n\n" - logger.info("Streaming completed successfully") - is_success = True - status_code = 200 # Assume 200 on success - break # Exit loop on success - except Exception as e: - retries += 1 - is_success = False # Mark as failed for this attempt - error_log_msg = str(e) - logger.warning( - f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}" - ) - # Parse error code for logging - match = re.search(r"status code (\d+)", error_log_msg) - if match: - status_code = int(match.group(1)) - else: - status_code = 500 # Default if parsing fails - - # Log error to error log table - await add_error_log( - gemini_key=current_attempt_key, # Log key used for this failed attempt - model_name=model, - error_log=error_log_msg, - error_code=status_code, - request_msg=payload - ) - - # Attempt to switch API Key - api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries) - if api_key: - logger.info(f"Switched to new API key: {api_key}") - else: # No more keys or retries exceeded by handle_api_failure logic - logger.error(f"No valid API key available after {retries} retries.") - break # Exit loop if no key available - - if retries >= max_retries: - logger.error( - f"Max retries ({max_retries}) reached for streaming." + while retries < max_retries: + request_datetime = datetime.datetime.now() + start_time = time.perf_counter() + current_attempt_key = api_key + final_api_key = current_attempt_key # Update final key used + try: + async for line in self.api_client.stream_generate_content( + payload, model, current_attempt_key + ): + # print(line) + if line.startswith("data:"): + line = line[6:] + response_data = self.response_handler.handle_response( + json.loads(line), model, stream=True ) - break # Exit loop after max retries - finally: - # Log the final outcome of the streaming request - end_time = time.perf_counter() - latency_ms = int((end_time - start_time) * 1000) - await add_request_log( - model_name=model, - api_key=final_api_key, # Log the last key used - is_success=is_success, # Log the final success status - status_code=status_code, # Log the last known status code - latency_ms=latency_ms, # Log total time including retries - request_time=request_datetime - ) - # If the loop finished due to failure, ensure an exception is raised if not already handled - if not is_success and retries >= max_retries: - # We need to raise an exception here if the loop exited due to max retries failure - # However, the original code structure doesn't explicitly raise here after the loop. - # For now, we just log. Consider raising HTTPException if needed. - pass + text = self._extract_text_from_response(response_data) + # 如果有文本内容，且开启了流式输出优化器，则使用流式输出优化器处理 + if text and settings.STREAM_OPTIMIZER_ENABLED: + # 使用流式输出优化器处理文本输出 + async for ( + optimized_chunk + ) in gemini_optimizer.optimize_stream_output( + text, + lambda t: self._create_char_response(response_data, t), + lambda c: "data: " + json.dumps(c) + "\n\n", + ): + yield optimized_chunk + else: + # 如果没有文本内容（如工具调用等），整块输出 + yield "data: " + json.dumps(response_data) + "\n\n" + logger.info("Streaming completed successfully") + is_success = True + status_code = 200 + break + except Exception as e: + retries += 1 + is_success = False + error_log_msg = str(e) + logger.warning( + f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}" + ) + # Parse error code for logging + match = re.search(r"status code (\d+)", error_log_msg) + if match: + status_code = int(match.group(1)) + else: + status_code = 500 + + # Log error to error log table + await add_error_log( + gemini_key=current_attempt_key, # Log key used for this failed attempt + model_name=model, + error_type="gemini-chat-stream", + error_log=error_log_msg, + error_code=status_code, + request_msg=payload + ) + + # Attempt to switch API Key + api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries) + if api_key: + logger.info(f"Switched to new API key: {api_key}") + else: # No more keys or retries exceeded by handle_api_failure logic + logger.error(f"No valid API key available after {retries} retries.") + break # Exit loop if no key available + + if retries >= max_retries: + logger.error( + f"Max retries ({max_retries}) reached for streaming." + ) + break # Exit loop after max retries + finally: + # Log the final outcome of the streaming request + end_time = time.perf_counter() + latency_ms = int((end_time - start_time) * 1000) + await add_request_log( + model_name=model, + api_key=final_api_key, # Log the last key used + is_success=is_success, # Log the final success status + status_code=status_code, # Log the last known status code + latency_ms=latency_ms, # Log total time including retries + request_time=request_datetime + ) diff --git a/app/service/chat/openai_chat_service.py b/app/service/chat/openai_chat_service.py index 97f713a..c1648c9 100644 --- a/app/service/chat/openai_chat_service.py +++ b/app/service/chat/openai_chat_service.py @@ -223,7 +223,7 @@ class OpenAIChatService: await add_error_log( gemini_key=api_key, # Note: Parameter name is gemini_key in add_error_log model_name=model, - error_type="openai_chat_service", # Indicate service type + error_type="openai-chat-non-stream", error_log=error_log_msg, error_code=status_code, request_msg=payload @@ -247,118 +247,117 @@ class OpenAIChatService: """处理流式聊天完成，添加重试逻辑""" retries = 0 max_retries = settings.MAX_RETRIES - start_time = time.perf_counter() # Record start time before loop - request_datetime = datetime.datetime.now() is_success = False status_code = None - final_api_key = api_key # Store the initial key + final_api_key = api_key - try: - while retries < max_retries: - current_attempt_key = api_key # Key used for this attempt - final_api_key = current_attempt_key # Update final key used - try: - tool_call_flag = False - async for line in self.api_client.stream_generate_content( - payload, model, current_attempt_key - ): - print(line) - if line.startswith("data:"): - chunk = json.loads(line[6:]) - openai_chunk = self.response_handler.handle_response( - chunk, model, stream=True, finish_reason=None - ) - if openai_chunk: - # 提取文本内容 - text = self._extract_text_from_openai_chunk(openai_chunk) - if text and settings.STREAM_OPTIMIZER_ENABLED: - # 使用流式输出优化器处理文本输出 - async for ( - optimized_chunk - ) in openai_optimizer.optimize_stream_output( - text, - lambda t: self._create_char_openai_chunk( - openai_chunk, t - ), - lambda c: f"data: {json.dumps(c)}\n\n", - ): - yield optimized_chunk - else: - # 如果没有文本内容（如工具调用等），整块输出 - if "tool_calls" in json.dumps(openai_chunk): - tool_call_flag = True - yield f"data: {json.dumps(openai_chunk)}\n\n" - if tool_call_flag: - yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='tool_calls'))}\n\n" - else: - yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='stop'))}\n\n" - yield "data: [DONE]\n\n" - logger.info("Streaming completed successfully") - is_success = True - status_code = 200 # Assume 200 on success - break # 成功后退出循环 - except Exception as e: - retries += 1 - is_success = False # Mark as failed for this attempt - error_log_msg = str(e) - logger.warning( - f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}" - ) - # Parse error code for logging - match = re.search(r"status code (\d+)", error_log_msg) - if match: - status_code = int(match.group(1)) - else: - status_code = 500 # Default if parsing fails - - # Log error to error log table - await add_error_log( - gemini_key=current_attempt_key, # Note: Parameter name is gemini_key - model_name=model, - error_type="openai_chat_service", # Indicate service type - error_log=error_log_msg, - error_code=status_code, - request_msg=payload - ) - - # Attempt to switch API Key - # Ensure key_manager is available (might need adjustment if not always passed) - if self.key_manager: - api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries) - if api_key: - logger.info(f"Switched to new API key: {api_key}") - else: - logger.error(f"No valid API key available after {retries} retries.") - break # Exit loop if no key available - else: - logger.error("KeyManager not available for retry logic.") - break # Exit loop if key manager is missing - - if retries >= max_retries: - logger.error( - f"Max retries ({max_retries}) reached for streaming." + while retries < max_retries: + start_time = time.perf_counter() + request_datetime = datetime.datetime.now() + current_attempt_key = api_key + final_api_key = current_attempt_key + try: + tool_call_flag = False + async for line in self.api_client.stream_generate_content( + payload, model, current_attempt_key + ): + if line.startswith("data:"): + chunk = json.loads(line[6:]) + openai_chunk = self.response_handler.handle_response( + chunk, model, stream=True, finish_reason=None ) - break # Exit loop after max retries - finally: - # Log the final outcome of the streaming request - end_time = time.perf_counter() - latency_ms = int((end_time - start_time) * 1000) - await add_request_log( - model_name=model, - api_key=final_api_key, # Log the last key used - is_success=is_success, # Log the final success status - status_code=status_code, # Log the last known status code - latency_ms=latency_ms, # Log total time including retries - request_time=request_datetime - ) - # If the loop finished due to failure, yield error and DONE - if not is_success and retries >= max_retries: - yield f"data: {json.dumps({'error': 'Streaming failed after retries'})}\n\n" - yield "data: [DONE]\n\n" + if openai_chunk: + # 提取文本内容 + text = self._extract_text_from_openai_chunk(openai_chunk) + if text and settings.STREAM_OPTIMIZER_ENABLED: + # 使用流式输出优化器处理文本输出 + async for ( + optimized_chunk + ) in openai_optimizer.optimize_stream_output( + text, + lambda t: self._create_char_openai_chunk( + openai_chunk, t + ), + lambda c: f"data: {json.dumps(c)}\n\n", + ): + yield optimized_chunk + else: + # 如果没有文本内容（如工具调用等），整块输出 + if "tool_calls" in json.dumps(openai_chunk): + tool_call_flag = True + yield f"data: {json.dumps(openai_chunk)}\n\n" + if tool_call_flag: + yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='tool_calls'))}\n\n" + else: + yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='stop'))}\n\n" + yield "data: [DONE]\n\n" + logger.info("Streaming completed successfully") + is_success = True + status_code = 200 # Assume 200 on success + break # 成功后退出循环 + except Exception as e: + retries += 1 + is_success = False + error_log_msg = str(e) + logger.warning( + f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}" + ) + # Parse error code for logging + match = re.search(r"status code (\d+)", error_log_msg) + if match: + status_code = int(match.group(1)) + else: + status_code = 500 # Default if parsing fails + + # Log error to error log table + await add_error_log( + gemini_key=current_attempt_key, + model_name=model, + error_type="openai-chat-stream", + error_log=error_log_msg, + error_code=status_code, + request_msg=payload + ) + + # Attempt to switch API Key + # Ensure key_manager is available (might need adjustment if not always passed) + if self.key_manager: + api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries) + if api_key: + logger.info(f"Switched to new API key: {api_key}") + else: + logger.error(f"No valid API key available after {retries} retries.") + break # Exit loop if no key available + else: + logger.error("KeyManager not available for retry logic.") + break # Exit loop if key manager is missing + + if retries >= max_retries: + logger.error( + f"Max retries ({max_retries}) reached for streaming." + ) + break # Exit loop after max retries + finally: + # Log the final outcome of the streaming request + end_time = time.perf_counter() + latency_ms = int((end_time - start_time) * 1000) + await add_request_log( + model_name=model, + api_key=final_api_key, # Log the last key used + is_success=is_success, # Log the final success status + status_code=status_code, # Log the last known status code + latency_ms=latency_ms, # Log total time including retries + request_time=request_datetime + ) + # If the loop finished due to failure, yield error and DONE + if not is_success and retries >= max_retries: + yield f"data: {json.dumps({'error': 'Streaming failed after retries'})}\n\n" + yield "data: [DONE]\n\n" async def create_image_chat_completion( self, request: ChatRequest, + api_key: str ) -> Union[Dict[str, Any], AsyncGenerator[str, None]]: image_generate_request = ImageGenerationRequest() @@ -368,41 +367,120 @@ class OpenAIChatService: ) if request.stream: - return self._handle_stream_image_completion(request.model, image_res) + return self._handle_stream_image_completion(request.model, image_res, api_key) else: - return self._handle_normal_image_completion(request.model, image_res) + return await self._handle_normal_image_completion(request.model, image_res, api_key) async def _handle_stream_image_completion( - self, model: str, image_data: str + self, model: str, image_data: str, api_key:str ) -> AsyncGenerator[str, None]: - if image_data: - openai_chunk = self.response_handler.handle_image_chat_response( - image_data, model, stream=True, finish_reason=None + logger.info(f"Starting stream image completion for model: {model}") + start_time = time.perf_counter() + request_datetime = datetime.datetime.now() # Although not used for DB log here + is_success = False + status_code = None # Although not used for DB log here + + try: + if image_data: + openai_chunk = self.response_handler.handle_image_chat_response( + image_data, model, stream=True, finish_reason=None + ) + if openai_chunk: + # 提取文本内容 + text = self._extract_text_from_openai_chunk(openai_chunk) + if text: + # 使用流式输出优化器处理文本输出 + async for ( + optimized_chunk + ) in openai_optimizer.optimize_stream_output( + text, + lambda t: self._create_char_openai_chunk(openai_chunk, t), + lambda c: f"data: {json.dumps(c)}\n\n", + ): + yield optimized_chunk + else: + # 如果没有文本内容（如图片URL等），整块输出 + yield f"data: {json.dumps(openai_chunk)}\n\n" + yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='stop'))}\n\n" + logger.info(f"Stream image completion finished successfully for model: {model}") + is_success = True + status_code = 200 + yield "data: [DONE]\n\n" + except Exception as e: + is_success = False + error_log_msg = f"Stream image completion failed for model {model}: {e}" + logger.error(error_log_msg) + status_code = 500 # Default error code + # Call add_error_log using the passed api_key + await add_error_log( + gemini_key=api_key, + model_name=model, + error_type="openai-image-stream", # Specific error type + error_log=error_log_msg, + error_code=status_code, + request_msg={"image_data_truncated": image_data[:1000]} # Log truncated data + ) + yield f"data: {json.dumps({'error': error_log_msg})}\n\n" # Send error to client + yield "data: [DONE]\n\n" # Still need DONE message + # Re-raising might break the stream, decide if needed + finally: + end_time = time.perf_counter() + latency_ms = int((end_time - start_time) * 1000) + logger.info(f"Stream image completion for model {model} took {latency_ms} ms. Success: {is_success}") + # Call add_request_log using the passed api_key + await add_request_log( + model_name=model, + api_key=api_key, + is_success=is_success, + status_code=status_code, + latency_ms=latency_ms, + request_time=request_datetime ) - if openai_chunk: - # 提取文本内容 - text = self._extract_text_from_openai_chunk(openai_chunk) - if text: - # 使用流式输出优化器处理文本输出 - async for ( - optimized_chunk - ) in openai_optimizer.optimize_stream_output( - text, - lambda t: self._create_char_openai_chunk(openai_chunk, t), - lambda c: f"data: {json.dumps(c)}\n\n", - ): - yield optimized_chunk - else: - # 如果没有文本内容（如图片URL等），整块输出 - yield f"data: {json.dumps(openai_chunk)}\n\n" - yield f"data: {json.dumps(self.response_handler.handle_response({}, model, stream=True, finish_reason='stop'))}\n\n" - yield "data: [DONE]\n\n" - logger.info("Image chat streaming completed successfully") - def _handle_normal_image_completion( - self, model: str, image_data: str + async def _handle_normal_image_completion( + self, model: str, image_data: str, api_key: str # Add api_key parameter ) -> Dict[str, Any]: + logger.info(f"Starting normal image completion for model: {model}") + start_time = time.perf_counter() + request_datetime = datetime.datetime.now() # Although not used for DB log here + is_success = False + status_code = None # Although not used for DB log here + result = None - return self.response_handler.handle_image_chat_response( - image_data, model, stream=False, finish_reason="stop" - ) + try: + result = self.response_handler.handle_image_chat_response( + image_data, model, stream=False, finish_reason="stop" + ) + logger.info(f"Normal image completion finished successfully for model: {model}") + is_success = True + status_code = 200 + return result + except Exception as e: + is_success = False + error_log_msg = f"Normal image completion failed for model {model}: {e}" + logger.error(error_log_msg) + status_code = 500 # Default error code + # Call add_error_log using the passed api_key + await add_error_log( + gemini_key=api_key, + model_name=model, + error_type="openai-image-non-stream", # Specific error type + error_log=error_log_msg, + error_code=status_code, + request_msg={"image_data_truncated": image_data[:1000]} # Log truncated data + ) + # Re-raise the exception so the caller knows about the failure + raise e + finally: + end_time = time.perf_counter() + latency_ms = int((end_time - start_time) * 1000) + logger.info(f"Normal image completion for model {model} took {latency_ms} ms. Success: {is_success}") + # Call add_request_log using the passed api_key + await add_request_log( + model_name=model, + api_key=api_key, + is_success=is_success, + status_code=status_code, + latency_ms=latency_ms, + request_time=request_datetime + ) diff --git a/app/service/embedding/embedding_service.py b/app/service/embedding/embedding_service.py index a874ae2..2aac856 100644 --- a/app/service/embedding/embedding_service.py +++ b/app/service/embedding/embedding_service.py @@ -1,9 +1,15 @@ +import datetime +import time +import re # For potential status code parsing from generic errors from typing import List, Union import openai +from openai import APIStatusError # Import specific error type from openai.types import CreateEmbeddingResponse + from app.config.config import settings from app.log.logger import get_embeddings_logger +from app.database.services import add_error_log, add_request_log # Import DB logging functions logger = get_embeddings_logger() @@ -13,11 +19,64 @@ class EmbeddingService: async def create_embedding( self, input_text: Union[str, List[str]], model: str, api_key: str ) -> CreateEmbeddingResponse: - """Create embeddings using OpenAI API""" + """Create embeddings using OpenAI API with database logging""" + start_time = time.perf_counter() + request_datetime = datetime.datetime.now() + is_success = False + status_code = None + response = None + error_log_msg = "" + # Prepare request message for logging (truncate if list or long string) + if isinstance(input_text, list): + request_msg_log = {"input_truncated": [str(item)[:100] + "..." if len(str(item)) > 100 else str(item) for item in input_text[:5]]} + if len(input_text) > 5: + request_msg_log["input_truncated"].append("...") + else: + request_msg_log = {"input_truncated": input_text[:1000] + "..." if len(input_text) > 1000 else input_text} + + try: client = openai.OpenAI(api_key=api_key, base_url=settings.BASE_URL) response = client.embeddings.create(input=input_text, model=model) + is_success = True + status_code = 200 # Assume 200 OK on success return response + except APIStatusError as e: + is_success = False + status_code = e.status_code + error_log_msg = f"OpenAI API error: {e}" + logger.error(f"Error creating embedding (APIStatusError): {error_log_msg}") + raise e # Re-raise the specific error except Exception as e: - logger.error(f"Error creating embedding: {str(e)}") - raise + is_success = False + error_log_msg = f"Generic error: {e}" + logger.error(f"Error creating embedding (Exception): {error_log_msg}") + # Try to parse status code from generic error (less reliable) + match = re.search(r"status code (\d+)", str(e)) + if match: + status_code = int(match.group(1)) + else: + status_code = 500 # Default if parsing fails + raise e # Re-raise the generic error + finally: + end_time = time.perf_counter() + latency_ms = int((end_time - start_time) * 1000) + if not is_success: + # Log error to database if it failed + await add_error_log( + gemini_key=api_key, # Using gemini_key parameter name for consistency + model_name=model, + error_type="openai-embedding", + error_log=error_log_msg, + error_code=status_code, + request_msg=request_msg_log + ) + # Log request outcome to database regardless of success/failure + await add_request_log( + model_name=model, + api_key=api_key, + is_success=is_success, + status_code=status_code, + latency_ms=latency_ms, + request_time=request_datetime + ) diff --git a/app/service/image/image_create_service.py b/app/service/image/image_create_service.py index 92efcd9..a2f1121 100644 --- a/app/service/image/image_create_service.py +++ b/app/service/image/image_create_service.py @@ -17,7 +17,6 @@ logger = get_image_create_logger() class ImageCreateService: def __init__(self, aspect_ratio="1:1"): self.image_model = settings.CREATE_IMAGE_MODEL - self.paid_key = settings.PAID_KEY self.aspect_ratio = aspect_ratio def parse_prompt_parameters(self, prompt: str) -> tuple: @@ -53,7 +52,7 @@ class ImageCreateService: return prompt, n, aspect_ratio def generate_images(self, request: ImageGenerationRequest): - client = genai.Client(api_key=self.paid_key) + client = genai.Client(api_key=settings.PAID_KEY) if request.size == "1024x1024": self.aspect_ratio = "1:1"