diff --git a/docs/content/docs/progress/pending-test.mdx b/docs/content/docs/progress/pending-test.mdx index ad3ceda..e656fbc 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -6,3 +6,5 @@ description: 当前版本已实现但仍需人工验证的变更项 # 待测试 - 在线 `Agent` 工具卡片详情不再保存和展示完整上下文消息,只展示当前工具调用和执行结果;需要验证详情内容体积不会随对话历史膨胀。 +- 文本生成和在线 `Agent` 的 Responses API 请求已改为流式输出;需要验证生成文本、工具确认前后的对话回复会实时刷新。 +- 在线 `Agent` 自动执行工具时会显示已完成工具卡片,并且从生成配置节点触发生成时会使用配置节点的提示词;需要验证关闭工具确认后连接节点并生图时请求不再发送空 `prompt`。 diff --git a/web/src/app/(user)/canvas/[id]/canvas-client-page.tsx b/web/src/app/(user)/canvas/[id]/canvas-client-page.tsx index 400a6b1..59a379c 100644 --- a/web/src/app/(user)/canvas/[id]/canvas-client-page.tsx +++ b/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 || "未命名画布" }; }, diff --git a/web/src/app/(user)/canvas/components/canvas-assistant-panel.tsx b/web/src/app/(user)/canvas/components/canvas-assistant-panel.tsx index 1bf7713..818e035 100644 --- a/web/src/app/(user)/canvas/components/canvas-assistant-panel.tsx +++ b/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[]) => { diff --git a/web/src/services/api/image.ts b/web/src/services/api/image.ts index 6a0d364..bcb4f5b 100644 --- a/web/src/services/api/image.ts +++ b/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>; @@ -263,6 +264,108 @@ function parseToolResponse(payload: ResponseApiPayload): ToolResponseResult { return { content, toolCalls }; } +function isRecord(value: unknown): value is Record { + 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; + 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, onDelta?: (text: string) => void): Promise { + 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( - aiApiUrl(requestConfig, "/responses"), - { - model: requestConfig.model, - input: toResponseInput(withSystemMessage(requestConfig, messages)), - }, - { - 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); + const answer = (await requestStreamingResponse(requestConfig, { + model: requestConfig.model, + input: toResponseInput(withSystemMessage(requestConfig, messages)), + }, onDelta)).content || "没有返回内容"; + if (answer === "没有返回内容") 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 { +export async function requestToolResponse(config: AiConfig, messages: ResponseInputMessage[], tools: ResponseFunctionTool[], toolChoice: ToolChoice = "auto", onDelta?: (text: string) => void): Promise { const requestConfig = resolveModelRequestConfig(config, config.model || config.textModel); try { - const response = await axios.post( - aiApiUrl(requestConfig, "/responses"), - { - model: requestConfig.model, - input: toResponseInput(withSystemMessage(requestConfig, messages)), - tools: tools.map(toResponseTool), - tool_choice: toolChoice, - 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); + return await requestStreamingResponse(requestConfig, { + model: requestConfig.model, + input: toResponseInput(withSystemMessage(requestConfig, messages)), + tools: tools.map(toResponseTool), + tool_choice: toolChoice, + parallel_tool_calls: false, + }, onDelta); } catch (error) { throw new Error(readAxiosError(error, "请求失败")); }