diff --git a/app/api/v1/chat.py b/app/api/v1/chat.py index b1a7035..f7b72ec 100644 --- a/app/api/v1/chat.py +++ b/app/api/v1/chat.py @@ -3,106 +3,35 @@ Chat Completions API 路由 """ from typing import Any, Dict, List, Optional, Union +import base64 +import binascii from fastapi import APIRouter from fastapi.responses import StreamingResponse, JSONResponse -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field from app.services.grok.services.chat import ChatService from app.services.grok.services.model import ModelService from app.core.exceptions import ValidationException -router = APIRouter(tags=["Chat"]) - - -VALID_ROLES = ["developer", "system", "user", "assistant", "tool"] -# 角色别名映射 (OpenAI 兼容: function -> tool) -ROLE_ALIASES = {"function": "tool"} -USER_CONTENT_TYPES = ["text", "image_url", "input_audio", "file"] - - class MessageItem(BaseModel): """消息项""" role: str content: Union[str, List[Dict[str, Any]]] - tool_call_id: Optional[str] = None # tool 角色需要的字段 - name: Optional[str] = None # function 角色的函数名 - @field_validator("role") - @classmethod - def validate_role(cls, v): - # 大小写归一化 - v_lower = v.lower() if isinstance(v, str) else v - # 别名映射 - v_normalized = ROLE_ALIASES.get(v_lower, v_lower) - if v_normalized not in VALID_ROLES: - raise ValueError(f"role must be one of {VALID_ROLES}") - return v_normalized + model_config = {"extra": "ignore"} class VideoConfig(BaseModel): """视频生成配置""" - aspect_ratio: Optional[str] = Field( - "3:2", description="视频比例: 3:2, 16:9, 1:1 等" - ) + aspect_ratio: Optional[str] = Field("3:2", description="视频比例: 1280x720(16:9), 720x1280(9:16), 1792x1024(3:2), 1024x1792(2:3), 1024x1024(1:1)") video_length: Optional[int] = Field(6, description="视频时长(秒): 6 / 10 / 15") resolution_name: Optional[str] = Field("480p", description="视频分辨率: 480p, 720p") preset: Optional[str] = Field("custom", description="风格预设: fun, normal, spicy") - @field_validator("aspect_ratio") - @classmethod - def validate_aspect_ratio(cls, v): - allowed = ["2:3", "3:2", "1:1", "9:16", "16:9"] - if v and v not in allowed: - raise ValidationException( - message=f"aspect_ratio must be one of {allowed}", - param="video_config.aspect_ratio", - code="invalid_aspect_ratio", - ) - return v - - @field_validator("video_length") - @classmethod - def validate_video_length(cls, v): - if v is not None: - if v not in (6, 10, 15): - raise ValidationException( - message="video_length must be 6, 10, or 15 seconds", - param="video_config.video_length", - code="invalid_video_length", - ) - return v - - @field_validator("resolution_name") - @classmethod - def validate_resolution(cls, v): - allowed = ["480p", "720p"] - if v and v not in allowed: - raise ValidationException( - message=f"resolution_name must be one of {allowed}", - param="video_config.resolution_name", - code="invalid_resolution", - ) - return v - - @field_validator("preset") - @classmethod - def validate_preset(cls, v): - # 允许为空，默认 custom - if not v: - return "custom" - allowed = ["fun", "normal", "spicy", "custom"] - if v not in allowed: - raise ValidationException( - message=f"preset must be one of {allowed}", - param="video_config.preset", - code="invalid_preset", - ) - return v - class ChatCompletionRequest(BaseModel): """Chat Completions 请求""" @@ -110,36 +39,165 @@ class ChatCompletionRequest(BaseModel): model: str = Field(..., description="模型名称") messages: List[MessageItem] = Field(..., description="消息数组") stream: Optional[bool] = Field(None, description="是否流式输出") - thinking: Optional[str] = Field(None, description="思考模式: enabled/disabled/None") - + reasoning_effort: Optional[str] = Field(None, description="推理强度: none/minimal/low/medium/high/xhigh") + temperature: Optional[float] = Field(0.8, description="采样温度: 0-2") + top_p: Optional[float] = Field(0.95, description="nucleus 采样: 0-1") # 视频生成配置 video_config: Optional[VideoConfig] = Field(None, description="视频生成参数") - - @field_validator("stream", mode="before") - @classmethod - def validate_stream(cls, v): - """确保 stream 参数被正确解析为布尔值""" - if v is None: - return None - if isinstance(v, bool): - return v - if isinstance(v, str): - if v.lower() in ("true", "1", "yes"): - return True - if v.lower() in ("false", "0", "no"): - return False - # 未识别的字符串值抛出错误 - raise ValueError( - f"Invalid stream value '{v}'. Must be a boolean or one of: true, false, 1, 0, yes, no" - ) - # 非布尔非字符串类型抛出错误 - raise ValueError( - f"Invalid stream value type '{type(v).__name__}'. Must be a boolean or string." - ) - model_config = {"extra": "ignore"} +VALID_ROLES = {"developer", "system", "user", "assistant"} +USER_CONTENT_TYPES = {"text", "image_url", "input_audio", "file"} + + +def _validate_media_input(value: str, field_name: str, param: str): + if not isinstance(value, str) or not value.strip(): + raise ValidationException( + message=f"{field_name} cannot be empty", + param=param, + code="empty_media", + ) + value = value.strip() + if value.startswith("data:"): + return + if value.startswith("http://") or value.startswith("https://"): + return + candidate = "".join(value.split()) + if len(candidate) >= 32 and len(candidate) % 4 == 0: + try: + base64.b64decode(candidate, validate=True) + raise ValidationException( + message=f"{field_name} base64 must be provided as a data URI (data:;base64,...)", + param=param, + code="invalid_media", + ) + except binascii.Error: + pass + raise ValidationException( + message=f"{field_name} must be a URL or data URI", + param=param, + code="invalid_media", + ) + + +def _normalize_stream(value: Any) -> Optional[bool]: + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, str): + if value.lower() in ("true", "1", "yes"): + return True + if value.lower() in ("false", "0", "no"): + return False + raise ValidationException( + message="stream must be a boolean", + param="stream", + code="invalid_stream", + ) + + +def _validate_reasoning_effort(value: Any) -> Optional[str]: + allowed = {"none", "minimal", "low", "medium", "high", "xhigh"} + if value is None: + return None + if not isinstance(value, str) or value not in allowed: + raise ValidationException( + message=f"reasoning_effort must be one of {sorted(allowed)}", + param="reasoning_effort", + code="invalid_reasoning_effort", + ) + return value + + +def _validate_temperature(value: Any) -> float: + if value is None: + return 0.8 + try: + val = float(value) + except Exception: + raise ValidationException( + message="temperature must be a float", + param="temperature", + code="invalid_temperature", + ) + if not (0 <= val <= 2): + raise ValidationException( + message="temperature must be between 0 and 2", + param="temperature", + code="invalid_temperature", + ) + return val + + +def _validate_top_p(value: Any) -> float: + if value is None: + return 0.95 + try: + val = float(value) + except Exception: + raise ValidationException( + message="top_p must be a float", + param="top_p", + code="invalid_top_p", + ) + if not (0 <= val <= 1): + raise ValidationException( + message="top_p must be between 0 and 1", + param="top_p", + code="invalid_top_p", + ) + return val + + +def _normalize_video_config(config: Optional[VideoConfig]) -> VideoConfig: + if config is None: + config = VideoConfig() + + ratio_map = { + "1280x720": "16:9", + "720x1280": "9:16", + "1792x1024": "3:2", + "1024x1792": "2:3", + "1024x1024": "1:1", + "16:9": "16:9", + "9:16": "9:16", + "3:2": "3:2", + "2:3": "2:3", + "1:1": "1:1", + } + if config.aspect_ratio is None: + config.aspect_ratio = "3:2" + if config.aspect_ratio not in ratio_map: + raise ValidationException( + message=f"aspect_ratio must be one of {list(ratio_map.keys())}", + param="video_config.aspect_ratio", + code="invalid_aspect_ratio", + ) + config.aspect_ratio = ratio_map[config.aspect_ratio] + + if config.video_length not in (6, 10, 15): + raise ValidationException( + message="video_length must be 6, 10, or 15 seconds", + param="video_config.video_length", + code="invalid_video_length", + ) + if config.resolution_name not in ("480p", "720p"): + raise ValidationException( + message="resolution_name must be one of ['480p', '720p']", + param="video_config.resolution_name", + code="invalid_resolution", + ) + if config.preset not in ("fun", "normal", "spicy", "custom"): + raise ValidationException( + message="preset must be one of ['fun', 'normal', 'spicy', 'custom']", + param="video_config.preset", + code="invalid_preset", + ) + return config + + def validate_request(request: ChatCompletionRequest): """验证请求参数""" # 验证模型 @@ -152,6 +210,12 @@ def validate_request(request: ChatCompletionRequest): # 验证消息 for idx, msg in enumerate(request.messages): + if not isinstance(msg.role, str) or msg.role not in VALID_ROLES: + raise ValidationException( + message=f"role must be one of {sorted(VALID_ROLES)}", + param=f"messages.{idx}.role", + code="invalid_role", + ) content = msg.content # 字符串内容 @@ -174,6 +238,12 @@ def validate_request(request: ChatCompletionRequest): for block_idx, block in enumerate(content): # 检查空对象 + if not isinstance(block, dict): + raise ValidationException( + message="Content block must be an object", + param=f"messages.{idx}.content.{block_idx}", + code="invalid_block", + ) if not block: raise ValidationException( message="Content block cannot be empty", @@ -211,20 +281,13 @@ def validate_request(request: ChatCompletionRequest): param=f"messages.{idx}.content.{block_idx}.type", code="invalid_type", ) - elif msg.role in ("tool", "function"): - # tool/function 角色只支持 text 类型，但内容可以是 JSON 字符串 + else: if block_type != "text": raise ValidationException( message=f"The `{msg.role}` role only supports 'text' type, got '{block_type}'", param=f"messages.{idx}.content.{block_idx}.type", code="invalid_type", ) - elif block_type != "text": - raise ValidationException( - message=f"The `{msg.role}` role only supports 'text' type, got '{block_type}'", - param=f"messages.{idx}.content.{block_idx}.type", - code="invalid_type", - ) # 验证字段是否存在 & 非空 if block_type == "text": @@ -237,14 +300,61 @@ def validate_request(request: ChatCompletionRequest): ) elif block_type == "image_url": image_url = block.get("image_url") - if not image_url or not ( - isinstance(image_url, dict) and image_url.get("url") - ): + if not image_url or not isinstance(image_url, dict): raise ValidationException( message="image_url must have a 'url' field", param=f"messages.{idx}.content.{block_idx}.image_url", code="missing_url", ) + _validate_media_input( + image_url.get("url", ""), + "image_url.url", + f"messages.{idx}.content.{block_idx}.image_url.url", + ) + elif block_type == "input_audio": + audio = block.get("input_audio") + if not audio or not isinstance(audio, dict): + raise ValidationException( + message="input_audio must have a 'data' field", + param=f"messages.{idx}.content.{block_idx}.input_audio", + code="missing_audio", + ) + _validate_media_input( + audio.get("data", ""), + "input_audio.data", + f"messages.{idx}.content.{block_idx}.input_audio.data", + ) + elif block_type == "file": + file_data = block.get("file") + if not file_data or not isinstance(file_data, dict): + raise ValidationException( + message="file must have a 'file_data' field", + param=f"messages.{idx}.content.{block_idx}.file", + code="missing_file", + ) + _validate_media_input( + file_data.get("file_data", ""), + "file.file_data", + f"messages.{idx}.content.{block_idx}.file.file_data", + ) + else: + raise ValidationException( + message="Message content must be a string or array", + param=f"messages.{idx}.content", + code="invalid_content", + ) + + request.stream = _normalize_stream(request.stream) + request.reasoning_effort = _validate_reasoning_effort(request.reasoning_effort) + request.temperature = _validate_temperature(request.temperature) + request.top_p = _validate_top_p(request.top_p) + + model_info = ModelService.get(request.model) + if model_info and model_info.is_video: + request.video_config = _normalize_video_config(request.video_config) + + +router = APIRouter(tags=["Chat"]) @router.post("/chat/completions") @@ -269,7 +379,7 @@ async def chat_completions(request: ChatCompletionRequest): model=request.model, messages=[msg.model_dump() for msg in request.messages], stream=request.stream, - thinking=request.thinking, + reasoning_effort=request.reasoning_effort, aspect_ratio=v_conf.aspect_ratio, video_length=v_conf.video_length, resolution=v_conf.resolution_name, @@ -280,7 +390,9 @@ async def chat_completions(request: ChatCompletionRequest): model=request.model, messages=[msg.model_dump() for msg in request.messages], stream=request.stream, - thinking=request.thinking, + reasoning_effort=request.reasoning_effort, + temperature=request.temperature, + top_p=request.top_p, ) if isinstance(result, dict): diff --git a/app/core/exceptions.py b/app/core/exceptions.py index 9ceff00..4ae9092 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -101,6 +101,14 @@ class UpstreamException(AppException): self.details = details +class StreamIdleTimeoutError(Exception): + """流空闲超时错误""" + + def __init__(self, idle_seconds: float): + self.idle_seconds = idle_seconds + super().__init__(f"Stream idle timeout after {idle_seconds}s") + + # ============= 异常处理器 ============= @@ -219,6 +227,7 @@ __all__ = [ "ValidationException", "AuthenticationException", "UpstreamException", + "StreamIdleTimeoutError", "error_response", "register_exception_handlers", ] diff --git a/app/services/grok/batch_services/nsfw.py b/app/services/grok/batch_services/nsfw.py index e7761e2..6fbdb9c 100644 --- a/app/services/grok/batch_services/nsfw.py +++ b/app/services/grok/batch_services/nsfw.py @@ -51,7 +51,7 @@ class NSFWService: status = err.details["status"] else: status = getattr(err, "status_code", None) - if status in (401, 403): + if status == 401: await mgr.record_fail(token, status, reason) return status or 0 diff --git a/app/services/grok/processors/__init__.py b/app/services/grok/processors/__init__.py index 72fb3d5..3cf7d61 100644 --- a/app/services/grok/processors/__init__.py +++ b/app/services/grok/processors/__init__.py @@ -2,8 +2,6 @@ OpenAI 响应格式处理器 """ -from .base import BaseProcessor, StreamIdleTimeoutError -from .chat import StreamProcessor, CollectProcessor from .video import VideoStreamProcessor, VideoCollectProcessor from .image import ( ImageStreamProcessor, @@ -13,10 +11,6 @@ from .image import ( ) __all__ = [ - "BaseProcessor", - "StreamIdleTimeoutError", - "StreamProcessor", - "CollectProcessor", "VideoStreamProcessor", "VideoCollectProcessor", "ImageStreamProcessor", diff --git a/app/services/grok/processors/chat.py b/app/services/grok/processors/chat.py deleted file mode 100644 index e5e7bd6..0000000 --- a/app/services/grok/processors/chat.py +++ /dev/null @@ -1,348 +0,0 @@ -""" -Chat response processors. -""" - -import asyncio -import uuid -import re -from typing import Any, AsyncGenerator, AsyncIterable - -import orjson -from curl_cffi.requests.errors import RequestsError - -from app.core.config import get_config -from app.core.logger import logger -from app.core.exceptions import UpstreamException -from .base import ( - BaseProcessor, - StreamIdleTimeoutError, - _with_idle_timeout, - _normalize_stream_line, - _collect_image_urls, - _is_http2_stream_error, -) - - -class StreamProcessor(BaseProcessor): - """Stream response processor.""" - - def __init__(self, model: str, token: str = "", think: bool = None): - super().__init__(model, token) - self.response_id: str = None - self.fingerprint: str = "" - self.think_opened: bool = False - self.role_sent: bool = False - self.filter_tags = get_config("chat.filter_tags") - self._tag_buffer: str = "" - self._in_filter_tag: bool = False - - if think is None: - self.show_think = get_config("chat.thinking") - else: - self.show_think = think - - def _filter_token(self, token: str) -> str: - """Filter special tags (supports cross-token tag filtering).""" - if not self.filter_tags: - return token - - result = [] - i = 0 - while i < len(token): - char = token[i] - - if self._in_filter_tag: - self._tag_buffer += char - if char == ">": - if "/>" in self._tag_buffer: - self._in_filter_tag = False - self._tag_buffer = "" - else: - for tag in self.filter_tags: - if f"" in self._tag_buffer: - self._in_filter_tag = False - self._tag_buffer = "" - break - i += 1 - continue - - if char == "<": - remaining = token[i:] - tag_started = False - for tag in self.filter_tags: - if remaining.startswith(f"<{tag}"): - tag_started = True - break - if len(remaining) < len(tag) + 1: - for j in range(1, len(remaining) + 1): - if f"<{tag}".startswith(remaining[:j]): - tag_started = True - break - - if tag_started: - self._in_filter_tag = True - self._tag_buffer = char - i += 1 - continue - - result.append(char) - i += 1 - - return "".join(result) - - def _sse(self, content: str = "", role: str = None, finish: str = None) -> str: - """Build SSE response.""" - delta = {} - if role: - delta["role"] = role - delta["content"] = "" - elif content: - delta["content"] = content - - chunk = { - "id": self.response_id or f"chatcmpl-{uuid.uuid4().hex[:24]}", - "object": "chat.completion.chunk", - "created": self.created, - "model": self.model, - "system_fingerprint": self.fingerprint, - "choices": [ - {"index": 0, "delta": delta, "logprobs": None, "finish_reason": finish} - ], - } - return f"data: {orjson.dumps(chunk).decode()}\n\n" - - async def process( - self, response: AsyncIterable[bytes] - ) -> AsyncGenerator[str, None]: - """Process stream response.""" - idle_timeout = get_config("timeout.stream_idle_timeout") - - try: - async for line in _with_idle_timeout(response, idle_timeout, self.model): - line = _normalize_stream_line(line) - if not line: - continue - try: - data = orjson.loads(line) - except orjson.JSONDecodeError: - continue - - resp = data.get("result", {}).get("response", {}) - - if (llm := resp.get("llmInfo")) and not self.fingerprint: - self.fingerprint = llm.get("modelHash", "") - if rid := resp.get("responseId"): - self.response_id = rid - - if not self.role_sent: - yield self._sse(role="assistant") - self.role_sent = True - - # Image generation progress - if img := resp.get("streamingImageGenerationResponse"): - if self.show_think: - if not self.think_opened: - yield self._sse("\n") - self.think_opened = True - idx = img.get("imageIndex", 0) + 1 - progress = img.get("progress", 0) - yield self._sse( - f"正在生成第{idx}张图片中，当前进度{progress}%\n" - ) - continue - - # modelResponse - if mr := resp.get("modelResponse"): - if self.think_opened and self.show_think: - if msg := mr.get("message"): - yield self._sse(msg + "\n") - yield self._sse("\n") - self.think_opened = False - - # Handle generated images - for url in _collect_image_urls(mr): - parts = url.split("/") - img_id = parts[-2] if len(parts) >= 2 else "image" - dl_service = self._get_dl() - rendered = await dl_service.render_image( - url, self.token, img_id - ) - yield self._sse(f"{rendered}\n") - - if ( - (meta := mr.get("metadata", {})) - .get("llm_info", {}) - .get("modelHash") - ): - self.fingerprint = meta["llm_info"]["modelHash"] - continue - - # Normal token - if (token := resp.get("token")) is not None: - if token: - filtered = self._filter_token(token) - if filtered: - yield self._sse(filtered) - - if self.think_opened: - yield self._sse("\n") - yield self._sse(finish="stop") - yield "data: [DONE]\n\n" - except asyncio.CancelledError: - logger.debug("Stream cancelled by client", extra={"model": self.model}) - except StreamIdleTimeoutError as e: - raise UpstreamException( - message=f"Stream idle timeout after {e.idle_seconds}s", - status_code=504, - details={ - "error": str(e), - "type": "stream_idle_timeout", - "idle_seconds": e.idle_seconds, - }, - ) - except RequestsError as e: - if _is_http2_stream_error(e): - logger.warning(f"HTTP/2 stream error: {e}", extra={"model": self.model}) - raise UpstreamException( - message="Upstream connection closed unexpectedly", - status_code=502, - details={"error": str(e), "type": "http2_stream_error"}, - ) - logger.error(f"Stream request error: {e}", extra={"model": self.model}) - raise UpstreamException( - message=f"Upstream request failed: {e}", - status_code=502, - details={"error": str(e)}, - ) - except Exception as e: - logger.error( - f"Stream processing error: {e}", - extra={"model": self.model, "error_type": type(e).__name__}, - ) - raise - finally: - await self.close() - - -class CollectProcessor(BaseProcessor): - """Non-stream response processor.""" - - def __init__(self, model: str, token: str = ""): - super().__init__(model, token) - self.filter_tags = get_config("chat.filter_tags") - - def _filter_content(self, content: str) -> str: - """Filter special tags in content.""" - if not content or not self.filter_tags: - return content - - result = content - for tag in self.filter_tags: - pattern = rf"<{re.escape(tag)}[^>]*>.*?|<{re.escape(tag)}[^>]*/>" - result = re.sub(pattern, "", result, flags=re.DOTALL) - - return result - - async def process(self, response: AsyncIterable[bytes]) -> dict[str, Any]: - """Process and collect full response.""" - response_id = "" - fingerprint = "" - content = "" - idle_timeout = get_config("timeout.stream_idle_timeout") - - try: - async for line in _with_idle_timeout(response, idle_timeout, self.model): - line = _normalize_stream_line(line) - if not line: - continue - try: - data = orjson.loads(line) - except orjson.JSONDecodeError: - continue - - resp = data.get("result", {}).get("response", {}) - - if (llm := resp.get("llmInfo")) and not fingerprint: - fingerprint = llm.get("modelHash", "") - - if mr := resp.get("modelResponse"): - response_id = mr.get("responseId", "") - content = mr.get("message", "") - - if urls := _collect_image_urls(mr): - content += "\n" - for url in urls: - parts = url.split("/") - img_id = parts[-2] if len(parts) >= 2 else "image" - dl_service = self._get_dl() - rendered = await dl_service.render_image( - url, self.token, img_id - ) - content += f"{rendered}\n" - - if ( - (meta := mr.get("metadata", {})) - .get("llm_info", {}) - .get("modelHash") - ): - fingerprint = meta["llm_info"]["modelHash"] - - except asyncio.CancelledError: - logger.debug("Collect cancelled by client", extra={"model": self.model}) - except StreamIdleTimeoutError as e: - logger.warning(f"Collect idle timeout: {e}", extra={"model": self.model}) - except RequestsError as e: - if _is_http2_stream_error(e): - logger.warning( - f"HTTP/2 stream error in collect: {e}", extra={"model": self.model} - ) - else: - logger.error(f"Collect request error: {e}", extra={"model": self.model}) - except Exception as e: - logger.error( - f"Collect processing error: {e}", - extra={"model": self.model, "error_type": type(e).__name__}, - ) - finally: - await self.close() - - content = self._filter_content(content) - - return { - "id": response_id, - "object": "chat.completion", - "created": self.created, - "model": self.model, - "system_fingerprint": fingerprint, - "choices": [ - { - "index": 0, - "message": { - "role": "assistant", - "content": content, - "refusal": None, - "annotations": [], - }, - "finish_reason": "stop", - } - ], - "usage": { - "prompt_tokens": 0, - "completion_tokens": 0, - "total_tokens": 0, - "prompt_tokens_details": { - "cached_tokens": 0, - "text_tokens": 0, - "audio_tokens": 0, - "image_tokens": 0, - }, - "completion_tokens_details": { - "text_tokens": 0, - "audio_tokens": 0, - "reasoning_tokens": 0, - }, - }, - } - - -__all__ = ["StreamProcessor", "CollectProcessor"] diff --git a/app/services/grok/processors/image.py b/app/services/grok/processors/image.py index f8f1f29..bbfdb3b 100644 --- a/app/services/grok/processors/image.py +++ b/app/services/grok/processors/image.py @@ -15,14 +15,13 @@ from curl_cffi.requests.errors import RequestsError from app.core.config import get_config from app.core.logger import logger from app.core.storage import DATA_DIR -from app.core.exceptions import UpstreamException -from .base import ( +from app.core.exceptions import UpstreamException, StreamIdleTimeoutError +from app.services.grok.utils.process import ( BaseProcessor, - StreamIdleTimeoutError, _with_idle_timeout, - _normalize_stream_line, - _collect_image_urls, - _is_http2_stream_error, + _normalize_line, + _collect_images, + _is_http2_error, ) @@ -57,7 +56,7 @@ class ImageStreamProcessor(BaseProcessor): try: async for line in _with_idle_timeout(response, idle_timeout, self.model): - line = _normalize_stream_line(line) + line = _normalize_line(line) if not line: continue try: @@ -90,7 +89,7 @@ class ImageStreamProcessor(BaseProcessor): # modelResponse if mr := resp.get("modelResponse"): - if urls := _collect_image_urls(mr): + if urls := _collect_images(mr): for url in urls: if self.response_format == "url": processed = await self.process_url(url, "image") @@ -155,7 +154,7 @@ class ImageStreamProcessor(BaseProcessor): }, ) except RequestsError as e: - if _is_http2_stream_error(e): + if _is_http2_error(e): logger.warning(f"HTTP/2 stream error in image: {e}") raise UpstreamException( message="Upstream connection closed unexpectedly", @@ -192,7 +191,7 @@ class ImageCollectProcessor(BaseProcessor): try: async for line in _with_idle_timeout(response, idle_timeout, self.model): - line = _normalize_stream_line(line) + line = _normalize_line(line) if not line: continue try: @@ -203,7 +202,7 @@ class ImageCollectProcessor(BaseProcessor): resp = data.get("result", {}).get("response", {}) if mr := resp.get("modelResponse"): - if urls := _collect_image_urls(mr): + if urls := _collect_images(mr): for url in urls: if self.response_format == "url": processed = await self.process_url(url, "image") @@ -234,7 +233,7 @@ class ImageCollectProcessor(BaseProcessor): except StreamIdleTimeoutError as e: logger.warning(f"Image collect idle timeout: {e}") except RequestsError as e: - if _is_http2_stream_error(e): + if _is_http2_error(e): logger.warning(f"HTTP/2 stream error in image collect: {e}") else: logger.error(f"Image collect request error: {e}") diff --git a/app/services/grok/processors/video.py b/app/services/grok/processors/video.py index 8ba68b3..fb501ca 100644 --- a/app/services/grok/processors/video.py +++ b/app/services/grok/processors/video.py @@ -11,29 +11,25 @@ from curl_cffi.requests.errors import RequestsError from app.core.config import get_config from app.core.logger import logger -from app.core.exceptions import UpstreamException -from .base import ( +from app.core.exceptions import UpstreamException, StreamIdleTimeoutError +from app.services.grok.utils.process import ( BaseProcessor, - StreamIdleTimeoutError, _with_idle_timeout, - _normalize_stream_line, - _is_http2_stream_error, + _normalize_line, + _is_http2_error, ) class VideoStreamProcessor(BaseProcessor): """Video stream response processor.""" - def __init__(self, model: str, token: str = "", think: bool = None): + def __init__(self, model: str, token: str = "", show_think: bool = None): super().__init__(model, token) self.response_id: Optional[str] = None self.think_opened: bool = False self.role_sent: bool = False - if think is None: - self.show_think = get_config("chat.thinking") - else: - self.show_think = think + self.show_think = bool(show_think) def _sse(self, content: str = "", role: str = None, finish: str = None) -> str: """Build SSE response.""" @@ -63,7 +59,7 @@ class VideoStreamProcessor(BaseProcessor): try: async for line in _with_idle_timeout(response, idle_timeout, self.model): - line = _normalize_stream_line(line) + line = _normalize_line(line) if not line: continue try: @@ -127,7 +123,7 @@ class VideoStreamProcessor(BaseProcessor): }, ) except RequestsError as e: - if _is_http2_stream_error(e): + if _is_http2_error(e): logger.warning( f"HTTP/2 stream error in video: {e}", extra={"model": self.model} ) @@ -167,7 +163,7 @@ class VideoCollectProcessor(BaseProcessor): try: async for line in _with_idle_timeout(response, idle_timeout, self.model): - line = _normalize_stream_line(line) + line = _normalize_line(line) if not line: continue try: @@ -199,7 +195,7 @@ class VideoCollectProcessor(BaseProcessor): f"Video collect idle timeout: {e}", extra={"model": self.model} ) except RequestsError as e: - if _is_http2_stream_error(e): + if _is_http2_error(e): logger.warning( f"HTTP/2 stream error in video collect: {e}", extra={"model": self.model}, diff --git a/app/services/grok/services/chat.py b/app/services/grok/services/chat.py index 8fe1e1a..e2a452b 100644 --- a/app/services/grok/services/chat.py +++ b/app/services/grok/services/chat.py @@ -2,10 +2,14 @@ Grok Chat 服务 """ -from typing import Dict, List, Any -from dataclasses import dataclass +import asyncio +import re +import uuid +from typing import Dict, List, Any, AsyncGenerator, AsyncIterable +import orjson from curl_cffi.requests import AsyncSession +from curl_cffi.requests.errors import RequestsError from app.core.logger import logger from app.core.config import get_config @@ -14,39 +18,29 @@ from app.core.exceptions import ( ValidationException, ErrorType, UpstreamException, + StreamIdleTimeoutError, ) from app.services.grok.services.model import ModelService from app.services.grok.utils.upload import UploadService -from app.services.grok.processors import StreamProcessor, CollectProcessor +from app.services.grok.utils import process as proc_base from app.services.reverse.app_chat import AppChatReverse from app.services.grok.utils.stream import wrap_stream_with_usage from app.services.token import get_token_manager, EffortType -@dataclass -class ChatRequest: - """聊天请求数据""" - - model: str - messages: List[Dict[str, Any]] - stream: bool = None - think: bool = None - - class MessageExtractor: """消息内容提取器""" @staticmethod - def extract( - messages: List[Dict[str, Any]], is_video: bool = False - ) -> tuple[str, List[tuple[str, str]]]: - """从 OpenAI 消息格式提取内容，返回 (text, attachments)""" + def extract(messages: List[Dict[str, Any]]) -> tuple[str, List[str], List[str]]: + """从 OpenAI 消息格式提取内容，返回 (text, file_attachments, image_attachments)""" texts = [] - attachments = [] + file_attachments: List[str] = [] + image_attachments: List[str] = [] extracted = [] for msg in messages: - role = msg.get("role", "") + role = msg.get("role", "") or "user" content = msg.get("content", "") parts = [] @@ -63,35 +57,21 @@ class MessageExtractor: elif item_type == "image_url": image_data = item.get("image_url", {}) - url = ( - image_data.get("url", "") - if isinstance(image_data, dict) - else str(image_data) - ) + url = image_data.get("url", "") if url: - attachments.append(("image", url)) + image_attachments.append(url) elif item_type == "input_audio": - if is_video: - raise ValueError("视频模型不支持 input_audio 类型") audio_data = item.get("input_audio", {}) - data = ( - audio_data.get("data", "") - if isinstance(audio_data, dict) - else str(audio_data) - ) + data = audio_data.get("data", "") if data: - attachments.append(("audio", data)) + file_attachments.append(data) elif item_type == "file": - if is_video: - raise ValueError("视频模型不支持 file 类型") file_data = item.get("file", {}) - url = file_data.get("url", "") or file_data.get("data", "") - if isinstance(file_data, str): - url = file_data - if url: - attachments.append(("file", url)) + raw = file_data.get("file_data", "") + if raw: + file_attachments.append(raw) if parts: extracted.append({"role": role, "text": "\n".join(parts)}) @@ -111,36 +91,12 @@ class MessageExtractor: text = item["text"] texts.append(text if i == last_user_index else f"{role}: {text}") - return "\n\n".join(texts), attachments - - -class ChatRequestBuilder: - """请求构造器""" - - @staticmethod - def build_payload( - message: str, - model: str, - mode: str = None, - file_attachments: List[str] = None, - image_attachments: List[str] = None, - ) -> Dict[str, Any]: - """构造请求体""" - return AppChatReverse.build_payload( - message=message, - model=model, - mode=mode, - file_attachments=file_attachments, - image_attachments=image_attachments, - ) + return "\n\n".join(texts), file_attachments, image_attachments class GrokChatService: """Grok API 调用服务""" - def __init__(self): - pass - async def chat( self, token: str, @@ -149,7 +105,6 @@ class GrokChatService: mode: str = None, stream: bool = None, file_attachments: List[str] = None, - image_attachments: List[str] = None, tool_overrides: Dict[str, Any] = None, model_config_override: Dict[str, Any] = None, ): @@ -171,7 +126,6 @@ class GrokChatService: model=model, mode=mode, file_attachments=file_attachments, - image_attachments=image_attachments, tool_overrides=tool_overrides, model_config_override=model_config_override, ) @@ -182,44 +136,58 @@ class GrokChatService: return stream_response - async def chat_openai(self, token: str, request: ChatRequest): + async def chat_openai( + self, + token: str, + model: str, + messages: List[Dict[str, Any]], + stream: bool = None, + reasoning_effort: str | None = None, + temperature: float = 0.8, + top_p: float = 0.95, + ): """OpenAI 兼容接口""" - model_info = ModelService.get(request.model) + model_info = ModelService.get(model) if not model_info: - raise ValidationException(f"Unknown model: {request.model}") + raise ValidationException(f"Unknown model: {model}") grok_model = model_info.grok_model mode = model_info.model_mode - is_video = model_info.is_video - # 提取消息和附件 - try: - message, attachments = MessageExtractor.extract( - request.messages, is_video=is_video - ) - logger.debug( - f"Extracted message length={len(message)}, attachments={len(attachments)}" - ) - except ValueError as e: - raise ValidationException(str(e)) + message, file_attachments, image_attachments = MessageExtractor.extract(messages) + logger.debug( + "Extracted message length=%s, files=%s, images=%s", + len(message), + len(file_attachments), + len(image_attachments), + ) # 上传附件 - file_ids = [] - if attachments: + file_ids: List[str] = [] + image_ids: List[str] = [] + if file_attachments or image_attachments: upload_service = UploadService() try: - for attach_type, attach_data in attachments: + for attach_data in file_attachments: file_id, _ = await upload_service.upload_file(attach_data, token) file_ids.append(file_id) - logger.debug( - f"Attachment uploaded: type={attach_type}, file_id={file_id}" - ) + logger.debug(f"Attachment uploaded: type=file, file_id={file_id}") + for attach_data in image_attachments: + file_id, _ = await upload_service.upload_file(attach_data, token) + image_ids.append(file_id) + logger.debug(f"Attachment uploaded: type=image, file_id={file_id}") finally: await upload_service.close() - stream = ( - request.stream if request.stream is not None else get_config("chat.stream") - ) + all_attachments = file_ids + image_ids + stream = stream if stream is not None else get_config("chat.stream") + + model_config_override = { + "temperature": temperature, + "topP": top_p, + } + if reasoning_effort is not None: + model_config_override["reasoningEffort"] = reasoning_effort response = await self.chat( token, @@ -227,11 +195,11 @@ class GrokChatService: grok_model, mode, stream, - file_attachments=file_ids, - image_attachments=[], + file_attachments=all_attachments, + model_config_override=model_config_override, ) - return response, stream, request.model + return response, stream, model class ChatService: @@ -242,29 +210,29 @@ class ChatService: model: str, messages: List[Dict[str, Any]], stream: bool = None, - thinking: str = None, + reasoning_effort: str | None = None, + temperature: float = 0.8, + top_p: float = 0.95, ): """Chat Completions 入口""" # 获取 token token_mgr = await get_token_manager() await token_mgr.reload_if_stale() - # 解析参数（只需解析一次） - think = {"enabled": True, "disabled": False}.get(thinking) + # 解析参数 + if reasoning_effort is None: + show_think = get_config("chat.thinking") + else: + show_think = reasoning_effort != "none" is_stream = stream if stream is not None else get_config("chat.stream") - # 构造请求（只需构造一次） - chat_request = ChatRequest( - model=model, messages=messages, stream=is_stream, think=think - ) - # 跨 Token 重试循环 tried_tokens = set() max_token_retries = int(get_config("retry.max_retry")) last_error = None for attempt in range(max_token_retries): - # 选择 token（排除已失败的） + # 选择 token token = None for pool_name in ModelService.pool_candidates_for_model(model): token = token_mgr.get_token(pool_name, exclude=tried_tokens) @@ -296,12 +264,20 @@ class ChatService: try: # 请求 Grok service = GrokChatService() - response, _, model_name = await service.chat_openai(token, chat_request) + response, _, model_name = await service.chat_openai( + token, + model, + messages, + stream=is_stream, + reasoning_effort=reasoning_effort, + temperature=temperature, + top_p=top_p, + ) # 处理响应 if is_stream: logger.debug(f"Processing stream response: model={model}") - processor = StreamProcessor(model_name, token, think) + processor = StreamProcessor(model_name, token, show_think) return wrap_stream_with_usage( processor.process(response), token_mgr, token, model ) @@ -349,10 +325,404 @@ class ChatService: ) +class StreamProcessor(proc_base.BaseProcessor): + """Stream response processor.""" + + def __init__(self, model: str, token: str = "", show_think: bool = None): + super().__init__(model, token) + self.response_id: str = None + self.fingerprint: str = "" + self.think_opened: bool = False + self.role_sent: bool = False + self.filter_tags = get_config("chat.filter_tags") + self._tag_buffer: str = "" + self._in_filter_tag: bool = False + + self.show_think = bool(show_think) + + def _filter_token(self, token: str) -> str: + """Filter special tags (supports cross-token tag filtering).""" + if not self.filter_tags: + return token + + result = [] + i = 0 + while i < len(token): + char = token[i] + + if self._in_filter_tag: + self._tag_buffer += char + if char == ">": + if "/>" in self._tag_buffer: + self._in_filter_tag = False + self._tag_buffer = "" + else: + for tag in self.filter_tags: + if f"" in self._tag_buffer: + self._in_filter_tag = False + self._tag_buffer = "" + break + i += 1 + continue + + if char == "<": + remaining = token[i:] + tag_started = False + for tag in self.filter_tags: + if remaining.startswith(f"<{tag}"): + tag_started = True + break + if len(remaining) < len(tag) + 1: + for j in range(1, len(remaining) + 1): + if f"<{tag}".startswith(remaining[:j]): + tag_started = True + break + + if tag_started: + self._in_filter_tag = True + self._tag_buffer = char + i += 1 + continue + + result.append(char) + i += 1 + + return "".join(result) + + def _sse(self, content: str = "", role: str = None, finish: str = None) -> str: + """Build SSE response.""" + delta = {} + if role: + delta["role"] = role + delta["content"] = "" + elif content: + delta["content"] = content + + chunk = { + "id": self.response_id or f"chatcmpl-{uuid.uuid4().hex[:24]}", + "object": "chat.completion.chunk", + "created": self.created, + "model": self.model, + "system_fingerprint": self.fingerprint, + "choices": [ + {"index": 0, "delta": delta, "logprobs": None, "finish_reason": finish} + ], + } + return f"data: {orjson.dumps(chunk).decode()}\n\n" + + async def process(self, response: AsyncIterable[bytes]) -> AsyncGenerator[str, None]: + """Process stream response. + + Args: + response: AsyncIterable[bytes], async iterable of bytes + + Returns: + AsyncGenerator[str, None], async generator of strings + """ + idle_timeout = get_config("timeout.stream_idle_timeout") + + try: + async for line in proc_base._with_idle_timeout( + response, idle_timeout, self.model + ): + line = proc_base._normalize_line(line) + if not line: + continue + try: + data = orjson.loads(line) + except orjson.JSONDecodeError: + continue + + resp = data.get("result", {}).get("response", {}) + is_thinking = bool(resp.get("isThinking")) + # isThinking controls tagging + # when absent, treat as False + + if (llm := resp.get("llmInfo")) and not self.fingerprint: + self.fingerprint = llm.get("modelHash", "") + if rid := resp.get("responseId"): + self.response_id = rid + + if not self.role_sent: + yield self._sse(role="assistant") + self.role_sent = True + + if img := resp.get("streamingImageGenerationResponse"): + if not self.show_think: + continue + if is_thinking and not self.think_opened: + yield self._sse("\n") + self.think_opened = True + if (not is_thinking) and self.think_opened: + yield self._sse("\n\n") + self.think_opened = False + idx = img.get("imageIndex", 0) + 1 + progress = img.get("progress", 0) + yield self._sse( + f"正在生成第{idx}张图片中，当前进度{progress}%\n" + ) + continue + + if mr := resp.get("modelResponse"): + for url in proc_base._collect_images(mr): + parts = url.split("/") + img_id = parts[-2] if len(parts) >= 2 else "image" + dl_service = self._get_dl() + rendered = await dl_service.render_image( + url, self.token, img_id + ) + yield self._sse(f"{rendered}\n") + + if ( + (meta := mr.get("metadata", {})) + .get("llm_info", {}) + .get("modelHash") + ): + self.fingerprint = meta["llm_info"]["modelHash"] + continue + + if card := resp.get("cardAttachment"): + json_data = card.get("jsonData") + if isinstance(json_data, str) and json_data.strip(): + try: + card_data = orjson.loads(json_data) + except orjson.JSONDecodeError: + card_data = None + if isinstance(card_data, dict): + image = card_data.get("image") or {} + original = image.get("original") + title = image.get("title") or "" + if original: + title_safe = title.replace("\n", " ").strip() + if title_safe: + yield self._sse(f"![{title_safe}]({original})\n") + else: + yield self._sse(f"![image]({original})\n") + continue + + if (token := resp.get("token")) is not None: + if not token: + continue + filtered = self._filter_token(token) + if not filtered: + continue + if is_thinking: + if not self.show_think: + continue + if not self.think_opened: + yield self._sse("\n") + self.think_opened = True + else: + if self.think_opened: + yield self._sse("\n\n") + self.think_opened = False + yield self._sse(filtered) + + if self.think_opened: + yield self._sse("\n") + yield self._sse(finish="stop") + yield "data: [DONE]\n\n" + except asyncio.CancelledError: + logger.debug("Stream cancelled by client", extra={"model": self.model}) + except StreamIdleTimeoutError as e: + raise UpstreamException( + message=f"Stream idle timeout after {e.idle_seconds}s", + status_code=504, + details={ + "error": str(e), + "type": "stream_idle_timeout", + "idle_seconds": e.idle_seconds, + }, + ) + except RequestsError as e: + if proc_base._is_http2_error(e): + logger.warning(f"HTTP/2 stream error: {e}", extra={"model": self.model}) + raise UpstreamException( + message="Upstream connection closed unexpectedly", + status_code=502, + details={"error": str(e), "type": "http2_stream_error"}, + ) + logger.error(f"Stream request error: {e}", extra={"model": self.model}) + raise UpstreamException( + message=f"Upstream request failed: {e}", + status_code=502, + details={"error": str(e)}, + ) + except Exception as e: + logger.error( + f"Stream processing error: {e}", + extra={"model": self.model, "error_type": type(e).__name__}, + ) + raise + finally: + await self.close() + + +class CollectProcessor(proc_base.BaseProcessor): + """Non-stream response processor.""" + + def __init__(self, model: str, token: str = ""): + super().__init__(model, token) + self.filter_tags = get_config("chat.filter_tags") + + def _filter_content(self, content: str) -> str: + """Filter special tags in content.""" + if not content or not self.filter_tags: + return content + + result = content + for tag in self.filter_tags: + pattern = rf"<{re.escape(tag)}[^>]*>.*?|<{re.escape(tag)}[^>]*/>" + result = re.sub(pattern, "", result, flags=re.DOTALL) + + return result + + async def process(self, response: AsyncIterable[bytes]) -> dict[str, Any]: + """Process and collect full response.""" + response_id = "" + fingerprint = "" + content = "" + idle_timeout = get_config("timeout.stream_idle_timeout") + + try: + async for line in proc_base._with_idle_timeout( + response, idle_timeout, self.model + ): + line = proc_base._normalize_line(line) + if not line: + continue + try: + data = orjson.loads(line) + except orjson.JSONDecodeError: + continue + + resp = data.get("result", {}).get("response", {}) + + if (llm := resp.get("llmInfo")) and not fingerprint: + fingerprint = llm.get("modelHash", "") + + if mr := resp.get("modelResponse"): + response_id = mr.get("responseId", "") + content = mr.get("message", "") + + card_map: dict[str, tuple[str, str]] = {} + for raw in mr.get("cardAttachmentsJson") or []: + if not isinstance(raw, str) or not raw.strip(): + continue + try: + card_data = orjson.loads(raw) + except orjson.JSONDecodeError: + continue + if not isinstance(card_data, dict): + continue + card_id = card_data.get("id") + image = card_data.get("image") or {} + original = image.get("original") + if not card_id or not original: + continue + title = image.get("title") or "" + card_map[card_id] = (title, original) + + if content and card_map: + def _render_card(match: re.Match) -> str: + card_id = match.group(1) + item = card_map.get(card_id) + if not item: + return "" + title, original = item + title_safe = title.replace("\n", " ").strip() or "image" + prefix = "" + if match.start() > 0: + prev = content[match.start() - 1] + if prev not in ("\n", "\r"): + prefix = "\n" + return f"{prefix}![{title_safe}]({original})" + + content = re.sub( + r']*card_id="([^"]+)"[^>]*>.*?', + _render_card, + content, + flags=re.DOTALL, + ) + + if urls := proc_base._collect_images(mr): + content += "\n" + for url in urls: + parts = url.split("/") + img_id = parts[-2] if len(parts) >= 2 else "image" + dl_service = self._get_dl() + rendered = await dl_service.render_image( + url, self.token, img_id + ) + content += f"{rendered}\n" + + if ( + (meta := mr.get("metadata", {})) + .get("llm_info", {}) + .get("modelHash") + ): + fingerprint = meta["llm_info"]["modelHash"] + + except asyncio.CancelledError: + logger.debug("Collect cancelled by client", extra={"model": self.model}) + except StreamIdleTimeoutError as e: + logger.warning(f"Collect idle timeout: {e}", extra={"model": self.model}) + except RequestsError as e: + if proc_base._is_http2_error(e): + logger.warning( + f"HTTP/2 stream error in collect: {e}", extra={"model": self.model} + ) + else: + logger.error(f"Collect request error: {e}", extra={"model": self.model}) + except Exception as e: + logger.error( + f"Collect processing error: {e}", + extra={"model": self.model, "error_type": type(e).__name__}, + ) + finally: + await self.close() + + content = self._filter_content(content) + + return { + "id": response_id, + "object": "chat.completion", + "created": self.created, + "model": self.model, + "system_fingerprint": fingerprint, + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": content, + "refusal": None, + "annotations": [], + }, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + "prompt_tokens_details": { + "cached_tokens": 0, + "text_tokens": 0, + "audio_tokens": 0, + "image_tokens": 0, + }, + "completion_tokens_details": { + "text_tokens": 0, + "audio_tokens": 0, + "reasoning_tokens": 0, + }, + }, + } + + __all__ = [ "GrokChatService", - "ChatRequest", - "ChatRequestBuilder", "MessageExtractor", "ChatService", ] diff --git a/app/services/grok/services/video.py b/app/services/grok/services/video.py index b848045..32c9f6e 100644 --- a/app/services/grok/services/video.py +++ b/app/services/grok/services/video.py @@ -224,7 +224,7 @@ class VideoService: model: str, messages: list, stream: bool = None, - thinking: str = None, + reasoning_effort: str | None = None, aspect_ratio: str = "3:2", video_length: int = 6, resolution: str = "480p", @@ -256,29 +256,30 @@ class VideoService: if token.startswith("sso="): token = token[4:] - think = {"enabled": True, "disabled": False}.get(thinking) + if reasoning_effort is None: + show_think = get_config("chat.thinking") + else: + show_think = reasoning_effort != "none" is_stream = stream if stream is not None else get_config("chat.stream") # Extract content. from app.services.grok.services.chat import MessageExtractor from app.services.grok.utils.upload import UploadService - try: - prompt, attachments = MessageExtractor.extract(messages, is_video=True) - except ValueError as e: - raise ValidationException(str(e)) + prompt, file_attachments, image_attachments = MessageExtractor.extract( + messages, is_video=True + ) # Handle image attachments. image_url = None - if attachments: + if image_attachments: upload_service = UploadService() try: - for attach_type, attach_data in attachments: - if attach_type == "image": - _, file_uri = await upload_service.upload_file(attach_data, token) - image_url = f"https://assets.grok.com/{file_uri}" - logger.info(f"Image uploaded for video: {image_url}") - break + for attach_data in image_attachments: + _, file_uri = await upload_service.upload_file(attach_data, token) + image_url = f"https://assets.grok.com/{file_uri}" + logger.info(f"Image uploaded for video: {image_url}") + break finally: await upload_service.close() @@ -295,7 +296,7 @@ class VideoService: # Process response. if is_stream: - processor = VideoStreamProcessor(model, token, think) + processor = VideoStreamProcessor(model, token, show_think) return wrap_stream_with_usage( processor.process(response), token_mgr, token, model ) diff --git a/app/services/grok/processors/base.py b/app/services/grok/utils/process.py similarity index 87% rename from app/services/grok/processors/base.py rename to app/services/grok/utils/process.py index cab0631..1224949 100644 --- a/app/services/grok/processors/base.py +++ b/app/services/grok/utils/process.py @@ -8,19 +8,20 @@ from typing import Any, AsyncGenerator, Optional, AsyncIterable, List, TypeVar from app.core.config import get_config from app.core.logger import logger +from app.core.exceptions import StreamIdleTimeoutError from app.services.grok.utils.download import DownloadService T = TypeVar("T") -def _is_http2_stream_error(e: Exception) -> bool: +def _is_http2_error(e: Exception) -> bool: """检查是否为 HTTP/2 流错误""" err_str = str(e).lower() return "http/2" in err_str or "curl: (92)" in err_str or "stream" in err_str -def _normalize_stream_line(line: Any) -> Optional[str]: +def _normalize_line(line: Any) -> Optional[str]: """规范化流式响应行，兼容 SSE data 前缀与空行""" if line is None: return None @@ -38,7 +39,7 @@ def _normalize_stream_line(line: Any) -> Optional[str]: return text -def _collect_image_urls(obj: Any) -> List[str]: +def _collect_images(obj: Any) -> List[str]: """递归收集响应中的图片 URL""" urls: List[str] = [] seen = set() @@ -69,14 +70,6 @@ def _collect_image_urls(obj: Any) -> List[str]: return urls -class StreamIdleTimeoutError(Exception): - """流空闲超时错误""" - - def __init__(self, idle_seconds: float): - self.idle_seconds = idle_seconds - super().__init__(f"Stream idle timeout after {idle_seconds}s") - - async def _with_idle_timeout( iterable: AsyncIterable[T], idle_timeout: float, model: str = "" ) -> AsyncGenerator[T, None]: @@ -138,9 +131,8 @@ class BaseProcessor: __all__ = [ "BaseProcessor", - "StreamIdleTimeoutError", "_with_idle_timeout", - "_normalize_stream_line", - "_collect_image_urls", - "_is_http2_stream_error", + "_normalize_line", + "_collect_images", + "_is_http2_error", ] diff --git a/app/services/reverse/app_chat.py b/app/services/reverse/app_chat.py index 77ce685..6ddeb07 100644 --- a/app/services/reverse/app_chat.py +++ b/app/services/reverse/app_chat.py @@ -25,17 +25,12 @@ class AppChatReverse: model: str, mode: str = None, file_attachments: List[str] = None, - image_attachments: List[str] = None, tool_overrides: Dict[str, Any] = None, model_config_override: Dict[str, Any] = None, ) -> Dict[str, Any]: """Build chat payload for Grok app-chat API.""" - attachments = [] - if file_attachments: - attachments.extend(file_attachments) - if image_attachments: - attachments.extend(image_attachments) + attachments = file_attachments or [] payload = { "deviceEnvInfo": { @@ -86,7 +81,6 @@ class AppChatReverse: model: str, mode: str = None, file_attachments: List[str] = None, - image_attachments: List[str] = None, tool_overrides: Dict[str, Any] = None, model_config_override: Dict[str, Any] = None, ) -> Any: @@ -99,7 +93,6 @@ class AppChatReverse: model: str, the model to use. mode: str, the mode to use. file_attachments: List[str], the file attachments to send. - image_attachments: List[str], the image attachments to send. tool_overrides: Dict[str, Any], the tool overrides to use. model_config_override: Dict[str, Any], the model config override to use. @@ -125,7 +118,6 @@ class AppChatReverse: model=model, mode=mode, file_attachments=file_attachments, - image_attachments=image_attachments, tool_overrides=tool_overrides, model_config_override=model_config_override, ) diff --git a/app/services/token/manager.py b/app/services/token/manager.py index 68bfd84..c543d81 100644 --- a/app/services/token/manager.py +++ b/app/services/token/manager.py @@ -389,7 +389,7 @@ class TokenManager: status = e.details["status"] else: status = getattr(e, "status_code", None) - if status in (401, 403): + if status == 401: await self.record_fail(token_str, status, "rate_limits_auth_failed") logger.warning( f"Token {raw_token[:10]}...: API sync failed, fallback to local ({e})" @@ -424,7 +424,7 @@ class TokenManager: for pool in self.pools.values(): token = pool.get(raw_token) if token: - if status_code in (401, 403): + if status_code == 401: token.record_fail(status_code, reason) logger.warning( f"Token {raw_token[:10]}...: recorded {status_code} failure " diff --git a/app/services/token/models.py b/app/services/token/models.py index 0701ab7..d2853d0 100644 --- a/app/services/token/models.py +++ b/app/services/token/models.py @@ -130,8 +130,8 @@ class TokenInfo(BaseModel): def record_fail(self, status_code: int = 401, reason: str = ""): """记录失败，达到阈值后自动标记为 expired""" - # 401/403 错误计入失败（都表示认证/授权失败） - if status_code not in (401, 403): + # 仅 401 计入失败 + if status_code != 401: return self.fail_count += 1