feat(agent): implement streaming responses for online Agent and enhance tool execution feedback

docs/content/docs/progress/pending-test.mdx

@@ -6,3 +6,5 @@ description: 当前版本已实现但仍需人工验证的变更项
 # 待测试
 
 - 在线 `Agent` 工具卡片详情不再保存和展示完整上下文消息，只展示当前工具调用和执行结果；需要验证详情内容体积不会随对话历史膨胀。
+- 文本生成和在线 `Agent` 的 Responses API 请求已改为流式输出；需要验证生成文本、工具确认前后的对话回复会实时刷新。
+- 在线 `Agent` 自动执行工具时会显示已完成工具卡片，并且从生成配置节点触发生成时会使用配置节点的提示词；需要验证关闭工具确认后连接节点并生图时请求不再发送空 `prompt`。

web/src/app/(user)/canvas/[id]/canvas-client-page.tsx

@@ -683,7 +683,13 @@ function InfiniteCanvasPage() {
             setViewport(next.viewport);
             setContextMenu(null);
             if (generationOps.length) {
-                queueMicrotask(() => generationOps.forEach((op) => void generateNodeRef.current?.(op.nodeId, op.mode || "image", op.prompt || "")));
+                queueMicrotask(() =>
+                    generationOps.forEach((op) => {
+                        const target = nodesRef.current.find((node) => node.id === op.nodeId);
+                        const prompt = op.prompt?.trim() ? op.prompt : target?.metadata?.composerContent ?? target?.metadata?.prompt ?? "";
+                        void generateNodeRef.current?.(op.nodeId, op.mode || target?.metadata?.generationMode || "image", prompt);
+                    }),
+                );
             }
             return { ...next, projectId, title: currentProject?.title || "未命名画布" };
         },

web/src/app/(user)/canvas/components/canvas-assistant-panel.tsx

