diff --git a/src/grok_search/config.py b/src/grok_search/config.py index 90ae3e9..b9f240f 100644 --- a/src/grok_search/config.py +++ b/src/grok_search/config.py @@ -64,6 +64,14 @@ def retry_multiplier(self) -> float: def retry_max_wait(self) -> int: return int(os.getenv("GROK_RETRY_MAX_WAIT", "10")) + @property + def read_timeout(self) -> float: + return float(os.getenv("GROK_READ_TIMEOUT", "300")) + + @property + def connect_timeout(self) -> float: + return float(os.getenv("GROK_CONNECT_TIMEOUT", "6")) + @property def guda_base_url(self) -> str: return os.getenv("GUDA_BASE_URL", self._DEFAULT_GUDA_BASE_URL) diff --git a/src/grok_search/providers/grok.py b/src/grok_search/providers/grok.py index bd2820a..57d0332 100644 --- a/src/grok_search/providers/grok.py +++ b/src/grok_search/providers/grok.py @@ -173,13 +173,14 @@ async def fetch(self, url: str, ctx=None) -> str: async def _parse_streaming_response(self, response, ctx=None) -> str: content = "" - full_body_buffer = [] - + reasoning = "" + full_body_buffer = [] + async for line in response.aiter_lines(): line = line.strip() if not line: continue - + full_body_buffer.append(line) # 兼容 "data: {...}" 和 "data:{...}" 两种 SSE 格式 @@ -193,11 +194,15 @@ async def _parse_streaming_response(self, response, ctx=None) -> str: choices = data.get("choices", []) if choices and len(choices) > 0: delta = choices[0].get("delta", {}) - if "content" in delta: + if "content" in delta and delta["content"]: content += delta["content"] + # reasoning 模型(如 expert)会先输出 reasoning_content, + # 作为无正文时的回退,避免返回空字符串。 + if "reasoning_content" in delta and delta["reasoning_content"]: + reasoning += delta["reasoning_content"] except (json.JSONDecodeError, IndexError): continue - + if not content and full_body_buffer: try: full_text = "".join(full_body_buffer) @@ -205,16 +210,21 @@ async def _parse_streaming_response(self, response, ctx=None) -> str: if "choices" in data and len(data["choices"]) > 0: message = data["choices"][0].get("message", {}) content = message.get("content", "") + if not content: + reasoning = reasoning or message.get("reasoning_content", "") except json.JSONDecodeError: pass - + + if not content and reasoning: + content = reasoning + await log_info(ctx, f"content: {content}", config.debug_enabled) return content async def _execute_stream_with_retry(self, headers: dict, payload: dict, ctx=None) -> str: """执行带重试机制的流式 HTTP 请求""" - timeout = httpx.Timeout(connect=6.0, read=120.0, write=10.0, pool=None) + timeout = httpx.Timeout(connect=config.connect_timeout, read=config.read_timeout, write=10.0, pool=None) async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: async for attempt in AsyncRetrying(