From 755fa30afaa6895909790f99d0ab6432d983033a Mon Sep 17 00:00:00 2001 From: XianYuDaXian Date: Sat, 28 Feb 2026 22:38:33 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=B9=B6=E5=85=A5=20responses=20?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E5=B9=B6=E5=85=BC=E5=AE=B9=E7=BF=BB=E8=AF=91?= =?UTF-8?q?=E6=8F=92=E4=BB=B6=20provider=20options?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/v1/chat.py | 41 ++ app/api/v1/response.py | 117 ++++ app/services/grok/services/chat.py | 2 +- app/services/grok/services/responses.py | 820 ++++++++++++++++++++++++ app/services/grok/utils/response.py | 144 +++++ app/services/grok/utils/tool_call.py | 351 ++++++++++ main.py | 8 + 7 files changed, 1482 insertions(+), 1 deletion(-) create mode 100644 app/api/v1/response.py create mode 100644 app/services/grok/services/responses.py create mode 100644 app/services/grok/utils/response.py create mode 100644 app/services/grok/utils/tool_call.py diff --git a/app/api/v1/chat.py b/app/api/v1/chat.py index 301619e..bf81ffc 100644 --- a/app/api/v1/chat.py +++ b/app/api/v1/chat.py @@ -62,6 +62,9 @@ class ChatCompletionRequest(BaseModel): video_config: Optional[VideoConfig] = Field(None, description="视频生成参数") # 图片生成配置 image_config: Optional[ImageConfig] = Field(None, description="图片生成参数") + # 兼容 Vercel AI SDK 等 provider-specific 扩展参数 + provider_options: Optional[Dict[str, Any]] = Field(None, description="Provider-specific options") + providerOptions: Optional[Dict[str, Any]] = Field(None, description="Provider-specific options (camelCase)") VALID_ROLES = {"developer", "system", "user", "assistant"} @@ -77,6 +80,35 @@ DEFAULT_VIDEO_PROMPT = ( "Animate this." ) +_ALLOWED_REASONING_EFFORTS = {"none", "minimal", "low", "medium", "high", "xhigh"} + + +def _resolve_reasoning_effort_from_provider_options( + provider_options: Optional[Dict[str, Any]], +) -> Optional[str]: + if not isinstance(provider_options, dict): + return None + + # 优先读取通用层字段 + for key in ("reasoning_effort", "reasoningEffort"): + value = provider_options.get(key) + if isinstance(value, str): + effort = value.strip().lower() + if effort in _ALLOWED_REASONING_EFFORTS: + return effort + + # 再读取各 provider 命名空间（如 xai/openai/anthropic） + for _, conf in provider_options.items(): + if not isinstance(conf, dict): + continue + for key in ("reasoning_effort", "reasoningEffort"): + value = conf.get(key) + if isinstance(value, str): + effort = value.strip().lower() + if effort in _ALLOWED_REASONING_EFFORTS: + return effort + return None + def _validate_media_input(value: str, field_name: str, param: str): """Verify media input is a valid URL or data URI""" @@ -692,6 +724,15 @@ def _build_streaming_response(stream_obj: Any, req: Request, model: str) -> Stre @router.post("/chat/completions") async def chat_completions(request: ChatCompletionRequest, raw_request: Request): """Chat Completions API - 兼容 OpenAI""" + # OpenAI 兼容：未显式传 stream 时默认非流式（false） + if request.stream is None: + request.stream = False + + # 兼容 provider-specific options 中的推理强度配置 + if request.reasoning_effort is None: + provider_opts = request.provider_options or request.providerOptions + request.reasoning_effort = _resolve_reasoning_effort_from_provider_options(provider_opts) + model_info = ModelService.get(request.model) if model_info and model_info.is_video: _ensure_video_default_prompt(request.messages) diff --git a/app/api/v1/response.py b/app/api/v1/response.py new file mode 100644 index 0000000..b0e8e4c --- /dev/null +++ b/app/api/v1/response.py @@ -0,0 +1,117 @@ +""" +Responses API 路由 (OpenAI compatible). +""" + +from typing import Any, Dict, List, Optional, Union + +from fastapi import APIRouter +from fastapi.responses import JSONResponse, StreamingResponse +from pydantic import BaseModel, Field + +from app.core.exceptions import ValidationException +from app.services.grok.services.responses import ResponsesService + + +router = APIRouter(tags=["Responses"]) + + +class ResponseCreateRequest(BaseModel): + model: str = Field(..., description="Model name") + input: Optional[Any] = Field(None, description="Input content") + instructions: Optional[str] = Field(None, description="System instructions") + stream: Optional[bool] = Field(False, description="Stream response") + max_output_tokens: Optional[int] = Field(None, description="Max output tokens") + temperature: Optional[float] = Field(None, description="Sampling temperature") + top_p: Optional[float] = Field(None, description="Nucleus sampling") + tools: Optional[List[Dict[str, Any]]] = Field(None, description="Tool definitions") + tool_choice: Optional[Union[str, Dict[str, Any]]] = Field(None, description="Tool choice") + parallel_tool_calls: Optional[bool] = Field(True, description="Allow parallel tool calls") + reasoning: Optional[Dict[str, Any]] = Field(None, description="Reasoning options") + metadata: Optional[Dict[str, Any]] = Field(None, description="Metadata") + user: Optional[str] = Field(None, description="User identifier") + store: Optional[bool] = Field(None, description="Store response") + previous_response_id: Optional[str] = Field(None, description="Previous response id") + truncation: Optional[str] = Field(None, description="Truncation behavior") + provider_options: Optional[Dict[str, Any]] = Field(None, description="Provider-specific options") + providerOptions: Optional[Dict[str, Any]] = Field(None, description="Provider-specific options (camelCase)") + + class Config: + extra = "allow" + + +@router.post("/responses") +async def create_response(request: ResponseCreateRequest): + if not request.model: + raise ValidationException(message="model is required", param="model", code="invalid_request_error") + + if request.input is None: + raise ValidationException(message="input is required", param="input", code="invalid_request_error") + + reasoning_effort = None + if isinstance(request.reasoning, dict): + reasoning_effort = request.reasoning.get("effort") or request.reasoning.get("reasoning_effort") + if not reasoning_effort: + provider_options = request.provider_options or request.providerOptions + if isinstance(provider_options, dict): + for key in ("reasoning_effort", "reasoningEffort"): + val = provider_options.get(key) + if isinstance(val, str) and val.strip(): + reasoning_effort = val.strip().lower() + break + if not reasoning_effort: + for _, conf in provider_options.items(): + if not isinstance(conf, dict): + continue + for key in ("reasoning_effort", "reasoningEffort"): + val = conf.get(key) + if isinstance(val, str) and val.strip(): + reasoning_effort = val.strip().lower() + break + if reasoning_effort: + break + + try: + result = await ResponsesService.create( + model=request.model, + input_value=request.input, + instructions=request.instructions, + stream=bool(request.stream), + temperature=request.temperature, + top_p=request.top_p, + tools=request.tools, + tool_choice=request.tool_choice, + parallel_tool_calls=request.parallel_tool_calls, + reasoning_effort=reasoning_effort, + max_output_tokens=request.max_output_tokens, + metadata=request.metadata, + user=request.user, + store=request.store, + previous_response_id=request.previous_response_id, + truncation=request.truncation, + ) + + if request.stream: + return StreamingResponse( + result, + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}, + ) + + return JSONResponse(content=result) + except ValidationException: + raise + except Exception as e: + return JSONResponse( + status_code=500, + content={ + "error": { + "message": str(e) or "responses request failed", + "type": "server_error", + "param": None, + "code": "responses_internal_error", + } + }, + ) + + +__all__ = ["router"] diff --git a/app/services/grok/services/chat.py b/app/services/grok/services/chat.py index 302d110..fbe0c7c 100644 --- a/app/services/grok/services/chat.py +++ b/app/services/grok/services/chat.py @@ -321,7 +321,7 @@ class ChatService: # 跨 Token 重试循环 tried_tokens = set() - max_token_retries = int(get_config("retry.max_retry")) + max_token_retries = int(get_config("retry.max_retry", 3) or 3) last_error = None for attempt in range(max_token_retries): diff --git a/app/services/grok/services/responses.py b/app/services/grok/services/responses.py new file mode 100644 index 0000000..01d1818 --- /dev/null +++ b/app/services/grok/services/responses.py @@ -0,0 +1,820 @@ +""" +Responses API bridge service (OpenAI-compatible). +""" + +import time +import uuid +from typing import Any, AsyncGenerator, Dict, List, Optional + +import orjson + +from app.services.grok.services.chat import ChatService +from app.services.grok.utils import process as proc_base + + +_TOOL_OUTPUT_TYPES = { + "tool_output", + "function_call_output", + "tool_call_output", + "input_tool_output", +} + +_BUILTIN_TOOL_TYPES = { + "web_search", + "web_search_2025_08_26", + "file_search", + "code_interpreter", +} + + +def _now_ts() -> int: + return int(time.time()) + + +def _new_response_id() -> str: + return f"resp_{uuid.uuid4().hex[:24]}" + + +def _new_message_id() -> str: + return f"msg_{uuid.uuid4().hex[:24]}" + + +def _new_tool_call_id() -> str: + return f"call_{uuid.uuid4().hex[:24]}" + + +def _new_function_call_id() -> str: + return f"fc_{uuid.uuid4().hex[:24]}" + + +def _normalize_tool_choice(tool_choice: Any) -> Any: + if isinstance(tool_choice, dict): + t_type = tool_choice.get("type") + if t_type and t_type != "function": + return {"type": "function", "function": {"name": t_type}} + return tool_choice + + +def _normalize_tools_for_chat(tools: Optional[List[Dict[str, Any]]]) -> Optional[List[Dict[str, Any]]]: + if not tools: + return None + normalized: List[Dict[str, Any]] = [] + for tool in tools: + if not isinstance(tool, dict): + continue + tool_type = tool.get("type") + if tool_type == "function": + normalized.append(tool) + continue + if tool_type in _BUILTIN_TOOL_TYPES: + if tool_type.startswith("web_search"): + normalized.append( + { + "type": "function", + "function": { + "name": tool_type, + "description": "Search the web for information and return results.", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + }, + } + ) + elif tool_type == "file_search": + normalized.append( + { + "type": "function", + "function": { + "name": tool_type, + "description": "Search provided files for relevant information.", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + }, + } + ) + elif tool_type == "code_interpreter": + normalized.append( + { + "type": "function", + "function": { + "name": tool_type, + "description": "Execute code to solve tasks and return results.", + "parameters": { + "type": "object", + "properties": {"code": {"type": "string"}}, + "required": ["code"], + }, + }, + } + ) + return normalized or None + + +def _content_item_from_input(item: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if not isinstance(item, dict): + return None + item_type = item.get("type") + + if item_type in {"input_text", "text", "output_text"}: + text = item.get("text") or item.get("content") or "" + return {"type": "text", "text": text} + + if item_type in {"input_image", "image", "image_url", "output_image"}: + image_url = item.get("image_url") + url = "" + detail = None + if isinstance(image_url, dict): + url = image_url.get("url") or "" + detail = image_url.get("detail") + elif isinstance(image_url, str): + url = image_url + else: + url = item.get("url") or item.get("image") or "" + + if not url: + return None + image_payload = {"url": url} + if detail: + image_payload["detail"] = detail + return {"type": "image_url", "image_url": image_payload} + + if item_type in {"input_file", "file"}: + file_data = item.get("file_data") + file_id = item.get("file_id") + if not file_data and isinstance(item.get("file"), dict): + file_data = item["file"].get("file_data") + file_id = item["file"].get("file_id") + file_payload: Dict[str, Any] = {} + if file_data: + file_payload["file_data"] = file_data + if file_id: + file_payload["file_id"] = file_id + if not file_payload: + return None + return {"type": "file", "file": file_payload} + + if item_type in {"input_audio", "audio"}: + audio = item.get("audio") or {} + data = audio.get("data") or item.get("data") + if not data: + return None + return {"type": "input_audio", "input_audio": {"data": data}} + + return None + + +def _message_from_item(item: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if not isinstance(item, dict): + return None + + if item.get("type") == "message": + role = item.get("role") or "user" + content = item.get("content") + return {"role": role, "content": _coerce_content(content)} + + if "role" in item and "content" in item: + return {"role": item.get("role") or "user", "content": _coerce_content(item.get("content"))} + + return None + + +def _coerce_content(content: Any) -> Any: + if content is None: + return "" + if isinstance(content, str): + return content + if isinstance(content, dict): + content = [content] + if isinstance(content, list): + blocks: List[Dict[str, Any]] = [] + for item in content: + if isinstance(item, dict) and item.get("type") in {"input_text", "output_text"}: + blocks.append({"type": "text", "text": item.get("text", "")}) + continue + block = _content_item_from_input(item) if isinstance(item, dict) else None + if block: + blocks.append(block) + return blocks if blocks else "" + return str(content) + + +def _coerce_input_to_messages(input_value: Any) -> List[Dict[str, Any]]: + if input_value is None: + return [] + if isinstance(input_value, str): + return [{"role": "user", "content": input_value}] + + if isinstance(input_value, dict): + msg = _message_from_item(input_value) + if msg: + return [msg] + content_item = _content_item_from_input(input_value) + if content_item: + return [{"role": "user", "content": [content_item]}] + return [] + + if not isinstance(input_value, list): + return [{"role": "user", "content": str(input_value)}] + + messages: List[Dict[str, Any]] = [] + pending_blocks: List[Dict[str, Any]] = [] + + def _flush_pending(): + nonlocal pending_blocks + if pending_blocks: + messages.append({"role": "user", "content": pending_blocks}) + pending_blocks = [] + + for item in input_value: + if isinstance(item, dict): + msg = _message_from_item(item) + if msg: + _flush_pending() + messages.append(msg) + continue + + item_type = item.get("type") + if item_type in _TOOL_OUTPUT_TYPES: + _flush_pending() + call_id = ( + item.get("call_id") + or item.get("tool_call_id") + or item.get("id") + or _new_tool_call_id() + ) + output = item.get("output") or item.get("content") or "" + messages.append({"role": "tool", "tool_call_id": call_id, "content": output}) + continue + + block = _content_item_from_input(item) + if block: + pending_blocks.append(block) + continue + + if isinstance(item, str): + pending_blocks.append({"type": "text", "text": item}) + + _flush_pending() + return messages + + +def _build_output_message( + text: str, + *, + message_id: Optional[str] = None, + status: str = "completed", +) -> Dict[str, Any]: + message_id = message_id or _new_message_id() + return { + "id": message_id, + "type": "message", + "role": "assistant", + "status": status, + "content": [ + { + "type": "output_text", + "text": text, + "annotations": [], + } + ], + } + + +def _build_output_tool_call( + tool_call: Dict[str, Any], + *, + item_id: Optional[str] = None, + status: str = "completed", +) -> Dict[str, Any]: + fn = tool_call.get("function") or {} + call_id = tool_call.get("id") or _new_tool_call_id() + item_id = item_id or _new_function_call_id() + return { + "id": item_id, + "type": "function_call", + "status": status, + "call_id": call_id, + "name": fn.get("name"), + "arguments": fn.get("arguments"), + } + + +def _build_response_object( + *, + model: str, + output_text: Optional[str] = None, + tool_calls: Optional[List[Dict[str, Any]]] = None, + response_id: Optional[str] = None, + usage: Optional[Dict[str, Any]] = None, + created_at: Optional[int] = None, + completed_at: Optional[int] = None, + status: str = "completed", + instructions: Optional[str] = None, + max_output_tokens: Optional[int] = None, + parallel_tool_calls: Optional[bool] = None, + previous_response_id: Optional[str] = None, + reasoning_effort: Optional[str] = None, + store: Optional[bool] = None, + temperature: Optional[float] = None, + tool_choice: Optional[Any] = None, + tools: Optional[List[Dict[str, Any]]] = None, + top_p: Optional[float] = None, + truncation: Optional[str] = None, + user: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + response_id = response_id or _new_response_id() + created_at = created_at or _now_ts() + if status == "completed" and completed_at is None: + completed_at = _now_ts() + + output: List[Dict[str, Any]] = [] + if output_text is not None: + output.append(_build_output_message(output_text)) + + if tool_calls: + for call in tool_calls: + output.append(_build_output_tool_call(call)) + + return { + "id": response_id, + "object": "response", + "created_at": created_at, + "completed_at": completed_at, + "status": status, + "error": None, + "incomplete_details": None, + "instructions": instructions, + "max_output_tokens": max_output_tokens, + "model": model, + "output": output, + "parallel_tool_calls": True if parallel_tool_calls is None else parallel_tool_calls, + "previous_response_id": previous_response_id, + "reasoning": {"effort": reasoning_effort, "summary": None}, + "store": True if store is None else store, + "temperature": 1.0 if temperature is None else temperature, + "text": {"format": {"type": "text"}}, + "tool_choice": tool_choice or "auto", + "tools": tools or [], + "top_p": 1.0 if top_p is None else top_p, + "truncation": truncation or "disabled", + "usage": usage, + "user": user, + "metadata": metadata or {}, + } + + +class ResponseStreamAdapter: + def __init__( + self, + *, + model: str, + response_id: str, + created_at: int, + instructions: Optional[str], + max_output_tokens: Optional[int], + parallel_tool_calls: Optional[bool], + previous_response_id: Optional[str], + reasoning_effort: Optional[str], + store: Optional[bool], + temperature: Optional[float], + tool_choice: Optional[Any], + tools: Optional[List[Dict[str, Any]]], + top_p: Optional[float], + truncation: Optional[str], + user: Optional[str], + metadata: Optional[Dict[str, Any]], + ): + self.model = model + self.response_id = response_id + self.created_at = created_at + self.instructions = instructions + self.max_output_tokens = max_output_tokens + self.parallel_tool_calls = parallel_tool_calls + self.previous_response_id = previous_response_id + self.reasoning_effort = reasoning_effort + self.store = store + self.temperature = temperature + self.tool_choice = tool_choice + self.tools = tools + self.top_p = top_p + self.truncation = truncation + self.user = user + self.metadata = metadata + + self.output_text_parts: List[str] = [] + self.tool_calls_by_index: Dict[int, Dict[str, Any]] = {} + self.tool_items: Dict[int, Dict[str, Any]] = {} + self.next_output_index = 0 + self.content_index = 0 + self.message_id = _new_message_id() + self.message_started = False + self.message_output_index: Optional[int] = None + + def _event(self, event_type: str, payload: Dict[str, Any]) -> str: + return f"event: {event_type}\ndata: {orjson.dumps(payload).decode()}\n\n" + + def _response_payload(self, *, status: str, output_text: Optional[str], usage: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + tool_calls = None + if status == "completed" and self.tool_calls_by_index: + tool_calls = [ + self.tool_calls_by_index[idx] + for idx in sorted(self.tool_calls_by_index.keys()) + ] + return _build_response_object( + model=self.model, + output_text=output_text, + tool_calls=tool_calls, + response_id=self.response_id, + usage=usage, + created_at=self.created_at, + status=status, + instructions=self.instructions, + max_output_tokens=self.max_output_tokens, + parallel_tool_calls=self.parallel_tool_calls, + previous_response_id=self.previous_response_id, + reasoning_effort=self.reasoning_effort, + store=self.store, + temperature=self.temperature, + tool_choice=self.tool_choice, + tools=self.tools, + top_p=self.top_p, + truncation=self.truncation, + user=self.user, + metadata=self.metadata, + ) + + def _alloc_output_index(self) -> int: + idx = self.next_output_index + self.next_output_index += 1 + return idx + + def created_event(self) -> str: + payload = { + "type": "response.created", + "response": self._response_payload(status="in_progress", output_text=None, usage=None), + } + return self._event("response.created", payload) + + def in_progress_event(self) -> str: + payload = { + "type": "response.in_progress", + "response": self._response_payload(status="in_progress", output_text=None, usage=None), + } + return self._event("response.in_progress", payload) + + def ensure_message_started(self) -> List[str]: + if self.message_started: + return [] + self.message_started = True + self.message_output_index = self._alloc_output_index() + item = _build_output_message("", message_id=self.message_id, status="in_progress") + item["content"] = [] + events = [ + self._event( + "response.output_item.added", + { + "type": "response.output_item.added", + "response_id": self.response_id, + "output_index": self.message_output_index, + "item": item, + }, + ), + self._event( + "response.content_part.added", + { + "type": "response.content_part.added", + "response_id": self.response_id, + "item_id": self.message_id, + "output_index": self.message_output_index, + "content_index": self.content_index, + "part": {"type": "output_text", "text": "", "annotations": []}, + }, + ), + ] + return events + + def output_delta_event(self, delta: str) -> str: + return self._event( + "response.output_text.delta", + { + "type": "response.output_text.delta", + "response_id": self.response_id, + "item_id": self.message_id, + "output_index": self.message_output_index, + "content_index": self.content_index, + "delta": delta, + }, + ) + + def output_done_events(self, text: str) -> List[str]: + if self.message_output_index is None: + return [] + return [ + self._event( + "response.output_text.done", + { + "type": "response.output_text.done", + "response_id": self.response_id, + "item_id": self.message_id, + "output_index": self.message_output_index, + "content_index": self.content_index, + "text": text, + }, + ), + self._event( + "response.content_part.done", + { + "type": "response.content_part.done", + "response_id": self.response_id, + "item_id": self.message_id, + "output_index": self.message_output_index, + "content_index": self.content_index, + "part": {"type": "output_text", "text": text, "annotations": []}, + }, + ), + self._event( + "response.output_item.done", + { + "type": "response.output_item.done", + "response_id": self.response_id, + "output_index": self.message_output_index, + "item": _build_output_message( + text, message_id=self.message_id, status="completed" + ), + }, + ), + ] + + def ensure_tool_item(self, tool_index: int, call_id: str, name: Optional[str]) -> List[str]: + if tool_index in self.tool_items: + item = self.tool_items[tool_index] + if name and not item.get("name"): + item["name"] = name + return [] + output_index = self._alloc_output_index() + item_id = _new_function_call_id() + self.tool_items[tool_index] = { + "item_id": item_id, + "output_index": output_index, + "call_id": call_id, + "name": name, + "arguments": "", + } + tool_item = _build_output_tool_call( + {"id": call_id, "function": {"name": name, "arguments": ""}}, + item_id=item_id, + status="in_progress", + ) + return [ + self._event( + "response.output_item.added", + { + "type": "response.output_item.added", + "response_id": self.response_id, + "output_index": output_index, + "item": tool_item, + }, + ) + ] + + def tool_arguments_delta_event(self, tool_index: int, delta: str) -> Optional[str]: + if not delta: + return None + item = self.tool_items.get(tool_index) + if not item: + return None + item["arguments"] += delta + return self._event( + "response.function_call_arguments.delta", + { + "type": "response.function_call_arguments.delta", + "response_id": self.response_id, + "item_id": item["item_id"], + "output_index": item["output_index"], + "delta": delta, + }, + ) + + def tool_arguments_done_events(self) -> List[str]: + events: List[str] = [] + for tool_index, item in sorted( + self.tool_items.items(), key=lambda kv: kv[1]["output_index"] + ): + events.append( + self._event( + "response.function_call_arguments.done", + { + "type": "response.function_call_arguments.done", + "response_id": self.response_id, + "item_id": item["item_id"], + "output_index": item["output_index"], + "arguments": item["arguments"], + }, + ) + ) + tool_item = _build_output_tool_call( + { + "id": item["call_id"], + "function": {"name": item.get("name"), "arguments": item["arguments"]}, + }, + item_id=item["item_id"], + status="completed", + ) + events.append( + self._event( + "response.output_item.done", + { + "type": "response.output_item.done", + "response_id": self.response_id, + "output_index": item["output_index"], + "item": tool_item, + }, + ) + ) + return events + + def record_tool_call(self, tool_index: int, call_id: str, name: Optional[str], arguments_delta: str) -> None: + tool_call = self.tool_calls_by_index.get(tool_index) + if not tool_call: + tool_call = { + "id": call_id or _new_tool_call_id(), + "type": "function", + "function": {"name": name, "arguments": ""}, + } + self.tool_calls_by_index[tool_index] = tool_call + if name and not tool_call["function"].get("name"): + tool_call["function"]["name"] = name + if arguments_delta: + tool_call["function"]["arguments"] += arguments_delta + + def completed_event(self, usage: Optional[Dict[str, Any]] = None) -> str: + response = self._response_payload( + status="completed", + output_text="".join(self.output_text_parts) if self.message_started else None, + usage=usage + or {"total_tokens": 0, "input_tokens": 0, "output_tokens": 0}, + ) + payload = {"type": "response.completed", "response": response} + return self._event("response.completed", payload) + + +class ResponsesService: + @staticmethod + async def create( + *, + model: str, + input_value: Any, + instructions: Optional[str] = None, + stream: bool = False, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + tools: Optional[List[Dict[str, Any]]] = None, + tool_choice: Any = None, + parallel_tool_calls: Optional[bool] = None, + reasoning_effort: Optional[str] = None, + max_output_tokens: Optional[int] = None, + metadata: Optional[Dict[str, Any]] = None, + user: Optional[str] = None, + store: Optional[bool] = None, + previous_response_id: Optional[str] = None, + truncation: Optional[str] = None, + ) -> Any: + messages = _coerce_input_to_messages(input_value) + if instructions: + messages = [{"role": "system", "content": instructions}] + messages + + if not messages: + raise ValueError("input is required") + + normalized_tools = _normalize_tools_for_chat(tools) + normalized_tool_choice = _normalize_tool_choice(tool_choice) + + chat_kwargs: Dict[str, Any] = { + "model": model, + "messages": messages, + "stream": stream, + "reasoning_effort": reasoning_effort, + } + if temperature is not None: + chat_kwargs["temperature"] = temperature + if top_p is not None: + chat_kwargs["top_p"] = top_p + # 当前分支 ChatService.completions 尚未开放 tools/tool_choice/parallel_tool_calls 参数， + # 此处先做兼容降级：Responses 接口可用，工具字段保留在响应对象中但不透传到底层。 + _ = normalized_tools, normalized_tool_choice, parallel_tool_calls + + result = await ChatService.completions(**chat_kwargs) + + if not stream: + if not isinstance(result, dict): + raise ValueError("Unexpected stream response for non-stream request") + choice = (result.get("choices") or [{}])[0] + message = choice.get("message") or {} + content = message.get("content") or "" + tool_calls = message.get("tool_calls") + return _build_response_object( + model=model, + output_text=content, + tool_calls=tool_calls, + usage=result.get("usage") + or {"total_tokens": 0, "input_tokens": 0, "output_tokens": 0}, + status="completed", + instructions=instructions, + max_output_tokens=max_output_tokens, + parallel_tool_calls=parallel_tool_calls, + previous_response_id=previous_response_id, + reasoning_effort=reasoning_effort, + store=store, + temperature=temperature, + tool_choice=tool_choice, + tools=tools, + top_p=top_p, + truncation=truncation, + user=user, + metadata=metadata, + ) + + if not hasattr(result, "__aiter__"): + raise ValueError("Unexpected non-stream response for stream request") + + created_at = _now_ts() + response_id = _new_response_id() + adapter = ResponseStreamAdapter( + model=model, + response_id=response_id, + created_at=created_at, + instructions=instructions, + max_output_tokens=max_output_tokens, + parallel_tool_calls=parallel_tool_calls, + previous_response_id=previous_response_id, + reasoning_effort=reasoning_effort, + store=store, + temperature=temperature, + tool_choice=tool_choice, + tools=tools, + top_p=top_p, + truncation=truncation, + user=user, + metadata=metadata, + ) + + async def _stream() -> AsyncGenerator[str, None]: + yield adapter.created_event() + yield adapter.in_progress_event() + async for chunk in result: + line = proc_base._normalize_line(chunk) + if not line: + continue + try: + data = orjson.loads(line) + except orjson.JSONDecodeError: + continue + + if data.get("object") == "chat.completion.chunk": + delta = (data.get("choices") or [{}])[0].get("delta") or {} + if "content" in delta and delta["content"]: + for event in adapter.ensure_message_started(): + yield event + adapter.output_text_parts.append(delta["content"]) + yield adapter.output_delta_event(delta["content"]) + tool_calls = delta.get("tool_calls") + if isinstance(tool_calls, list): + for tool in tool_calls: + if not isinstance(tool, dict): + continue + tool_index = tool.get("index", 0) + call_id = tool.get("id") or _new_tool_call_id() + fn = tool.get("function") or {} + name = fn.get("name") + args_delta = fn.get("arguments") or "" + adapter.record_tool_call( + tool_index, call_id, name, args_delta + ) + for event in adapter.ensure_tool_item( + tool_index, call_id, name + ): + yield event + delta_event = adapter.tool_arguments_delta_event( + tool_index, args_delta + ) + if delta_event: + yield delta_event + + full_text = "".join(adapter.output_text_parts) + if full_text and adapter.message_started: + for event in adapter.output_done_events(full_text): + yield event + for event in adapter.tool_arguments_done_events(): + yield event + yield adapter.completed_event() + + return _stream() + + +__all__ = ["ResponsesService"] diff --git a/app/services/grok/utils/response.py b/app/services/grok/utils/response.py new file mode 100644 index 0000000..572c676 --- /dev/null +++ b/app/services/grok/utils/response.py @@ -0,0 +1,144 @@ +""" +Response formatting utilities for OpenAI-compatible API responses. +""" + +import os +import time +import uuid +from typing import Optional + + +def make_response_id() -> str: + """Generate a unique response ID.""" + return f"chatcmpl-{int(time.time() * 1000)}{os.urandom(4).hex()}" + + +def make_chat_chunk( + response_id: str, + model: str, + content: str, + index: int = 0, + role: str = "assistant", + is_final: bool = False, +) -> dict: + """ + Create an OpenAI-compatible chat completion chunk. + + Args: + response_id: Unique response ID + model: Model name + content: Content to send + index: Choice index + role: Role (assistant) + is_final: Whether this is the final chunk (includes finish_reason) + + Returns: + Chat completion chunk dict + """ + choice: dict = { + "index": index, + "delta": { + "role": role, + "content": content, + }, + } + + if is_final: + choice["finish_reason"] = "stop" + + chunk: dict = { + "id": response_id, + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": model, + "choices": [choice], + } + + if is_final: + chunk["usage"] = { + "total_tokens": 0, + "input_tokens": 0, + "output_tokens": 0, + "input_tokens_details": {"text_tokens": 0, "image_tokens": 0}, + } + + return chunk + + +def make_chat_response( + model: str, + content: str, + response_id: Optional[str] = None, + index: int = 0, + usage: Optional[dict] = None, +) -> dict: + """ + Create an OpenAI-compatible non-streaming chat completion response. + + Args: + model: Model name + content: Response content + response_id: Unique response ID (generated if not provided) + index: Choice index + usage: Custom usage dict (defaults to zeros) + + Returns: + Chat completion response dict + """ + if response_id is None: + response_id = f"chatcmpl-{uuid.uuid4().hex[:8]}" + + if usage is None: + usage = { + "total_tokens": 0, + "input_tokens": 0, + "output_tokens": 0, + "input_tokens_details": {"text_tokens": 0, "image_tokens": 0}, + } + + return { + "id": response_id, + "object": "chat.completion", + "created": int(time.time()), + "model": model, + "choices": [ + { + "index": index, + "message": { + "role": "assistant", + "content": content, + "refusal": None, + }, + "finish_reason": "stop", + } + ], + "usage": usage, + } + + +def wrap_image_content(content: str, response_format: str = "url") -> str: + """ + Wrap image content in markdown format for chat interface. + + Args: + content: Image URL or base64 data + response_format: "url" or "b64_json"/"base64" + + Returns: + Markdown-wrapped image content + """ + if not content: + return content + + if response_format == "url": + return f"![image]({content})" + else: + return f"![image](data:image/png;base64,{content})" + + +__all__ = [ + "make_response_id", + "make_chat_chunk", + "make_chat_response", + "wrap_image_content", +] diff --git a/app/services/grok/utils/tool_call.py b/app/services/grok/utils/tool_call.py new file mode 100644 index 0000000..360d2bd --- /dev/null +++ b/app/services/grok/utils/tool_call.py @@ -0,0 +1,351 @@ +""" +Tool call utilities for OpenAI-compatible function calling. + +Provides prompt-based emulation of tool calls by injecting tool definitions +into the system prompt and parsing structured responses. +""" + +import json +import re +import uuid +from typing import Any, Dict, List, Optional, Tuple + + +def build_tool_prompt( + tools: List[Dict[str, Any]], + tool_choice: Optional[Any] = None, + parallel_tool_calls: bool = True, +) -> str: + """Generate a system prompt block describing available tools. + + Args: + tools: List of OpenAI-format tool definitions. + tool_choice: "auto", "required", "none", or {"type":"function","function":{"name":"..."}}. + parallel_tool_calls: Whether multiple tool calls are allowed. + + Returns: + System prompt string to prepend to the conversation. + """ + if not tools: + return "" + + # tool_choice="none" means don't mention tools at all + if tool_choice == "none": + return "" + + lines = [ + "# Available Tools", + "", + "You have access to the following tools. To call a tool, output a block with a JSON object containing \"name\" and \"arguments\".", + "", + "Format:", + "", + '{"name": "function_name", "arguments": {"param": "value"}}', + "", + "", + ] + + if parallel_tool_calls: + lines.append("You may make multiple tool calls in a single response by using multiple blocks.") + lines.append("") + + # Describe each tool + lines.append("## Tool Definitions") + lines.append("") + for tool in tools: + if tool.get("type") != "function": + continue + func = tool.get("function", {}) + name = func.get("name", "") + desc = func.get("description", "") + params = func.get("parameters", {}) + + lines.append(f"### {name}") + if desc: + lines.append(f"{desc}") + if params: + lines.append(f"Parameters: {json.dumps(params, ensure_ascii=False)}") + lines.append("") + + # Handle tool_choice directives + if tool_choice == "required": + lines.append("IMPORTANT: You MUST call at least one tool in your response. Do not respond with only text.") + elif isinstance(tool_choice, dict): + func_info = tool_choice.get("function", {}) + forced_name = func_info.get("name", "") + if forced_name: + lines.append(f"IMPORTANT: You MUST call the tool \"{forced_name}\" in your response.") + else: + # "auto" or default + lines.append("Decide whether to call a tool based on the user's request. If you don't need a tool, respond normally with text only.") + + lines.append("") + lines.append("When you call a tool, you may include text before or after the blocks, but the tool call blocks must be valid JSON.") + + return "\n".join(lines) + + +_TOOL_CALL_RE = re.compile( + r"\s*(.*?)\s*", + re.DOTALL, +) + + +def _strip_code_fences(text: str) -> str: + if not text: + return text + cleaned = text.strip() + if cleaned.startswith("```"): + cleaned = re.sub(r"^```[a-zA-Z0-9_-]*\s*", "", cleaned) + cleaned = re.sub(r"\s*```$", "", cleaned) + return cleaned.strip() + + +def _extract_json_object(text: str) -> str: + if not text: + return text + start = text.find("{") + if start == -1: + return text + end = text.rfind("}") + if end == -1: + return text[start:] + if end < start: + return text + return text[start : end + 1] + + +def _remove_trailing_commas(text: str) -> str: + if not text: + return text + return re.sub(r",\s*([}\]])", r"\1", text) + + +def _balance_braces(text: str) -> str: + if not text: + return text + open_count = 0 + close_count = 0 + in_string = False + escape = False + for ch in text: + if escape: + escape = False + continue + if ch == "\\" and in_string: + escape = True + continue + if ch == '"': + in_string = not in_string + continue + if in_string: + continue + if ch == "{": + open_count += 1 + elif ch == "}": + close_count += 1 + if open_count > close_count: + text = text + ("}" * (open_count - close_count)) + return text + + +def _repair_json(text: str) -> Optional[Any]: + if not text: + return None + cleaned = _strip_code_fences(text) + cleaned = _extract_json_object(cleaned) + cleaned = cleaned.replace("\r\n", "\n").replace("\r", "\n") + cleaned = cleaned.replace("\n", " ") + cleaned = _remove_trailing_commas(cleaned) + cleaned = _balance_braces(cleaned) + try: + return json.loads(cleaned) + except json.JSONDecodeError: + return None + + +def parse_tool_call_block( + raw_json: str, + tools: Optional[List[Dict[str, Any]]] = None, +) -> Optional[Dict[str, Any]]: + if not raw_json: + return None + parsed = None + try: + parsed = json.loads(raw_json) + except json.JSONDecodeError: + parsed = _repair_json(raw_json) + if not isinstance(parsed, dict): + return None + + name = parsed.get("name") + arguments = parsed.get("arguments", {}) + if not name: + return None + + valid_names = set() + if tools: + for tool in tools: + func = tool.get("function", {}) + tool_name = func.get("name") + if tool_name: + valid_names.add(tool_name) + if valid_names and name not in valid_names: + return None + + if isinstance(arguments, dict): + arguments_str = json.dumps(arguments, ensure_ascii=False) + elif isinstance(arguments, str): + arguments_str = arguments + else: + arguments_str = json.dumps(arguments, ensure_ascii=False) + + return { + "id": f"call_{uuid.uuid4().hex[:24]}", + "type": "function", + "function": {"name": name, "arguments": arguments_str}, + } + + +def parse_tool_calls( + content: str, + tools: Optional[List[Dict[str, Any]]] = None, +) -> Tuple[Optional[str], Optional[List[Dict[str, Any]]]]: + """Parse tool call blocks from model output. + + Detects ``...`` blocks, parses JSON from each block, + and returns OpenAI-format tool call objects. + + Args: + content: Raw model output text. + tools: Optional list of tool definitions for name validation. + + Returns: + Tuple of (text_content, tool_calls_list). + - text_content: text outside blocks (None if empty). + - tool_calls_list: list of OpenAI tool call dicts, or None if no calls found. + """ + if not content: + return content, None + + matches = list(_TOOL_CALL_RE.finditer(content)) + if not matches: + return content, None + + tool_calls = [] + for match in matches: + raw_json = match.group(1).strip() + tool_call = parse_tool_call_block(raw_json, tools) + if tool_call: + tool_calls.append(tool_call) + + if not tool_calls: + return content, None + + # Extract text outside of tool_call blocks + text_parts = [] + last_end = 0 + for match in matches: + before = content[last_end:match.start()] + if before.strip(): + text_parts.append(before.strip()) + last_end = match.end() + trailing = content[last_end:] + if trailing.strip(): + text_parts.append(trailing.strip()) + + text_content = "\n".join(text_parts) if text_parts else None + + return text_content, tool_calls + + +def build_tool_overrides(tools: List[Dict[str, Any]]) -> Dict[str, Any]: + """Convert OpenAI tool format to Grok's toolOverrides format (experimental). + + Best-effort mapping for passthrough mode. + + Args: + tools: List of OpenAI-format tool definitions. + + Returns: + Dict suitable for the toolOverrides field in Grok API payload. + """ + if not tools: + return {} + + tool_overrides = {} + for tool in tools: + if tool.get("type") != "function": + continue + func = tool.get("function", {}) + name = func.get("name", "") + if not name: + continue + tool_overrides[name] = { + "enabled": True, + "description": func.get("description", ""), + "parameters": func.get("parameters", {}), + } + + return tool_overrides + + +def format_tool_history(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Convert assistant messages with tool_calls and tool role messages into text format. + + Since Grok's web API only accepts a single message string, this converts + tool-related messages back to a text representation for multi-turn conversations. + + Args: + messages: List of OpenAI-format messages that may contain tool_calls and tool roles. + + Returns: + List of messages with tool content converted to text format. + """ + result = [] + for msg in messages: + role = msg.get("role", "") + content = msg.get("content") + tool_calls = msg.get("tool_calls") + tool_call_id = msg.get("tool_call_id") + name = msg.get("name") + + if role == "assistant" and tool_calls: + # Convert assistant tool_calls to text representation + parts = [] + if content: + parts.append(content if isinstance(content, str) else str(content)) + for tc in tool_calls: + func = tc.get("function", {}) + tc_name = func.get("name", "") + tc_args = func.get("arguments", "{}") + tc_id = tc.get("id", "") + parts.append(f'{{"name":"{tc_name}","arguments":{tc_args}}}') + result.append({ + "role": "assistant", + "content": "\n".join(parts), + }) + + elif role == "tool": + # Convert tool result to text format + tool_name = name or "unknown" + call_id = tool_call_id or "" + content_str = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False) if content else "" + result.append({ + "role": "user", + "content": f"tool ({tool_name}, {call_id}): {content_str}", + }) + + else: + result.append(msg) + + return result + + +__all__ = [ + "build_tool_prompt", + "parse_tool_calls", + "build_tool_overrides", + "format_tool_history", + "parse_tool_call_block", +] diff --git a/main.py b/main.py index d20fa4d..0eb204b 100644 --- a/main.py +++ b/main.py @@ -41,6 +41,7 @@ from app.api.v1.image import router as image_router # noqa: E402 from app.api.v1.nsfw import router as nsfw_router # noqa: E402 from app.api.v1.files import router as files_router # noqa: E402 from app.api.v1.models import router as models_router # noqa: E402 +from app.api.v1.response import router as responses_router # noqa: E402 from app.services.token import get_scheduler # noqa: E402 from app.api.v1.admin_api import router as admin_router from app.api.v1.public_api import router as public_router @@ -207,6 +208,13 @@ def create_app() -> FastAPI: app.include_router( models_router, prefix="/v1", dependencies=[Depends(verify_api_key)] ) + app.include_router( + responses_router, prefix="/v1", dependencies=[Depends(verify_api_key)] + ) + # 兼容部分客户端直接请求 /responses（不带 /v1） + app.include_router( + responses_router, dependencies=[Depends(verify_api_key)] + ) app.include_router(files_router, prefix="/v1/files") # 静态文件服务