Merge pull request #1208 from CaixyPromise/update/enhance-gemini-with-proxy

Update: 提升Gemini代理配置能力，支持function_call调用与stream流式对话能力

main/xiaozhi-server/core/providers/llm/gemini/gemini.py

@@ -1,140 +1,180 @@
-import google.generativeai as genai
+import os, json, uuid
-from core.utils.util import check_model_key
+from types import SimpleNamespace
-from core.providers.llm.base import LLMProviderBase
+from typing import Any, Dict, List
-from config.logger import setup_logging
-import requests
-import json
 
+import requests
+from google import generativeai as genai
+from google.generativeai import types, GenerationConfig
+
+from core.providers.llm.base import LLMProviderBase
+from core.utils.util import check_model_key
+from config.logger import setup_logging
+from google.generativeai.types import GenerateContentResponse
+from requests import RequestException
+
+log = setup_logging()
 TAG = __name__
-logger = setup_logging()
+
+
+def test_proxy(proxy_url: str, test_url: str) -> bool:
+    try:
+        resp = requests.get(test_url, proxies={"http": proxy_url, "https": proxy_url})
+        return 200 <= resp.status_code < 400
+    except RequestException:
+        return False
+
+
+def setup_proxy_env(http_proxy: str | None, https_proxy: str | None):
+    """
+    分别测试 HTTP 和 HTTPS 代理是否可用，并设置环境变量。
+    如果 HTTPS 代理不可用但 HTTP 可用，会将 HTTPS_PROXY 也指向 HTTP。
+    """
+    test_http_url = "http://www.google.com"
+    test_https_url = "https://www.google.com"
+
+    ok_http = ok_https = False
+
+    if http_proxy:
+        ok_http = test_proxy(http_proxy, test_http_url)
+        if ok_http:
+            os.environ["HTTP_PROXY"] = http_proxy
+            log.bind(tag=TAG).info(f"配置提供的Gemini HTTPS代理连通成功: {http_proxy}")
+        else:
+            log.bind(tag=TAG).warn(f"配置提供的Gemini HTTP代理不可用: {http_proxy}")
+
+    if https_proxy:
+        ok_https = test_proxy(https_proxy, test_https_url)
+        if ok_https:
+            os.environ["HTTPS_PROXY"] = https_proxy
+            log.bind(tag=TAG).info(f"配置提供的Gemini HTTPS代理连通成功: {https_proxy}")
+        else:
+            log.bind(tag=TAG).warning(f"配置提供的Gemini HTTPS代理不可用: {https_proxy}")
+
+    # 如果https_proxy不可用，但http_proxy可用且能走通https，则复用http_proxy作为https_proxy
+    if ok_http and not ok_https:
+        if test_proxy(http_proxy, test_https_url):
+            os.environ["HTTPS_PROXY"] = http_proxy
+            ok_https = True
+            log.bind(tag=TAG).info(f"复用HTTP代理作为HTTPS代理: {http_proxy}")
+
+    if not ok_http and not ok_https:
+        log.bind(tag=TAG).error(f"Gemini 代理设置失败: HTTP 和 HTTPS 代理都不可用，请检查配置")
+        raise RuntimeError("HTTP 和 HTTPS 代理都不可用，请检查配置")
 
 
 class LLMProvider(LLMProviderBase):
-    def __init__(self, config):
+    def __init__(self, cfg: Dict[str, Any]):
-        """初始化Gemini LLM Provider"""
+        self.model_name = cfg.get("model_name", "gemini-2.0-flash")
-        self.model_name = config.get("model_name", "gemini-1.5-pro")
+        self.api_key = cfg["api_key"]
-        self.api_key = config.get("api_key")
+        http_proxy = cfg.get("http_proxy")
-        self.http_proxy = config.get("http_proxy")
+        https_proxy = cfg.get("https_proxy")
-        self.https_proxy = config.get("https_proxy")
-        have_key = check_model_key("LLM", self.api_key)
 
