From 481a9c9d19808dcbb1857019657bd68306560713 Mon Sep 17 00:00:00 2001 From: XianYuDaXian Date: Mon, 9 Mar 2026 22:59:37 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AF=B9=E9=BD=90=E4=B8=8A=E6=B8=B8=20?= =?UTF-8?q?5aa29aa/3a77067/02d4f91=20=E5=B9=B6=E5=85=A5=20tool=20calling?= =?UTF-8?q?=20=E5=85=BC=E5=AE=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/v1/chat.py | 28 ++- app/services/grok/services/chat.py | 305 ++++++++++++++++++++++-- app/services/grok/services/responses.py | 12 +- 3 files changed, 311 insertions(+), 34 deletions(-) diff --git a/app/api/v1/chat.py b/app/api/v1/chat.py index 01ae373..ae8ae1e 100644 --- a/app/api/v1/chat.py +++ b/app/api/v1/chat.py @@ -27,7 +27,10 @@ class MessageItem(BaseModel): """消息项""" role: str - content: Union[str, List[Dict[str, Any]]] + content: Optional[Union[str, Dict[str, Any], List[Dict[str, Any]]]] + tool_calls: Optional[List[Dict[str, Any]]] = None + tool_call_id: Optional[str] = None + name: Optional[str] = None class VideoConfig(BaseModel): @@ -62,12 +65,16 @@ class ChatCompletionRequest(BaseModel): video_config: Optional[VideoConfig] = Field(None, description="视频生成参数") # 图片生成配置 image_config: Optional[ImageConfig] = Field(None, description="图片生成参数") + # Tool calling + tools: Optional[List[Dict[str, Any]]] = Field(None, description="Tool definitions") + tool_choice: Optional[Union[str, Dict[str, Any]]] = Field(None, description="Tool choice: auto/required/none/specific") + parallel_tool_calls: Optional[bool] = Field(True, description="Allow parallel tool calls") # 兼容 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"} +VALID_ROLES = {"developer", "system", "user", "assistant", "tool"} USER_CONTENT_TYPES = {"text", "image_url", "input_audio", "file"} ALLOWED_IMAGE_SIZES = { "1280x720", @@ -352,8 +359,12 @@ def validate_request(request: ChatCompletionRequest): code="empty_content", ) + elif isinstance(content, dict): + msg.content = [content] + content = msg.content + # 列表内容 - elif isinstance(content, list): + if isinstance(content, list): if not content: raise ValidationException( message="Message content cannot be an empty array", @@ -462,9 +473,11 @@ def validate_request(request: ChatCompletionRequest): "file.file_data", f"messages.{idx}.content.{block_idx}.file.file_data", ) - else: + elif content is None and msg.role == "tool": + pass + elif not isinstance(content, list): raise ValidationException( - message="Message content must be a string or array", + message="Message content must be a string, object, or array", param=f"messages.{idx}.content", code="invalid_content", ) @@ -967,6 +980,11 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request) reasoning_effort=request.reasoning_effort, temperature=request.temperature, top_p=request.top_p, + tools=request.tools, + tool_choice=request.tool_choice, + parallel_tool_calls=( + True if request.parallel_tool_calls is None else request.parallel_tool_calls + ), ) if isinstance(result, dict): diff --git a/app/services/grok/services/chat.py b/app/services/grok/services/chat.py index dae6ecc..cbfb6ba 100644 --- a/app/services/grok/services/chat.py +++ b/app/services/grok/services/chat.py @@ -27,6 +27,12 @@ from app.services.grok.utils.retry import pick_token, rate_limited 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 +from app.services.grok.utils.tool_call import ( + build_tool_prompt, + parse_tool_calls, + parse_tool_call_block, + format_tool_history, +) _CHAT_SEMAPHORE = None @@ -102,8 +108,16 @@ class MessageExtractor: """消息内容提取器""" @staticmethod - def extract(messages: List[Dict[str, Any]]) -> tuple[str, List[str], List[str]]: + def extract( + messages: List[Dict[str, Any]], + tools: List[Dict[str, Any]] = None, + tool_choice: Any = None, + parallel_tool_calls: bool = True, + ) -> tuple[str, List[str], List[str]]: """从 OpenAI 消息格式提取内容，返回 (text, file_attachments, image_attachments)""" + if tools: + messages = format_tool_history(messages) + texts = [] file_attachments: List[str] = [] image_attachments: List[str] = [] @@ -117,6 +131,34 @@ class MessageExtractor: if isinstance(content, str): if content.strip(): parts.append(content) + elif isinstance(content, dict): + content = [content] + for item in content: + if not isinstance(item, dict): + continue + item_type = item.get("type", "") + + if item_type == "text": + if text := item.get("text", "").strip(): + parts.append(text) + + elif item_type == "image_url": + image_data = item.get("image_url", {}) + url = image_data.get("url", "") + if url: + image_attachments.append(url) + + elif item_type == "input_audio": + audio_data = item.get("input_audio", {}) + data = audio_data.get("data", "") + if data: + file_attachments.append(data) + + elif item_type == "file": + file_data = item.get("file", {}) + raw = file_data.get("file_data", "") + if raw: + file_attachments.append(raw) elif isinstance(content, list): for item in content: item_type = item.get("type", "") @@ -143,8 +185,36 @@ class MessageExtractor: if raw: file_attachments.append(raw) + tool_calls = msg.get("tool_calls") + if role == "assistant" and not parts and isinstance(tool_calls, list): + for call in tool_calls: + if not isinstance(call, dict): + continue + fn = call.get("function", {}) + if not isinstance(fn, dict): + fn = {} + name = fn.get("name") or call.get("name") or "tool" + arguments = fn.get("arguments", "") + if isinstance(arguments, (dict, list)): + try: + arguments = orjson.dumps(arguments).decode() + except Exception: + arguments = str(arguments) + if not isinstance(arguments, str): + arguments = str(arguments) + arguments = arguments.strip() + parts.append(f"[tool_call] {name} {arguments}".strip()) + if parts: - extracted.append({"role": role, "text": "\n".join(parts)}) + role_label = role + if role == "tool": + name = msg.get("name") + call_id = msg.get("tool_call_id") + if isinstance(name, str) and name.strip(): + role_label = f"tool[{name.strip()}]" + if isinstance(call_id, str) and call_id.strip(): + role_label = f"{role_label}#{call_id.strip()}" + extracted.append({"role": role_label, "text": "\n".join(parts)}) # 找到最后一条 user 消息 last_user_index = next( @@ -161,7 +231,17 @@ class MessageExtractor: text = item["text"] texts.append(text if i == last_user_index else f"{role}: {text}") - return "\n\n".join(texts), file_attachments, image_attachments + combined = "\n\n".join(texts) + + if (not combined.strip()) and (file_attachments or image_attachments): + combined = "Refer to the following content:" + + if tools: + tool_prompt = build_tool_prompt(tools, tool_choice, parallel_tool_calls) + if tool_prompt: + combined = f"{tool_prompt}\n\n{combined}" + + return combined, file_attachments, image_attachments class GrokChatService: @@ -235,6 +315,9 @@ class GrokChatService: reasoning_effort: str | None = None, temperature: float = 0.8, top_p: float = 0.95, + tools: List[Dict[str, Any]] = None, + tool_choice: Any = None, + parallel_tool_calls: bool = True, ): """OpenAI 兼容接口""" model_info = ModelService.get(model) @@ -244,7 +327,12 @@ class GrokChatService: grok_model = model_info.grok_model mode = model_info.model_mode # 提取消息和附件 - message, file_attachments, image_attachments = MessageExtractor.extract(messages) + message, file_attachments, image_attachments = MessageExtractor.extract( + messages, + tools=tools, + tool_choice=tool_choice, + parallel_tool_calls=parallel_tool_calls, + ) if not (message or "").strip() and (file_attachments or image_attachments): # 对齐官网行为：仅附件时仍发送非空 message，避免上游 400 message = "参考以下内容：" @@ -282,12 +370,6 @@ class GrokChatService: if reasoning_effort is not None: model_config_override["reasoningEffort"] = reasoning_effort - # Passthrough mode: build tool_overrides for Grok API - tool_overrides_payload = None - # We need to extract tools from some upper context if passed, but since the upstream - # signature changed, we will just patch it carefully. - # Since this is a partial restore, we initialize to None and just pass it. - response = await self.chat( token=token, message=message, @@ -296,7 +378,7 @@ class GrokChatService: mode=mode, stream=stream, file_attachments=all_attachments, - tool_overrides=tool_overrides_payload, + tool_overrides=None, model_config_override=model_config_override, ) @@ -314,6 +396,9 @@ class ChatService: reasoning_effort: str | None = None, temperature: float = 0.8, top_p: float = 0.95, + tools: List[Dict[str, Any]] = None, + tool_choice: Any = None, + parallel_tool_calls: bool = True, ): """Chat Completions 入口""" # 获取 token @@ -358,19 +443,33 @@ class ChatService: reasoning_effort=reasoning_effort, temperature=temperature, top_p=top_p, + tools=tools, + tool_choice=tool_choice, + parallel_tool_calls=parallel_tool_calls, ) # 处理响应 if is_stream: logger.debug(f"Processing stream response: model={model}") - processor = StreamProcessor(model_name, token, show_think) + processor = StreamProcessor( + model_name, + token, + show_think, + tools=tools, + tool_choice=tool_choice, + ) return wrap_stream_with_usage( processor.process(response), token_mgr, token, model ) # 非流式 logger.debug(f"Processing non-stream response: model={model}") - result = await CollectProcessor(model_name, token).process(response) + result = await CollectProcessor( + model_name, + token, + tools=tools, + tool_choice=tool_choice, + ).process(response) try: model_info = ModelService.get(model) effort = ( @@ -413,7 +512,14 @@ class ChatService: class StreamProcessor(proc_base.BaseProcessor): """Stream response processor.""" - def __init__(self, model: str, token: str = "", show_think: bool = None): + def __init__( + self, + model: str, + token: str = "", + show_think: bool = None, + tools: List[Dict[str, Any]] = None, + tool_choice: Any = None, + ): super().__init__(model, token) self.response_id: str = None self.rollout_id: str = "" @@ -431,6 +537,14 @@ class StreamProcessor(proc_base.BaseProcessor): self._tool_usage_buffer = "" self.show_think = bool(show_think) + self.tools = tools + self.tool_choice = tool_choice + self._tool_stream_enabled = bool(tools) and tool_choice != "none" + self._tool_state = "text" + self._tool_buffer = "" + self._tool_partial = "" + self._tool_calls_seen = False + self._tool_call_index = 0 def _filter_tool_card(self, token: str) -> str: if not token or not self.tool_usage_enabled: @@ -505,12 +619,104 @@ class StreamProcessor(proc_base.BaseProcessor): return token - def _sse(self, content: str = "", role: str = None, finish: str = None) -> str: + def _with_tool_index(self, tool_call: Dict[str, Any]) -> Dict[str, Any]: + tool_call = dict(tool_call or {}) + tool_call["index"] = self._tool_call_index + self._tool_call_index += 1 + return tool_call + + @staticmethod + def _suffix_prefix(text: str, marker: str) -> int: + if not text or not marker: + return 0 + max_len = min(len(text), len(marker) - 1) + for keep in range(max_len, 0, -1): + if text.endswith(marker[:keep]): + return keep + return 0 + + def _handle_tool_stream(self, chunk: str) -> list[tuple[str, Any]]: + if not chunk: + return [] + + start_tag = "" + end_tag = "" + events: list[tuple[str, Any]] = [] + data = f"{self._tool_partial}{chunk}" + self._tool_partial = "" + + while data: + if self._tool_state == "text": + start_idx = data.find(start_tag) + if start_idx == -1: + keep = self._suffix_prefix(data, start_tag) + emit = data[:-keep] if keep else data + if emit: + events.append(("text", emit)) + self._tool_partial = data[-keep:] if keep else "" + break + + before = data[:start_idx] + if before: + events.append(("text", before)) + data = data[start_idx + len(start_tag):] + self._tool_state = "tool" + continue + + end_idx = data.find(end_tag) + if end_idx == -1: + keep = self._suffix_prefix(data, end_tag) + append = data[:-keep] if keep else data + if append: + self._tool_buffer += append + self._tool_partial = data[-keep:] if keep else "" + break + + self._tool_buffer += data[:end_idx] + data = data[end_idx + len(end_tag):] + tool_call = parse_tool_call_block(self._tool_buffer, self.tools) + if tool_call: + events.append(("tool", self._with_tool_index(tool_call))) + self._tool_calls_seen = True + self._tool_buffer = "" + self._tool_state = "text" + + return events + + def _flush_tool_stream(self) -> list[tuple[str, Any]]: + events: list[tuple[str, Any]] = [] + if self._tool_state == "text": + if self._tool_partial: + events.append(("text", self._tool_partial)) + self._tool_partial = "" + return events + + raw = f"{self._tool_buffer}{self._tool_partial}" + tool_call = parse_tool_call_block(raw, self.tools) + if tool_call: + events.append(("tool", self._with_tool_index(tool_call))) + self._tool_calls_seen = True + elif raw: + events.append(("text", f"{raw}")) + self._tool_buffer = "" + self._tool_partial = "" + self._tool_state = "text" + return events + + def _sse( + self, + content: str = "", + role: str = None, + finish: str = None, + tool_calls: list = None, + ) -> str: """Build SSE response.""" delta = {} if role: delta["role"] = role delta["content"] = "" + elif tool_calls is not None: + delta["tool_calls"] = tool_calls elif content: delta["content"] = content @@ -648,15 +854,43 @@ class StreamProcessor(proc_base.BaseProcessor): self.answer_buffer.append(filtered) continue + if self._tool_stream_enabled: + for kind, payload in self._handle_tool_stream(filtered): + if kind == "text": + yield self._sse(payload) + elif kind == "tool": + yield self._sse(tool_calls=[payload]) + continue + yield self._sse(filtered) if self.think_opened: yield self._sse("\n") self.think_opened = False if self.answer_buffer: - yield self._sse("".join(self.answer_buffer)) + buffered = "".join(self.answer_buffer) + if self._tool_stream_enabled: + for kind, payload in self._handle_tool_stream(buffered): + if kind == "text": + yield self._sse(payload) + elif kind == "tool": + yield self._sse(tool_calls=[payload]) + for kind, payload in self._flush_tool_stream(): + if kind == "text": + yield self._sse(payload) + elif kind == "tool": + yield self._sse(tool_calls=[payload]) + else: + yield self._sse(buffered) self.answer_buffer = [] - yield self._sse(finish="stop") + if self._tool_stream_enabled and not self.answer_buffer: + for kind, payload in self._flush_tool_stream(): + if kind == "text": + yield self._sse(payload) + elif kind == "tool": + yield self._sse(tool_calls=[payload]) + finish_reason = "tool_calls" if self._tool_stream_enabled and self._tool_calls_seen else "stop" + yield self._sse(finish=finish_reason) yield "data: [DONE]\n\n" except asyncio.CancelledError: logger.debug("Stream cancelled by client", extra={"model": self.model}) @@ -710,9 +944,17 @@ class StreamProcessor(proc_base.BaseProcessor): class CollectProcessor(proc_base.BaseProcessor): """Non-stream response processor.""" - def __init__(self, model: str, token: str = ""): + def __init__( + self, + model: str, + token: str = "", + tools: List[Dict[str, Any]] = None, + tool_choice: Any = None, + ): super().__init__(model, token) self.filter_tags = get_config("app.filter_tags") + self.tools = tools + self.tool_choice = tool_choice def _filter_content(self, content: str) -> str: """Filter special tags in content.""" @@ -855,6 +1097,24 @@ class CollectProcessor(proc_base.BaseProcessor): content = self._filter_content(content) + finish_reason = "stop" + tool_calls_result = None + if self.tools and self.tool_choice != "none": + text_content, tool_calls_list = parse_tool_calls(content, self.tools) + if tool_calls_list: + tool_calls_result = tool_calls_list + content = text_content + finish_reason = "tool_calls" + + message_obj = { + "role": "assistant", + "content": content, + "refusal": None, + "annotations": [], + } + if tool_calls_result: + message_obj["tool_calls"] = tool_calls_result + return { "id": response_id, "object": "chat.completion", @@ -864,13 +1124,8 @@ class CollectProcessor(proc_base.BaseProcessor): "choices": [ { "index": 0, - "message": { - "role": "assistant", - "content": content, - "refusal": None, - "annotations": [], - }, - "finish_reason": "stop", + "message": message_obj, + "finish_reason": finish_reason, } ], "usage": { diff --git a/app/services/grok/services/responses.py b/app/services/grok/services/responses.py index 01d1818..66cb234 100644 --- a/app/services/grok/services/responses.py +++ b/app/services/grok/services/responses.py @@ -699,15 +699,19 @@ class ResponsesService: "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 + if normalized_tools is not None: + chat_kwargs["tools"] = normalized_tools + if normalized_tool_choice is not None: + chat_kwargs["tool_choice"] = normalized_tool_choice + if parallel_tool_calls is not None: + chat_kwargs["parallel_tool_calls"] = parallel_tool_calls + if reasoning_effort is not None: + chat_kwargs["reasoning_effort"] = reasoning_effort result = await ChatService.completions(**chat_kwargs)