♻️ refactor(llm): 重构 LLM 服务架构，引入中间件与组件化适配器 (#2073)

* ♻️ refactor(llm): 重构 LLM 服务架构，引入中间件与组件化适配器

- 【重构】LLM 服务核心架构：
    - 引入中间件管道，统一处理请求生命周期（重试、密钥选择、日志、网络请求）。
    - 适配器重构为组件化设计，分离配置映射、消息转换、响应解析和工具序列化逻辑。
    - 移除 `with_smart_retry` 装饰器，其功能由中间件接管。
    - 移除 `LLMToolExecutor`，工具执行逻辑集成到 `ToolInvoker`。
- 【功能】增强配置系统：
    - `LLMGenerationConfig` 采用组件化结构（Core, Reasoning, Visual, Output, Safety, ToolConfig）。
    - 新增 `GenConfigBuilder` 提供语义化配置构建方式。
    - 新增 `LLMEmbeddingConfig` 用于嵌入专用配置。
    - `CommonOverrides` 迁移并更新至新配置结构。
- 【功能】强化工具系统：
    - 引入 `ToolInvoker` 实现更灵活的工具执行，支持回调与结构化错误。
    - `function_tool` 装饰器支持动态 Pydantic 模型创建和依赖注入 (`ToolParam`, `RunContext`)。
    - 平台原生工具支持 (`GeminiCodeExecution`, `GeminiGoogleSearch`, `GeminiUrlContext`)。
- 【功能】高级生成与嵌入：
    - `generate_structured` 方法支持 In-Context Validation and Repair (IVR) 循环和 AutoCoT (思维链) 包装。
    - 新增 `embed_query` 和 `embed_documents` 便捷嵌入 API。
    - `OpenAIImageAdapter` 支持 OpenAI 兼容的图像生成。
    - `SmartAdapter` 实现模型名称智能路由。
- 【重构】消息与类型系统：
    - `LLMContentPart` 扩展支持更多模态和代码执行相关内容。
    - `LLMMessage` 和 `LLMResponse` 结构更新，支持 `content_parts` 和思维链签名。
    - 统一 `LLMErrorCode` 和用户友好错误消息，提供更详细的网络/代理错误提示。
    - `pyproject.toml` 移除 `bilireq`，新增 `json_repair`。
- 【优化】日志与调试：
    - 引入 `DebugLogOptions`，提供细粒度日志脱敏控制。
    - 增强日志净化器，处理更多敏感数据和长字符串。
- 【清理】删除废弃模块：
    - `zhenxun/services/llm/memory.py`
    - `zhenxun/services/llm/executor.py`
    - `zhenxun/services/llm/config/presets.py`
    - `zhenxun/services/llm/types/content.py`
    - `zhenxun/services/llm/types/enums.py`
    - `zhenxun/services/llm/tools/__init__.py`
    - `zhenxun/services/llm/tools/manager.py`

* 📦️ build(deps): 移除 bilireq 并添加 json_repair 依赖

* 🐛 (llm): 移除图片生成模型能力预检查

* ♻️ refactor(llm.session): 重构记忆系统以分离存储和策略

* 🐛 fix(reload_setting): 重载配置时清除LLM缓存

* ✨ feat(llm): 支持结构化生成函数接收 UniMessage

* ✨ feat(search): 为搜索功能默认启用 Gemini Google Search 工具

* 🚨 auto fix by pre-commit hooks

---------

Co-authored-by: webjoin111 <455457521@qq.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

zhenxun/services/llm/adapters/gemini.py

@@ -2,27 +2,35 @@
 Gemini API 适配器
 """
 
-import base64
 from typing import TYPE_CHECKING, Any
 
 from zhenxun.services.log import logger
 
+from ..config.generation import ResponseFormat
+from ..types import LLMContentPart
 from ..types.exceptions import LLMErrorCode, LLMException
-from ..utils import sanitize_schema_for_llm
+from ..types.models import BasePlatformTool, ToolChoice
 from .base import BaseAdapter, RequestData, ResponseData
+from .components.gemini_components import (
+    GeminiConfigMapper,
+    GeminiMessageConverter,
+    GeminiResponseParser,
+    GeminiToolSerializer,
+)
 
 if TYPE_CHECKING:
-    from ..config.generation import LLMGenerationConfig
+    from ..config.generation import LLMEmbeddingConfig, LLMGenerationConfig
     from ..service import LLMModel
-    from ..types.content import LLMMessage
-    from ..types.enums import EmbeddingTaskType
-    from ..types.models import LLMToolCall
-    from ..types.protocols import ToolExecutable
+    from ..types import LLMMessage
 
 
 class GeminiAdapter(BaseAdapter):
     """Gemini API 适配器"""
 
+    @property
+    def log_sanitization_context(self) -> str:
+        return "gemini_request"
+
     @property
     def api_type(self) -> str:
         return "gemini"
@@ -47,110 +55,75 @@ class GeminiAdapter(BaseAdapter):
         api_key: str,
         messages: list["LLMMessage"],
         config: "LLMGenerationConfig | None" = None,
-        tools: dict[str, "ToolExecutable"] | None = None,
-        tool_choice: str | dict[str, Any] | None = None,
+        tools: list[Any] | None = None,
+        tool_choice: str | dict[str, Any] | ToolChoice | None = None,
     ) -> RequestData:
         """准备高级请求"""
         effective_config = config if config is not None else model._generation_config
 
+        if tools:
+            from ..types.models import GeminiUrlContext
+
+            context_urls: list[str] = []
+            for tool in tools:
+                if isinstance(tool, GeminiUrlContext):
+                    context_urls.extend(tool.urls)
+
+            if context_urls and messages:
+                last_msg = messages[-1]
+                if last_msg.role == "user":
+                    url_text = "\n\n[Context URLs]:\n" + "\n".join(context_urls)
+                    if isinstance(last_msg.content, str):
+                        last_msg.content += url_text
+                    elif isinstance(last_msg.content, list):
+                        last_msg.content.append(LLMContentPart.text_part(url_text))
+
+        has_function_tools = False
+        if tools:
+            has_function_tools = any(hasattr(tool, "get_definition") for tool in tools)
+
+        is_structured = False
+        if effective_config and effective_config.output:
+            if (
+                effective_config.output.response_schema
+                or effective_config.output.response_format == ResponseFormat.JSON
+                or effective_config.output.response_mime_type == "application/json"
+            ):
+                is_structured = True
+
+        if (has_function_tools or is_structured) and effective_config:
+            if effective_config.reasoning is None:
+                from ..config.generation import ReasoningConfig
+
+                effective_config.reasoning = ReasoningConfig()
+
+            if (
+                effective_config.reasoning.budget_tokens is None
+                and effective_config.reasoning.effort is None
+            ):
+                reason_desc = "工具调用" if has_function_tools else "结构化输出"
+                logger.debug(
+                    f"检测到{reason_desc}，自动为模型 {model.model_name} 开启思维链增强"
+                )
+                effective_config.reasoning.budget_tokens = -1
+
         endpoint = self._get_gemini_endpoint(model, effective_config)
         url = self.get_api_url(model, endpoint)
         headers = self.get_base_headers(api_key)
 
-        gemini_contents: list[dict[str, Any]] = []
+        converter = GeminiMessageConverter()
         system_instruction_parts: list[dict[str, Any]] | None = None
-
         for msg in messages:
-            current_parts: list[dict[str, Any]] = []
             if msg.role == "system":
                 if isinstance(msg.content, str):
                     system_instruction_parts = [{"text": msg.content}]
                 elif isinstance(msg.content, list):
                     system_instruction_parts = [
-                        await part.convert_for_api_async("gemini")
-                        for part in msg.content
+                        await converter.convert_part(part) for part in msg.content
                     ]
                 continue
 
-            elif msg.role == "user":
-                if isinstance(msg.content, str):
-                    current_parts.append({"text": msg.content})
-                elif isinstance(msg.content, list):
-                    for part_obj in msg.content:
-                        current_parts.append(
-                            await part_obj.convert_for_api_async("gemini")
-                        )
-                gemini_contents.append({"role": "user", "parts": current_parts})
-
-            elif msg.role == "assistant" or msg.role == "model":
-                if isinstance(msg.content, str) and msg.content:
-                    current_parts.append({"text": msg.content})
-                elif isinstance(msg.content, list):
-                    for part_obj in msg.content:
-                        current_parts.append(
-                            await part_obj.convert_for_api_async("gemini")
-                        )
-
-                if msg.tool_calls:
-                    import json
-
-                    for call in msg.tool_calls:
-                        current_parts.append(
-                            {
-                                "functionCall": {
-                                    "name": call.function.name,
-                                    "args": json.loads(call.function.arguments),
-                                }
-                            }
-                        )
-                if current_parts:
-                    gemini_contents.append({"role": "model", "parts": current_parts})
-
-            elif msg.role == "tool":
-                if not msg.name:
-                    raise ValueError("Gemini 工具消息必须包含 'name' 字段（函数名）。")
-
-                import json
-
-                try:
-                    content_str = (
-                        msg.content
-                        if isinstance(msg.content, str)
-                        else str(msg.content)
-                    )
-                    tool_result_obj = json.loads(content_str)
-                except json.JSONDecodeError:
-                    content_str = (
-                        msg.content
-                        if isinstance(msg.content, str)
-                        else str(msg.content)
-                    )
-                    logger.warning(
-                        f"工具 {msg.name} 的结果不是有效的 JSON: {content_str}. "
-                        f"包装为原始字符串。"
-                    )
-                    tool_result_obj = {"raw_output": content_str}
-
-                if isinstance(tool_result_obj, list):
-                    logger.debug(
-                        f"工具 '{msg.name}' 的返回结果是列表，"
-                        f"正在为Gemini API包装为JSON对象。"
-                    )
-                    final_response_payload = {"result": tool_result_obj}
-                elif not isinstance(tool_result_obj, dict):
-                    final_response_payload = {"result": tool_result_obj}
-                else:
-                    final_response_payload = tool_result_obj
-
-                current_parts.append(
-                    {
-                        "functionResponse": {
-                            "name": msg.name,
-                            "response": final_response_payload,
-                        }
-                    }
-                )
-                gemini_contents.append({"role": "function", "parts": current_parts})
+        gemini_contents = await converter.convert_messages_async(messages)
 
         body: dict[str, Any] = {"contents": gemini_contents}
 
@@ -158,75 +131,78 @@ class GeminiAdapter(BaseAdapter):
             body["systemInstruction"] = {"parts": system_instruction_parts}
 
         all_tools_for_request = []
+        has_user_functions = False
         if tools:
-            import asyncio
+            from ..types.protocols import ToolExecutable
 
-            from zhenxun.utils.pydantic_compat import model_dump
+            function_tools: list[ToolExecutable] = []
+            gemini_tools_dict: dict[str, Any] = {}
 
-            definition_tasks = [
-                executable.get_definition() for executable in tools.values()
-            ]
-            tool_definitions = await asyncio.gather(*definition_tasks)
+            for tool in tools:
+                if isinstance(tool, BasePlatformTool):
+                    declaration = tool.get_tool_declaration()
+                    if declaration:
+                        gemini_tools_dict.update(declaration)
+                elif hasattr(tool, "get_definition"):
+                    function_tools.append(tool)
 
-            function_declarations = []
-            for tool_def in tool_definitions:
-                tool_def.parameters = sanitize_schema_for_llm(
-                    tool_def.parameters, api_type="gemini"
-                )
-                function_declarations.append(model_dump(tool_def))
+            if function_tools:
+                import asyncio
 
-            if function_declarations:
-                all_tools_for_request.append(
-                    {"functionDeclarations": function_declarations}
-                )
+                definition_tasks = [
+                    executable.get_definition() for executable in function_tools
+                ]
+                tool_definitions = await asyncio.gather(*definition_tasks)
 
-        if effective_config:
-            if getattr(effective_config, "enable_grounding", False):
-                has_explicit_gs_tool = any(
-                    "googleSearch" in tool_item for tool_item in all_tools_for_request
-                )
-                if not has_explicit_gs_tool:
-                    all_tools_for_request.append({"googleSearch": {}})
-                    logger.debug("隐式启用 Google Search 工具进行信息来源关联。")
+                serializer = GeminiToolSerializer()
+                function_declarations = serializer.serialize_tools(tool_definitions)
 
-            if getattr(effective_config, "enable_code_execution", False):
-                has_explicit_ce_tool = any(
-                    "codeExecution" in tool_item for tool_item in all_tools_for_request
-                )
-                if not has_explicit_ce_tool:
-                    all_tools_for_request.append({"codeExecution": {}})
-                    logger.debug("隐式启用代码执行工具。")
+                if function_declarations:
+                    gemini_tools_dict["functionDeclarations"] = function_declarations
+                    has_user_functions = True
+
+            if gemini_tools_dict:
+                all_tools_for_request.append(gemini_tools_dict)
 
         if all_tools_for_request:
             body["tools"] = all_tools_for_request
 
-        final_tool_choice = tool_choice
-        if final_tool_choice is None and effective_config:
-            final_tool_choice = getattr(effective_config, "tool_choice", None)
+        tool_config_updates: dict[str, Any] = {}
+        if (
+            effective_config
+            and effective_config.custom_params
+            and "user_location" in effective_config.custom_params
+        ):
+            tool_config_updates["retrievalConfig"] = {
+                "latLng": effective_config.custom_params["user_location"]
+            }
 
-        if final_tool_choice:
-            if isinstance(final_tool_choice, str):
-                mode_upper = final_tool_choice.upper()
-                if mode_upper in ["AUTO", "NONE", "ANY"]:
-                    body["toolConfig"] = {"functionCallingConfig": {"mode": mode_upper}}
-                else:
-                    body["toolConfig"] = self._convert_tool_choice_to_gemini(
-                        final_tool_choice
-                    )
-            else:
-                body["toolConfig"] = self._convert_tool_choice_to_gemini(
-                    final_tool_choice
+        if tool_config_updates:
+            body.setdefault("toolConfig", {}).update(tool_config_updates)
+
+        converted_params: dict[str, Any] = {}
+        if effective_config:
+            converted_params = self.convert_generation_config(effective_config, model)
+
+        if converted_params:
+            if "toolConfig" in converted_params:
+                tool_config_payload = converted_params.pop("toolConfig")
+                fc_config = tool_config_payload.get("functionCallingConfig")
+                should_apply_fc = has_user_functions or (
+                    fc_config and fc_config.get("mode") == "NONE"
                 )
+                if should_apply_fc:
+                    body.setdefault("toolConfig", {}).update(tool_config_payload)
+                elif fc_config and fc_config.get("mode") != "AUTO":
+                    logger.debug(
+                        "Gemini: 忽略针对纯内置工具的 functionCallingConfig (API限制)"
+                    )
 
-        final_generation_config = self._build_gemini_generation_config(
-            model, effective_config
-        )
-        if final_generation_config:
-            body["generationConfig"] = final_generation_config
+            if "safetySettings" in converted_params:
+                body["safetySettings"] = converted_params.pop("safetySettings")
 
-        safety_settings = self._build_safety_settings(effective_config)
-        if safety_settings:
-            body["safetySettings"] = safety_settings
+            if converted_params:
+                body["generationConfig"] = converted_params
 
         return RequestData(url=url, headers=headers, body=body)
 
@@ -242,317 +218,56 @@ class GeminiAdapter(BaseAdapter):
     def _get_gemini_endpoint(
         self, model: "LLMModel", config: "LLMGenerationConfig | None" = None
     ) -> str:
-        """根据配置选择Gemini API端点"""
-        if config:
-            if getattr(config, "enable_code_execution", False):
-                return f"/v1beta/models/{model.model_name}:generateContent"
-
-            if getattr(config, "enable_grounding", False):
-                return f"/v1beta/models/{model.model_name}:generateContent"
-
+        """返回Gemini generateContent 端点"""
         return f"/v1beta/models/{model.model_name}:generateContent"
 
-    def _convert_tool_choice_to_gemini(
-        self, tool_choice_value: str | dict[str, Any]
-    ) -> dict[str, Any]:
-        """转换工具选择策略为Gemini格式"""
-        if isinstance(tool_choice_value, str):
-            mode_upper = tool_choice_value.upper()
-            if mode_upper in ["AUTO", "NONE", "ANY"]:
-                return {"functionCallingConfig": {"mode": mode_upper}}
-            else:
-                logger.warning(
-                    f"不支持的 tool_choice 字符串值: '{tool_choice_value}'。"
-                    f"回退到 AUTO。"
-                )
-                return {"functionCallingConfig": {"mode": "AUTO"}}
-
-        elif isinstance(tool_choice_value, dict):
-            if (
-                tool_choice_value.get("type") == "function"
-                and "function" in tool_choice_value
-            ):
-                func_name = tool_choice_value["function"].get("name")
-                if func_name:
-                    return {
-                        "functionCallingConfig": {
-                            "mode": "ANY",
-                            "allowedFunctionNames": [func_name],
-                        }
-                    }
-                else:
-                    logger.warning(
-                        f"tool_choice dict 中的函数名无效: {tool_choice_value}。"
-                        f"回退到 AUTO。"
-                    )
-                    return {"functionCallingConfig": {"mode": "AUTO"}}
-
-            elif "functionCallingConfig" in tool_choice_value:
-                return {
-                    "functionCallingConfig": tool_choice_value["functionCallingConfig"]
-                }
-
-            else:
-                logger.warning(
-                    f"不支持的 tool_choice dict 值: {tool_choice_value}。回退到 AUTO。"
-                )
-                return {"functionCallingConfig": {"mode": "AUTO"}}
-
-        logger.warning(
-            f"tool_choice 的类型无效: {type(tool_choice_value)}。回退到 AUTO。"
-        )
-        return {"functionCallingConfig": {"mode": "AUTO"}}
-
-    def _build_gemini_generation_config(
-        self, model: "LLMModel", config: "LLMGenerationConfig | None" = None
-    ) -> dict[str, Any]:
-        """构建Gemini生成配置"""
-        effective_config = config if config is not None else model._generation_config
-
-        if not effective_config:
-            return {}
-
-        generation_config = effective_config.to_api_params(
-            api_type="gemini", model_name=model.model_name
-        )
-
-        if generation_config:
-            param_keys = list(generation_config.keys())
-            logger.debug(
-                f"构建Gemini生成配置完成，包含 {len(generation_config)} 个参数: "
-                f"{param_keys}"
-            )
-
-        return generation_config
-
-    def _build_safety_settings(
-        self, config: "LLMGenerationConfig | None" = None
-    ) -> list[dict[str, Any]] | None:
-        """构建安全设置"""
-        if not config:
-            return None
-
-        safety_settings = []
-
-        safety_categories = [
-            "HARM_CATEGORY_HARASSMENT",
-            "HARM_CATEGORY_HATE_SPEECH",
-            "HARM_CATEGORY_SEXUALLY_EXPLICIT",
-            "HARM_CATEGORY_DANGEROUS_CONTENT",
-        ]
-
-        custom_safety_settings = getattr(config, "safety_settings", None)
-        if custom_safety_settings:
-            for category, threshold in custom_safety_settings.items():
-                safety_settings.append({"category": category, "threshold": threshold})
-        else:
-            from ..config.providers import get_gemini_safety_threshold
-
-            threshold = get_gemini_safety_threshold()
-            for category in safety_categories:
-                safety_settings.append({"category": category, "threshold": threshold})
-
-        return safety_settings if safety_settings else None
-
-    def validate_response(self, response_json: dict[str, Any]) -> None:
-        """验证 Gemini API 响应，增加对 promptFeedback 的检查"""
-        super().validate_response(response_json)
-
-        if prompt_feedback := response_json.get("promptFeedback"):
-            if block_reason := prompt_feedback.get("blockReason"):
-                logger.warning(
-                    f"Gemini 内容因 promptFeedback 被安全过滤: {block_reason}"
-                )
-                raise LLMException(
-                    f"内容被安全过滤: {block_reason}",
-                    code=LLMErrorCode.CONTENT_FILTERED,
-                    details={
-                        "block_reason": block_reason,
-                        "safety_ratings": prompt_feedback.get("safetyRatings"),
-                    },
-                )
-
     def parse_response(
         self,
         model: "LLMModel",
         response_json: dict[str, Any],
         is_advanced: bool = False,
-    ) -> ResponseData:
-        """解析API响应"""
-        return self._parse_response(model, response_json, is_advanced)
-
-    def _parse_response(
-        self,
-        model: "LLMModel",
-        response_json: dict[str, Any],
-        is_advanced: bool = False,
     ) -> ResponseData:
         """解析 Gemini API 响应"""
-        _ = is_advanced
-        self.validate_response(response_json)
-
-        try:
-            if "image_generation" in response_json and isinstance(
-                response_json["image_generation"], dict
-            ):
-                candidates_source = response_json["image_generation"]
-            else:
-                candidates_source = response_json
-
-            candidates = candidates_source.get("candidates", [])
-            usage_info = response_json.get("usageMetadata")
-
-            if not candidates:
-                logger.debug("Gemini响应中没有candidates。")
-                return ResponseData(text="", raw_response=response_json)
-
-            candidate = candidates[0]
-
-            if candidate.get("finishReason") in [
-                "RECITATION",
-                "OTHER",
-            ] and not candidate.get("content"):
-                logger.warning(
-                    f"Gemini candidate finished with reason "
-                    f"'{candidate.get('finishReason')}' and no content."
-                )
-                return ResponseData(
-                    text="",
-                    raw_response=response_json,
-                    usage_info=response_json.get("usageMetadata"),
-                )
-
-            content_data = candidate.get("content", {})
-            parts = content_data.get("parts", [])
-
-            text_content = ""
-            images_bytes: list[bytes] = []
-            parsed_tool_calls: list["LLMToolCall"] | None = None
-            thought_summary_parts = []
-            answer_parts = []
-
-            for part in parts:
-                if "text" in part:
-                    answer_parts.append(part["text"])
-                elif "thought" in part:
-                    thought_summary_parts.append(part["thought"])
-                elif "thoughtSummary" in part:
-                    thought_summary_parts.append(part["thoughtSummary"])
-                elif "inlineData" in part:
-                    inline_data = part["inlineData"]
-                    if "data" in inline_data:
-                        images_bytes.append(base64.b64decode(inline_data["data"]))
-
-                elif "functionCall" in part:
-                    if parsed_tool_calls is None:
-                        parsed_tool_calls = []
-                    fc_data = part["functionCall"]
-                    try:
-                        import json
-
-                        from ..types.models import LLMToolCall, LLMToolFunction
-
-                        call_id = f"call_{model.provider_name}_{len(parsed_tool_calls)}"
-                        parsed_tool_calls.append(
-                            LLMToolCall(
-                                id=call_id,
-                                function=LLMToolFunction(
-                                    name=fc_data["name"],
-                                    arguments=json.dumps(fc_data["args"]),
-                                ),
-                            )
-                        )
-                    except KeyError as e:
-                        logger.warning(
-                            f"解析Gemini functionCall时缺少键: {fc_data}, 错误: {e}"
-                        )
-                    except Exception as e:
-                        logger.warning(
-                            f"解析Gemini functionCall时出错: {fc_data}, 错误: {e}"
-                        )
-                elif "codeExecutionResult" in part:
-                    result = part["codeExecutionResult"]
-                    if result.get("outcome") == "OK":
-                        output = result.get("output", "")
-                        answer_parts.append(f"\n[代码执行结果]:\n```\n{output}\n```\n")
-                    else:
-                        answer_parts.append(
-                            f"\n[代码执行失败]: {result.get('outcome', 'UNKNOWN')}\n"
-                        )
-
-            if thought_summary_parts:
-                full_thought_summary = "\n".join(thought_summary_parts).strip()
-                full_answer = "".join(answer_parts).strip()
-
-                formatted_parts = []
-                if full_thought_summary:
-                    formatted_parts.append(f"🤔 **思考过程**\n\n{full_thought_summary}")
-                if full_answer:
-                    separator = "\n\n---\n\n" if full_thought_summary else ""
-                    formatted_parts.append(f"{separator}✅ **回答**\n\n{full_answer}")
-
-                text_content = "".join(formatted_parts)
-            else:
-                text_content = "".join(answer_parts)
-
-            usage_info = response_json.get("usageMetadata")
-
-            grounding_metadata_obj = None
-            if grounding_data := candidate.get("groundingMetadata"):
-                try:
-                    from ..types.models import LLMGroundingMetadata
-
-                    grounding_metadata_obj = LLMGroundingMetadata(**grounding_data)
-                except Exception as e:
-                    logger.warning(f"无法解析Grounding元数据: {grounding_data}, {e}")
-
-            return ResponseData(
-                text=text_content,
-                tool_calls=parsed_tool_calls,
-                images=images_bytes if images_bytes else None,
-                usage_info=usage_info,
-                raw_response=response_json,
-                grounding_metadata=grounding_metadata_obj,
-            )
-
-        except Exception as e:
-            logger.error(f"解析 Gemini 响应失败: {e}", e=e)
-            raise LLMException(
-                f"解析API响应失败: {e}",
-                code=LLMErrorCode.RESPONSE_PARSE_ERROR,
-                cause=e,
-            )
+        _ = model, is_advanced
+        parser = GeminiResponseParser()
+        return parser.parse(response_json)
 
     def prepare_embedding_request(
         self,
         model: "LLMModel",
         api_key: str,
         texts: list[str],
-        task_type: "EmbeddingTaskType | str",
-        **kwargs: Any,
+        config: "LLMEmbeddingConfig",
     ) -> RequestData:
         """准备文本嵌入请求"""
         api_model_name = model.model_name
         if not api_model_name.startswith("models/"):
             api_model_name = f"models/{api_model_name}"
 
-        url = self.get_api_url(model, f"/{api_model_name}:batchEmbedContents")
+        if not model.api_base:
+            raise LLMException(
+                f"模型 {model.model_name} 的 api_base 未设置",
+                code=LLMErrorCode.CONFIGURATION_ERROR,
+            )
+
+        base_url = model.api_base.rstrip("/")
+        url = f"{base_url}/v1beta/{api_model_name}:batchEmbedContents"
         headers = self.get_base_headers(api_key)
 
         requests_payload = []
         for text_content in texts:
+            safe_text = text_content if text_content else " "
             request_item: dict[str, Any] = {
-                "content": {"parts": [{"text": text_content}]},
+                "model": api_model_name,
+                "content": {"parts": [{"text": safe_text}]},
             }
 
-            from ..types.enums import EmbeddingTaskType
-
-            if task_type and task_type != EmbeddingTaskType.RETRIEVAL_DOCUMENT:
-                request_item["task_type"] = str(task_type).upper()
-            if title := kwargs.get("title"):
-                request_item["title"] = title
-            if output_dimensionality := kwargs.get("output_dimensionality"):
-                request_item["output_dimensionality"] = output_dimensionality
+            if config.task_type:
+                request_item["task_type"] = str(config.task_type).upper()
+            if config.title:
+                request_item["title"] = config.title
+            if config.output_dimensionality:
+                request_item["output_dimensionality"] = config.output_dimensionality
 
             requests_payload.append(request_item)
 
@@ -601,3 +316,9 @@ class GeminiAdapter(BaseAdapter):
                     code=LLMErrorCode.RESPONSE_PARSE_ERROR,
                     details=response_json,
                 )
+
+    def convert_generation_config(
+        self, config: "LLMGenerationConfig", model: "LLMModel"
+    ) -> dict[str, Any]:
+        mapper = GeminiConfigMapper()
+        return mapper.map_config(config, model.model_detail, model.capabilities)