-        if not have_key:
+        if not check_model_key("LLM", self.api_key):
-            return
+            raise ValueError("无效的Gemini API Key，请检查是否配置正确")
 
-        try:
+        if http_proxy or https_proxy:
-            # 初始化Gemini客户端
+            log.bind(tag=TAG).info(f"检测到Gemini代理配置，开始测试代理连通性和设置代理环境...")
-            # 配置代理（如果提供了代理配置）
+            setup_proxy_env(http_proxy, https_proxy)
-            self.proxies = None
+            log.bind(tag=TAG).info(f"Gemini 代理设置成功 - HTTP: {http_proxy}, HTTPS: {https_proxy}")
-            if self.http_proxy is not "" or self.https_proxy is not "":
+        genai.configure(api_key=self.api_key)
+        self.model = genai.GenerativeModel(self.model_name)
 
-                self.proxies = {
+        self.gen_cfg = GenerationConfig(
-                    "http": self.http_proxy,
+            temperature=0.7,
-                    "https": self.https_proxy,
+            top_p=0.9,
-                }
+            top_k=40,
-                logger.bind(tag=TAG).info(f"Gemini set proxys:{self.proxies}")
+            max_output_tokens=2048,
-                # 使用猴子补丁修改 google-generativeai 库的请求会话
+        )
 
-                # 使用 session 对象配置 genai
+    @staticmethod
-
+    def _build_tools(funcs: List[Dict[str, Any]] | None):
-            genai.configure(api_key=self.api_key)
+        if not funcs:
-            self.model = genai.GenerativeModel(self.model_name)
+            return None
-
+        return [types.Tool(function_declarations=[
-            # 设置生成参数
+            types.FunctionDeclaration(
-            self.generation_config = {
+                name=f["function"]["name"],
-                "temperature": 0.7,
+                description=f["function"]["description"],
-                "top_p": 0.9,
+                parameters=f["function"]["parameters"],
-                "top_k": 40,
+            )
-                "max_output_tokens": 2048,
+            for f in funcs
-            }
+        ])]
-            self.chat = None
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Gemini初始化失败: {e}")
-            self.model = None
 
+    # Gemini文档提到，无需维护session-id，直接用dialogue拼接而成
     def response(self, session_id, dialogue):
-        """生成Gemini对话响应"""
+        yield from self._generate(dialogue, None)
-        if not self.model:
-            yield "【Gemini服务未正确初始化】"
-            return
-
-        try:
-            # 处理对话历史
-            chat_history = []
-            for msg in dialogue[:-1]:  # 历史对话
-                role = "model" if msg["role"] == "assistant" else "user"
-                content = msg["content"].strip()
-                if content:
-                    chat_history.append({"role": role, "parts": [{"text": content}]})
-
-            # 获取当前消息
-            current_msg = dialogue[-1]["content"]
-
-            # 构建请求体
-            request_body = {
-                "contents": chat_history
-                + [{"role": "user", "parts": [{"text": current_msg}]}],
-                "generationConfig": self.generation_config,
-            }
-
-            # 构建请求URL
-            url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model_name}:generateContent?key={self.api_key}"
-
-            # 构建请求头
-            headers = {
-                "Content-Type": "application/json",
-            }
-
-            # 发送POST请求,经测试手动 request 无法使用 stream 模式
-            if self.proxies:
-                response = requests.post(
-                    url,
-                    headers=headers,
-                    json=request_body,
-                    stream=False,
-                    proxies=self.proxies,
-                )
-                try:
-                    data = response.json()  # 直接解析JSON
-                    if "candidates" in data and data["candidates"]:
-                        yield data["candidates"][0]["content"]["parts"][0]["text"]
-                    else:
-                        yield "未找到候选回复。"
-                except json.JSONDecodeError as e:
-                    yield f"JSON解码错误：{e}"
-                except Exception as e:
-                    yield f"发生错误：{e}"
-            else:
-                logger.bind(tag=TAG).info(f"Gemini stream mode ")
-                chat = self.model.start_chat(history=chat_history)
-
-                # 发送消息并获取流式响应
-                response = chat.send_message(
-                    current_msg, stream=True, generation_config=self.generation_config
-                )
-                # 处理流式响应
-                for chunk in response:
-                    if hasattr(chunk, "text") and chunk.text:
-                        yield chunk.text
-
-        except Exception as e:
-            error_msg = str(e)
-            logger.bind(tag=TAG).error(f"Gemini响应生成错误: {error_msg}")
-
-            # 针对不同错误返回友好提示
-            if "Rate limit" in error_msg:
-                yield "【Gemini服务请求太频繁,请稍后再试】"
-            elif "Invalid API key" in error_msg:
-                yield "【Gemini API key无效】"
-            else:
-                yield f"【Gemini服务响应异常: {error_msg}】"
-
-        except requests.exceptions.RequestException as e:
-            yield f"请求失败：{e}"
-        except json.JSONDecodeError as e:
-            yield f"JSON解码错误：{e}"
-        except Exception as e:
-            yield f"发生错误：{e}"
 
     def response_with_functions(self, session_id, dialogue, functions=None):
