refactor: streamline chat and video processing services, enhance validation, and remove unused components

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:<mime>;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):

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",
 ]

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
 

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",

app/services/grok/processors/chat.py

@@ -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"</{tag}>" 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("<think>\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("</think>\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("</think>\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)}>|<{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"]

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}")

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},

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"</{tag}>" 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 <think> 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("<think>\n")
+                        self.think_opened = True
+                    if (not is_thinking) and self.think_opened:
+                        yield self._sse("\n</think>\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("<think>\n")
+                            self.think_opened = True
+                    else:
+                        if self.think_opened:
+                            yield self._sse("\n</think>\n")
+                            self.think_opened = False
+                    yield self._sse(filtered)
+
+            if self.think_opened:
+                yield self._sse("</think>\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)}>|<{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'<grok:render[^>]*card_id="([^"]+)"[^>]*>.*?</grok:render>',
+                            _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",
 ]

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
             )

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",
 ]

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,
             )

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 "

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