From 511b155da8fb7331a92e20cbff533efccf817447 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 5 Mar 2026 22:29:01 +0800 Subject: [PATCH] feat(ai-chat-api): implement streaming tool calls with incremental parameter accumulation - Update API documentation to reflect new streaming tool call mechanism - Add `message_id` optional field to content and tool_call events - Change tool_call structure from array to single object with incremental updates - Add `tool_call_id` to tool_start events for better event correlation - Implement ToolCallStreamAccumulator class to handle parameter accumulation - Provide frontend example code for handling streaming tool calls - Maintain backward compatibility with existing session_id tracking --- docs/gns3-copilot/ai-chat-api-design.md | 127 +++++++++++-- .../agent/gns3_copilot/agent_service.py | 69 ++++--- .../tools_v2/display_tools_nornir.py | 4 +- .../gns3_copilot/utils/tool_call_stream.py | 173 ++++++++++++++++++ 4 files changed, 318 insertions(+), 55 deletions(-) create mode 100644 gns3server/agent/gns3_copilot/utils/tool_call_stream.py diff --git a/docs/gns3-copilot/ai-chat-api-design.md b/docs/gns3-copilot/ai-chat-api-design.md index 8867cf31c..951e96e71 100644 --- a/docs/gns3-copilot/ai-chat-api-design.md +++ b/docs/gns3-copilot/ai-chat-api-design.md @@ -241,9 +241,9 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。 | type | 说明 | 包含字段 | |------|------|----------| -| content | AI 文本内容（流式） | content | -| tool_call | LLM 决定调用工具（可能多个） | tool_calls (数组, 每项包含 id, name, args), session_id | -| tool_start | 工具开始执行 | tool_name, session_id | +| content | AI 文本内容（流式） | content, message_id (可选) | +| tool_call | LLM 决定调用工具（流式，参数逐次累积） | tool_call (对象, 包含 id, type, function), session_id, message_id (可选) | +| tool_start | 工具开始执行 | tool_name, tool_call_id, session_id | | tool_end | 工具执行完成 | tool_name, tool_output, session_id | | error | 错误信息 | error, session_id | | done | 流结束 | session_id | @@ -255,27 +255,71 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。 // AI 文本流式输出 {"type": "content", "content": "Hello! How can I help"} -// LLM 决定调用工具（单个或多个） +// LLM 决定调用工具（流式传输，参数逐次累积） + +// 第 1 个 chunk：工具调用开始（参数为空） { "type": "tool_call", - "tool_calls": [ - { - "id": "call_123", - "name": "execute_multiple_device_commands", - "args": { - "device_names": ["R1", "R2"], - "commands": ["show version"] - } - } - ], + "tool_call": { + "id": "call_123", + "type": "function", + "function": {"name": "execute_multiple_device_commands", "arguments": ""} + }, "session_id": "xxx" } -// 工具开始执行 -{"type": "tool_start", "tool_name": "execute_multiple_device_commands", "session_id": "xxx"} +// 第 2 个 chunk：参数累积中 +{ + "type": "tool_call", + "tool_call": { + "id": "call_123", + "type": "function", + "function": {"name": "execute_multiple_device_commands", "arguments": "{\"device_names\": [\"R1\"], "} + }, + "session_id": "xxx" +} + +// 第 3 个 chunk：参数累积中 +{ + "type": "tool_call", + "tool_call": { + "id": "call_123", + "type": "function", + "function": {"name": "execute_multiple_device_commands", "arguments": "{\"device_names\": [\"R1\"], \"commands\": [\"show ver\"]}"} + }, + "session_id": "xxx" +} + +// 第 4 个 chunk：参数完整（标记 complete=true） +{ + "type": "tool_call", + "tool_call": { + "id": "call_123", + "type": "function", + "function": { + "name": "execute_multiple_device_commands", + "arguments": "{\"device_names\": [\"R1\"], \"commands\": [\"show ver\"]}", + "complete": true + } + }, + "session_id": "xxx" +} + +// 工具开始执行（通过 tool_call_id 关联） +{ + "type": "tool_start", + "tool_name": "execute_multiple_device_commands", + "tool_call_id": "call_123", + "session_id": "xxx" +} // 工具执行完成 -{"type": "tool_end", "tool_name": "execute_multiple_device_commands", "tool_output": "{...}", "session_id": "xxx"} +{ + "type": "tool_end", + "tool_name": "execute_multiple_device_commands", + "tool_output": "{...}", + "session_id": "xxx" +} // 流结束 {"type": "done", "session_id": "xxx"} @@ -284,6 +328,55 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。 {"type": "error", "error": "Project not found", "session_id": "xxx"} ``` +### 流式工具调用机制 + +**背景**：LLM 生成工具调用参数时是逐字符流式输出的，就像文本内容一样。 + +**实现**：使用 `ToolCallStreamAccumulator` 类维护状态，处理三个阶段： + +1. **初始化阶段**：从 `tool_calls` 获取工具 ID 和名称，发送初始 `tool_call` 事件（参数为空） + +2. **累积阶段**：从 `tool_call_chunks` 逐个获取参数片段，通过字符串拼接累积完整参数，每次累积后发送更新的 `tool_call` 事件 + +3. **完成阶段**：检测 `finish_reason == "tool_calls"` 或 `"stop"`，发送最终 `tool_call` 事件（标记 `complete: true`） + +**前端处理**： +- 收到 `tool_call` 事件时，根据 `tool_call.id` 判断是否为新工具调用 +- 同一个 ID 的后续事件用于更新参数显示 +- 当 `function.complete: true` 时，参数已完整，可以执行工具 +- `tool_start` 事件包含 `tool_call_id`，可以关联到之前的 `tool_call` 事件 + +**示例代码**（前端）： +```javascript +// 维护当前工具调用状态 +let currentToolCall = null; + +function handleToolCallEvent(chunk) { + const toolCall = chunk.tool_call; + + if (!currentToolCall || currentToolCall.id !== toolCall.id) { + // 新工具调用 + currentToolCall = { + id: toolCall.id, + name: toolCall.function.name, + arguments: toolCall.function.arguments, + complete: toolCall.function.complete || false + }; + displayToolCallStarted(currentToolCall); + } else { + // 更新现有工具调用的参数 + currentToolCall.arguments = toolCall.function.arguments; + currentToolCall.complete = toolCall.function.complete || false; + updateToolCallArguments(currentToolCall); + } + + if (currentToolCall.complete) { + // 参数完整，准备执行工具 + displayToolCallReady(currentToolCall); + } +} +``` + ### 心跳机制 **作用**：防止代理服务器/负载均衡器因超时断开 SSE 连接。 diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py index 82c855952..74f31fb09 100644 --- a/gns3server/agent/gns3_copilot/agent_service.py +++ b/gns3server/agent/gns3_copilot/agent_service.py @@ -59,6 +59,9 @@ from gns3server.agent.gns3_copilot.gns3_client.context_helpers import ( from gns3server.agent.gns3_copilot.utils.message_converters import ( convert_langchain_to_openai, ) +from gns3server.agent.gns3_copilot.utils.tool_call_stream import ( + ToolCallStreamAccumulator, +) log = logging.getLogger(__name__) @@ -294,6 +297,9 @@ class AgentService: ai_response_counted = False tool_messages_counted = 0 + # Initialize tool call stream accumulator for handling progressive tool call arguments + tool_call_accumulator = ToolCallStreamAccumulator() + # Stream events try: async for event in graph.astream_events(inputs, config=config, version="v2"): @@ -350,10 +356,20 @@ class AgentService: log.debug("Tool message counted, message_count=%d", message_count) # Convert event to chunk for SSE streaming - chunk = self._convert_event_to_chunk(event, session_id) - if chunk: - log.debug("Yielding chunk: type=%s", chunk.get("type")) - yield chunk + # Use accumulator for on_chat_model_stream events to handle progressive tool calls + if event_type == "on_chat_model_stream": + chunks = tool_call_accumulator.process_event(event) + for chunk in chunks: + # Add session_id to each chunk + chunk["session_id"] = session_id + log.debug("Yielding accumulated chunk: type=%s", chunk.get("type")) + yield chunk + else: + # Use stateless converter for other events + chunk = self._convert_event_to_chunk(event, session_id) + if chunk: + log.debug("Yielding chunk: type=%s", chunk.get("type")) + yield chunk # Update session statistics after successful stream await repo.update_session( @@ -398,43 +414,24 @@ class AgentService: Returns: Dict for SSE response or None if event should be filtered + + Note: + on_chat_model_stream events are handled by ToolCallStreamAccumulator + before calling this method, so they are not processed here. """ event_type = event.get("event", "") data = event.get("data", {}) - if event_type == "on_chat_model_stream": - # Streaming text content from LLM - chunk = data.get("chunk", {}) - # chunk is AIMessageChunk object, access content directly - content = getattr(chunk, "content", "") - if content: - return {"type": "content", "content": content} - - elif event_type == "on_chat_model_end": - # LLM call completed, check if it decided to call tools - output = data.get("output", {}) - if hasattr(output, "tool_calls") and output.tool_calls: - # Extract tool calls information - tool_calls_data = [] - for tc in output.tool_calls: - # Convert to dict if it's an object - tc_dict = tc if isinstance(tc, dict) else tc.model_dump() - tool_calls_data.append( - { - "id": tc_dict.get("id", ""), - "name": tc_dict.get("name", ""), - "args": tc_dict.get("args", {}), - } - ) - return { - "type": "tool_call", - "tool_calls": tool_calls_data, - "session_id": session_id, - } - - elif event_type == "on_tool_start": + if event_type == "on_tool_start": # Tool execution started - return {"type": "tool_start", "tool_name": event.get("name", ""), "session_id": session_id} + # Extract tool_call_id from event metadata to associate with tool_call event + tool_call_id = event.get("metadata", {}).get("tool_call_id", "") + return { + "type": "tool_start", + "tool_name": event.get("name", ""), + "tool_call_id": tool_call_id, + "session_id": session_id, + } elif event_type == "on_tool_end": # Tool execution completed diff --git a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py index 61b084d3d..59cbc362a 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -178,7 +178,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): List[Dict[str, Any]]: A list of dictionaries containing device names and command outputs. """ # Log received input - logger.info("Received input: %s", tool_input) + logger.debug("Received input: %s", tool_input) # Validate input device_configs_list, project_id = self._validate_tool_input(tool_input) @@ -219,7 +219,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): logger.error("Error executing display on all devices: %s", e) return [{"error": f"Execution error: {str(e)}"}] - logger.info( + logger.debug( "Multiple device display execution completed. Results: %s", json.dumps(results, indent=2, ensure_ascii=False), ) diff --git a/gns3server/agent/gns3_copilot/utils/tool_call_stream.py b/gns3server/agent/gns3_copilot/utils/tool_call_stream.py new file mode 100644 index 000000000..c598adb84 --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/tool_call_stream.py @@ -0,0 +1,173 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +# +# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3 +# +# This file is part of GNS3-Copilot project. +# +# GNS3-Copilot is free software: you can redistribute it and/or modify it +# under the terms of the GNU General Public License as published by the +# Free Software Foundation, either version 3 of the License, or (at your +# option) any later version. +# +# GNS3-Copilot is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY +# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License +# for more details. +# +# You should have received a copy of the GNU General Public License +# along with GNS3-Copilot. If not, see . +# +# Copyright (C) 2025 Guobin Yue +# Author: Guobin Yue +# +# Project Home: https://github.com/yueguobin/gns3-copilot +# + +""" +Tool call streaming accumulator for handling progressive tool call arguments +Maintains state for streaming tool call chunks +Based on FlowNet-Lab implementation +""" + +from typing import Any, Dict, List, Optional + + +class ToolCallStreamAccumulator: + """ + Accumulates tool call information from streaming events + Handles the progressive build-up of tool call arguments + + This class processes LangGraph streaming events and accumulates + tool call arguments that come in chunks, emitting progressive + tool_call events to the frontend. + """ + + def __init__(self) -> None: + # Current active tool call being accumulated + # Format: {"id": str, "name": str, "args_string": str} + self._current_tool_call: Optional[Dict[str, str]] = None + + def process_event(self, event: Dict[str, Any]) -> List[Dict[str, Any]]: + """ + Process a streaming event and return one or more response chunks + + Args: + event: LangGraph streaming event (on_chat_model_stream) + + Returns: + List of response chunks to send to frontend + + Processing phases: + 1. Initialize: Extract tool ID and name from tool_calls + 2. Accumulate: Concatenate argument strings from tool_call_chunks + 3. Complete: Mark as complete when finish_reason is "tool_calls" or "stop" + """ + event_type = event.get("event", "") + chunks = [] + + if event_type == "on_chat_model_stream": + chunk = event.get("data", {}).get("chunk", {}) + + # ========== Phase 1: Initialize tool call from tool_calls ========== + # Get metadata (ID and name) from tool_calls + if hasattr(chunk, "tool_calls") and chunk.tool_calls: + for tool_call in chunk.tool_calls: + if isinstance(tool_call, dict): + tc_id = tool_call.get("id") + tc_name = tool_call.get("name", "") + else: + tc_id = getattr(tool_call, "id", None) + tc_name = getattr(tool_call, "name", "") + + # Only when ID is not empty, consider it as the start of a new tool call + if tc_id: + # Initialize current tool state (this is the only time to get ID) + # Note: only one tool can be called at a time + self._current_tool_call = { + "id": tc_id, + "name": tc_name if tc_name else "UNKNOWN_TOOL", + "args_string": "", + } + + # Send initial tool_call event with empty args + chunks.append({ + "type": "tool_call", + "tool_call": { + "id": tc_id, + "type": "function", + "function": { + "name": self._current_tool_call["name"], + "arguments": "" + } + } + }) + + # ========== Phase 2: Concatenate parameter strings from tool_call_chunks ========== + # Concatenate parameter strings from tool_call_chunk + if hasattr(chunk, "tool_call_chunks") and chunk.tool_call_chunks: + if self._current_tool_call: + tool_data = self._current_tool_call + for tc_chunk in chunk.tool_call_chunks: + # Default to "" instead of None + if isinstance(tc_chunk, dict): + args_chunk = tc_chunk.get("args", "") + else: + args_chunk = getattr(tc_chunk, "args", "") + + # Core: string concatenation + if isinstance(args_chunk, str): + tool_data["args_string"] += args_chunk + + # Send updated tool_call event with accumulated args + chunks.append({ + "type": "tool_call", + "tool_call": { + "id": tool_data["id"], + "type": "function", + "function": { + "name": tool_data["name"], + "arguments": tool_data["args_string"] + } + } + }) + + # ========== Phase 3: Determine if tool_calls_chunks output is complete ========== + # Check finish_reason == "tool_calls" or "STOP" + response_metadata = getattr(chunk, "response_metadata", {}) + finish_reason = response_metadata.get("finish_reason") if isinstance(response_metadata, dict) else None + + if (finish_reason == "tool_calls") or (finish_reason == "stop" and self._current_tool_call is not None): + if self._current_tool_call: + tool_data = self._current_tool_call + + # Send final complete tool_call event + chunks.append({ + "type": "tool_call", + "tool_call": { + "id": tool_data["id"], + "type": "function", + "function": { + "name": tool_data["name"], + "arguments": tool_data["args_string"], + "complete": True # Mark as complete + } + } + }) + + # Clear the current tool call state + self._current_tool_call = None + + # Also handle regular content (when not in tool call mode) + content = getattr(chunk, "content", "") + if content and not self._current_tool_call: + chunks.append({ + "type": "content", + "content": content, + "message_id": event.get("metadata", {}).get("msg_id") + }) + + return chunks + + def reset(self) -> None: + """Reset the accumulator state""" + self._current_tool_call = None