-        logger.bind(tag=TAG).info(f"gemini暂未实现完整的工具调用（function call）")
+        yield from self._generate(dialogue, self._build_tools(functions))
-        return self.response(session_id, dialogue)
+
+    def _generate(self, dialogue, tools):
+        role_map = {"assistant": "model", "user": "user"}
+        contents: list = []
+        # 拼接对话
+        for m in dialogue:
+            r = m["role"]
+
+            if r == "assistant" and "tool_calls" in m:
+                tc = m["tool_calls"][0]
+                contents.append({
+                    "role": "model",
+                    "parts": [{"function_call": {
+                        "name": tc["function"]["name"],
+                        "args": json.loads(tc["function"]["arguments"]),
+                    }}],
+                })
+                continue
+
+            if r == "tool":
+                contents.append({
+                    "role": "model",
+                    "parts": [{"text": str(m.get("content", ""))}],
+                })
+                continue
+
+            contents.append({
+                "role": role_map.get(r, "user"),
+                "parts": [{"text": str(m.get("content", ""))}],
+            })
+
+        stream: GenerateContentResponse = self.model.generate_content(
+            contents=contents,
+            generation_config=self.gen_cfg,
+            tools=tools,
+            stream=True,
+        )
+
+        try:
+            for chunk in stream:
+                cand = chunk.candidates[0]
+                for part in cand.content.parts:
+                    # a) 函数调用-通常是最后一段话才是函数调用
+                    if getattr(part, "function_call", None):
+                        fc = part.function_call
+                        yield None, [SimpleNamespace(
+                            id=uuid.uuid4().hex,
+                            type="function",
+                            function=SimpleNamespace(
+                                name=fc.name,
+                                arguments=json.dumps(dict(fc.args),
+                                                     ensure_ascii=False),
+                            ),
+                        )]
+                        return
+                    # b) 普通文本
+                    if getattr(part, "text", None):
+                        yield part.text if tools is None else (part.text, None)
+
+        finally:
+            if tools is not None:
+                yield None, None  # function‑mode 结束，返回哑包
+
+    # 关闭stream，预留后续打断对话功能的功能方法，官方文档推荐打断对话要关闭上一个流，可以有效减少配额计费和资源占用
+    @staticmethod
+    def _safe_finish_stream(stream: GenerateContentResponse):
+        if hasattr(stream, "resolve"):
+            stream.resolve()  # Gemini SDK version ≥ 0.5.0
+        elif hasattr(stream, "close"):
+            stream.close()  # Gemini SDK version < 0.5.0
+        else:
+            for _ in stream:  # 兜底耗尽
+                pass