@@ -287,12 +287,16 @@ export function CanvasAssistantPanel({ nodes, selectedNodeIds, snapshot, session
             setIsRunning(true);
             const messages = await buildToolAgentMessages(snapshotRef.current, history, userMessage);
             addOnlineLog(`Agent Tool Loop ${loop.step} 开始`, { toolChoice: "required" });
-            const result = await requestToolResponse({ ...requestConfig, systemPrompt: "" }, messages, ONLINE_AGENT_TOOLS, "required");
+            let streamed = "";
+            const result = await requestToolResponse({ ...requestConfig, systemPrompt: "" }, messages, ONLINE_AGENT_TOOLS, "required", (text) => {
+                streamed = text;
+                if (text.trim()) upsertMessage(sessionId, { id: assistantId, role: "assistant", text });
+            });
             addOnlineLog("模型工具回复", result);
             if (result.toolCalls.length) {
                 const writableCalls = result.toolCalls.filter(isWritableToolCall);
                 if (confirmTools && writableCalls.length) {
-                    upsertMessage(sessionId, { id: assistantId, role: "assistant", text: result.content || "准备执行工具，等待确认。" });
+                    upsertMessage(sessionId, { id: assistantId, role: "assistant", text: result.content || streamed || "准备执行工具，等待确认。" });
                     const toolMessageId = nanoid();
                     pendingToolContextRef.current.set(toolMessageId, { messages, toolCalls: result.toolCalls, assistantId, step: loop.step });
                     const toolMessage: CanvasAssistantMessage = { id: toolMessageId, role: "tool", title: "确认工具调用", text: summarizeToolCalls(result.toolCalls), detail: { status: "pending", step: loop.step, toolCalls: result.toolCalls } };
@@ -303,7 +307,7 @@ export function CanvasAssistantPanel({ nodes, selectedNodeIds, snapshot, session
                 await continueOnlineToolLoop(sessionId, assistantId, messages, result, loop.step);
             } else {
                 if (!result.content.trim()) throw new Error("模型没有返回工具调用，画布操作未执行。");
-                upsertMessage(sessionId, { id: assistantId, role: "assistant", text: result.content || "没有返回内容。" });
+                upsertMessage(sessionId, { id: assistantId, role: "assistant", text: result.content || streamed || "没有返回内容。" });
                 addOnlineLog(`Agent Tool Loop ${loop.step} 结束`, { reply: result.content });
             }
         } catch (error) {
@@ -317,6 +321,13 @@ export function CanvasAssistantPanel({ nodes, selectedNodeIds, snapshot, session
     const continueOnlineToolLoop = async (sessionId: string, assistantId: string, messages: ResponseInputMessage[], result: { content: string; toolCalls: ResponseToolCall[] }, step: number) => {
         const toolResults = executeOnlineToolCalls(result.toolCalls);
         addOnlineLog("工具执行结果", toolResults);
+        appendMessage(sessionId, {
+            id: nanoid(),
+            role: "tool",
+            title: "工具自动执行完成",
+            text: toolResults.map((item) => toolResultText(item.result)).join("\n"),
+            detail: { status: "completed", step, toolCalls: result.toolCalls, results: toolResults },
+        });
         await continueOnlineToolLoopAfterResults(sessionId, assistantId, messages, result.toolCalls, toolResults, step);
     };
 
@@ -332,12 +343,16 @@ export function CanvasAssistantPanel({ nodes, selectedNodeIds, snapshot, session
             return;
         }
         const requestConfig = { ...effectiveConfig, model: effectiveConfig.textModel || effectiveConfig.model };
-        const next = await requestToolResponse({ ...requestConfig, systemPrompt: "" }, nextMessages, ONLINE_AGENT_TOOLS);
+        let streamed = "";
+        const next = await requestToolResponse({ ...requestConfig, systemPrompt: "" }, nextMessages, ONLINE_AGENT_TOOLS, "auto", (text) => {
+            streamed = text;
+            if (text.trim()) upsertMessage(sessionId, { id: assistantId, role: "assistant", text });
+        });
         addOnlineLog(`Agent Tool Loop ${step + 1} 回复`, next);
         if (next.toolCalls.length) {
             const writableCalls = next.toolCalls.filter(isWritableToolCall);
             if (confirmTools && writableCalls.length) {
-                upsertMessage(sessionId, { id: assistantId, role: "assistant", text: next.content || "准备执行工具，等待确认。" });
+                upsertMessage(sessionId, { id: assistantId, role: "assistant", text: next.content || streamed || "准备执行工具，等待确认。" });
                 const toolMessageId = nanoid();
                 pendingToolContextRef.current.set(toolMessageId, { messages: nextMessages, toolCalls: next.toolCalls, assistantId, step: step + 1 });
                 appendMessage(sessionId, { id: toolMessageId, role: "tool", title: "确认工具调用", text: summarizeToolCalls(next.toolCalls), detail: { status: "pending", step: step + 1, toolCalls: next.toolCalls } });
@@ -347,7 +362,7 @@ export function CanvasAssistantPanel({ nodes, selectedNodeIds, snapshot, session
             await continueOnlineToolLoop(sessionId, assistantId, nextMessages, next, step + 1);
             return;
         }
-        upsertMessage(sessionId, { id: assistantId, role: "assistant", text: next.content || toolResults.map((item) => toolResultText(item.result)).join("\n") || "工具已执行。" });
+        upsertMessage(sessionId, { id: assistantId, role: "assistant", text: next.content || streamed || toolResults.map((item) => toolResultText(item.result)).join("\n") || "工具已执行。" });
     };
 
     const executeOps = (ops: CanvasAgentOp[]) => {

web/src/services/api/image.ts

@@ -63,6 +63,7 @@ type ResponseApiPayload = {
     code?: number;
     msg?: string;
 };
+type ResponseStreamState = { buffer: string; text: string; payload?: ResponseApiPayload; error?: string };
 
 type ImageApiResponse = {
     data?: Array<Record<string, unknown>>;
@@ -263,6 +264,108 @@ function parseToolResponse(payload: ResponseApiPayload): ToolResponseResult {
     return { content, toolCalls };
 }
 
+function isRecord(value: unknown): value is Record<string, unknown> {
+    return Boolean(value && typeof value === "object" && !Array.isArray(value));
+}
+
+function responseErrorMessage(value: unknown) {
+    if (!isRecord(value)) return "";
+    const error = isRecord(value.error) ? value.error : undefined;
+    const response = isRecord(value.response) ? value.response : undefined;
+    const responseError = response && isRecord(response.error) ? response.error : undefined;
+    return stringValue(value.msg) || stringValue(error?.message) || stringValue(responseError?.message);
+}
+
+function stringValue(value: unknown) {
+    return typeof value === "string" ? value : "";
+}
+
+function validateResponsePayload(payload: ResponseApiPayload) {
+    if (typeof payload.code === "number" && payload.code !== 0) throw new Error(payload.msg || "请求失败");
+    if (payload.error?.message) throw new Error(payload.error.message);
+}
+
+async function readFetchError(response: Response, fallback: string) {
+    const text = await response.text();
+    if (!text) return readStatusError(response.status, fallback);
+    try {
+        return responseErrorMessage(JSON.parse(text)) || readStatusError(response.status, fallback);
+    } catch {
+        return text.slice(0, 300) || readStatusError(response.status, fallback);
+    }
+}
+
+function consumeResponseStreamBlock(block: string, state: ResponseStreamState, onDelta?: (text: string) => void) {
+    const data = block
+        .split(/\r?\n/)
+        .filter((line) => line.startsWith("data:"))
+        .map((line) => line.slice(5).replace(/^ /, ""))
+        .join("\n")
+        .trim();
+    if (!data || data === "[DONE]") return;
+    const event = JSON.parse(data) as Record<string, unknown>;
+    const type = stringValue(event.type);
+    const errorMessage = responseErrorMessage(event);
+    if (errorMessage) state.error = errorMessage;
+    if (type === "response.output_text.delta" && typeof event.delta === "string") {
+        state.text += event.delta;
+        onDelta?.(state.text);
+    }
+    if (type === "response.output_text.done" && !state.text && typeof event.text === "string") {
+        state.text = event.text;
+        onDelta?.(state.text);
+    }
+    if (type === "response.completed" && isRecord(event.response)) {
+        state.payload = event.response as ResponseApiPayload;
+    } else if (Array.isArray(event.output)) {
+        state.payload = event as ResponseApiPayload;
+    }
+}
+
+function consumeResponseStreamText(state: ResponseStreamState, text: string, onDelta?: (text: string) => void, flush = false) {
+    state.buffer += text;
+    for (;;) {
+        const match = state.buffer.match(/\r?\n\r?\n/);
+        if (!match) break;
+        consumeResponseStreamBlock(state.buffer.slice(0, match.index), state, onDelta);
+        state.buffer = state.buffer.slice(match.index + match[0].length);
+    }
+    if (flush && state.buffer.trim()) {
+        consumeResponseStreamBlock(state.buffer, state, onDelta);
+        state.buffer = "";
+    }
+}
+
+async function requestStreamingResponse(config: AiConfig, body: Record<string, unknown>, onDelta?: (text: string) => void): Promise<ToolResponseResult> {
+    const response = await fetch(aiApiUrl(config, "/responses"), {
+        method: "POST",
+        headers: { ...aiHeaders(config, "application/json"), Accept: "text/event-stream" },
+        body: JSON.stringify({ ...body, stream: true }),
+    });
+    if (!response.ok) throw new Error(await readFetchError(response, "请求失败"));
+    if (!response.body) {
+        const payload = (await response.json()) as ResponseApiPayload;
+        validateResponsePayload(payload);
+        return parseToolResponse(payload);
+    }
+
+    const reader = response.body.getReader();
+    const decoder = new TextDecoder();
+    const state: ResponseStreamState = { buffer: "", text: "" };
+    for (;;) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        consumeResponseStreamText(state, decoder.decode(value, { stream: true }), onDelta);
+        if (state.error) throw new Error(state.error);
+    }
+    consumeResponseStreamText(state, decoder.decode(), onDelta, true);
+    if (state.error) throw new Error(state.error);
+    if (!state.payload) return { content: state.text, toolCalls: [] };
+    validateResponsePayload(state.payload);
+    const result = parseToolResponse(state.payload);
+    return { ...result, content: state.text || result.content };
+}
+
 export async function requestGeneration(config: AiConfig, prompt: string) {
     const requestConfig = resolveModelRequestConfig(config, config.model || config.imageModel);
     const n = Math.max(1, Math.min(15, Math.floor(Math.abs(Number(config.count)) || 1)));
@@ -325,45 +428,27 @@ export async function requestEdit(config: AiConfig, prompt: string, references:
 export async function requestImageQuestion(config: AiConfig, messages: AiTextMessage[], onDelta: (text: string) => void) {
     const requestConfig = resolveModelRequestConfig(config, config.model || config.textModel);
     try {
-        const response = await axios.post<ResponseApiPayload>(
+        const answer = (await requestStreamingResponse(requestConfig, {
-            aiApiUrl(requestConfig, "/responses"),
+            model: requestConfig.model,
-            {
+            input: toResponseInput(withSystemMessage(requestConfig, messages)),
-                model: requestConfig.model,
+        }, onDelta)).content || "没有返回内容";
-                input: toResponseInput(withSystemMessage(requestConfig, messages)),
+        if (answer === "没有返回内容") onDelta(answer);
-            },
-            {
-                headers: aiHeaders(requestConfig, "application/json"),
-            },
-        );
-        if (typeof response.data.code === "number" && response.data.code !== 0) throw new Error(response.data.msg || "请求失败");
-        if (response.data.error?.message) throw new Error(response.data.error.message);
-        const answer = parseToolResponse(response.data).content || "没有返回内容";
-        onDelta(answer);
         return answer;
     } catch (error) {
         throw new Error(readAxiosError(error, "请求失败"));
     }
 }
 
-export async function requestToolResponse(config: AiConfig, messages: ResponseInputMessage[], tools: ResponseFunctionTool[], toolChoice: ToolChoice = "auto"): Promise<ToolResponseResult> {
+export async function requestToolResponse(config: AiConfig, messages: ResponseInputMessage[], tools: ResponseFunctionTool[], toolChoice: ToolChoice = "auto", onDelta?: (text: string) => void): Promise<ToolResponseResult> {
     const requestConfig = resolveModelRequestConfig(config, config.model || config.textModel);
     try {
-        const response = await axios.post<ResponseApiPayload>(
+        return await requestStreamingResponse(requestConfig, {
-            aiApiUrl(requestConfig, "/responses"),
+            model: requestConfig.model,
-            {
+            input: toResponseInput(withSystemMessage(requestConfig, messages)),
-                model: requestConfig.model,
+            tools: tools.map(toResponseTool),
-                input: toResponseInput(withSystemMessage(requestConfig, messages)),
+            tool_choice: toolChoice,
-                tools: tools.map(toResponseTool),
+            parallel_tool_calls: false,
-                tool_choice: toolChoice,
+        }, onDelta);
-                parallel_tool_calls: false,
-            },
-            {
-                headers: aiHeaders(requestConfig, "application/json"),
-            },
-        );
-        if (typeof response.data.code === "number" && response.data.code !== 0) throw new Error(response.data.msg || "请求失败");
-        if (response.data.error?.message) throw new Error(response.data.error.message);
-        return parseToolResponse(response.data);
     } catch (error) {
         throw new Error(readAxiosError(error, "请求失败"));
     }