From ef4d310e17389a290668a42fe9ccf51b36b76560 Mon Sep 17 00:00:00 2001 From: qnsh Date: Sat, 29 Aug 2026 13:28:44 +0800 Subject: [PATCH] update preview text add process --- openai/openai_text_node.py | 52 ++++++++-- utils/skill_utils.py | 167 ++++++++++++++++++++--------- web/js/preview_api_result.js | 196 +++++++++++++++++++++++++++++++---- 3 files changed, 337 insertions(+), 78 deletions(-) diff --git a/openai/openai_text_node.py b/openai/openai_text_node.py index d96dc38..85ca4f4 100644 --- a/openai/openai_text_node.py +++ b/openai/openai_text_node.py @@ -48,6 +48,7 @@ class _TextStreamSink: self.client_id = getattr(server, "client_id", None) self.seq = 0 self.last_activity = None + self.current_round = 0 def _send(self, phase, **extra): PromptServer.instance.send_sync( @@ -71,10 +72,45 @@ class _TextStreamSink: if value: self._send("delta", delta=value) - def activity(self, kind): - if kind and kind != self.last_activity: - self.last_activity = kind - self._send("activity", activity=kind) + def activity(self, kind, detail=None): + marker = (kind, detail or "") + if kind and marker != self.last_activity: + self.last_activity = marker + extra = {"activity": kind} + if detail: + extra["detail"] = detail + self._send("activity", **extra) + + def round_start(self, round_index, tools_enabled=True): + self.current_round = int(round_index) + self._send("round_start", round=int(round_index), tools_enabled=bool(tools_enabled)) + + def candidate_delta(self, value, round_index=None): + if value: + extra = {"delta": value} + extra["round"] = int(self.current_round if round_index is None else round_index) + self._send("candidate_delta", **extra) + + def tool_call_start(self, call_id, name, path=None, round_index=None): + extra = {"call_id": call_id, "tool": name} + if path: + extra["path"] = path + extra["round"] = int(self.current_round if round_index is None else round_index) + self._send("tool_call_start", **extra) + + def tool_call_end(self, call_id, name, status="success", path=None, round_index=None): + extra = {"call_id": call_id, "tool": name, "status": status} + if path: + extra["path"] = path + extra["round"] = int(self.current_round if round_index is None else round_index) + self._send("tool_call_end", **extra) + + def round_end(self, round_index, has_tool_calls): + self._send( + "round_end", + round=int(round_index), + has_tool_calls=bool(has_tool_calls), + ) def end(self, value): self._send("end", text=value) @@ -147,7 +183,7 @@ class OpenAITextAPI(io.ComfyNode): io.Boolean.Input(id="clear_history", default=False), io.Boolean.Input( id="stream", - default=False, + default=True, tooltip=( "If true, the model response is streamed to the client as it is " "generated using server-sent events (SSE)." @@ -357,8 +393,12 @@ class OpenAITextAPI(io.ComfyNode): persist_context=persist_context, stream=stream_enabled, post_stream=post_openai_sse_events, - on_delta=sink.delta if sink is not None else None, + on_delta=sink.candidate_delta if sink is not None else None, on_activity=sink.activity if sink is not None else None, + on_round_start=sink.round_start if sink is not None else None, + on_round_end=sink.round_end if sink is not None else None, + on_tool_call_start=sink.tool_call_start if sink is not None else None, + on_tool_call_end=sink.tool_call_end if sink is not None else None, ) except Exception as exc: if sink is not None: diff --git a/utils/skill_utils.py b/utils/skill_utils.py index a79e286..bf02cf5 100644 --- a/utils/skill_utils.py +++ b/utils/skill_utils.py @@ -406,6 +406,7 @@ def load_skill(snapshot): LOAD_SKILL_TOOL = "load_skill" READ_SKILL_FILE_TOOL = "read_skill_file" +READ_TOOL = "read" INVOCATION_POLICY = "required_once_per_session" EXECUTION_MODE = "pi_skill_agent" @@ -478,14 +479,30 @@ def _function_definition(name): "additionalProperties": False, }, } + if name == READ_TOOL: + definition = _function_definition(READ_SKILL_FILE_TOOL) + definition["name"] = READ_TOOL + definition["description"] = ( + "Read a text resource from the loaded Skill manifest. " + "Use offset (1-based line) and limit for large files." + ) + definition["parameters"]["properties"].update({ + "offset": { + "type": "integer", "minimum": 1, + "description": "1-based line number to start reading from.", + }, + "limit": { + "type": "integer", "minimum": 1, "maximum": 2000, + "description": "Maximum number of lines to return.", + }, + }) + return definition raise ValueError(f"Unknown Skill tool definition: {name}") -def skill_tool_definitions(protocol="openai-completions", loaded=False): +def skill_tool_definitions(protocol="openai-completions", loaded=False, read_tool=READ_SKILL_FILE_TOOL): """Return the provider schema for the only tool registered in this phase.""" - definition = _function_definition( - READ_SKILL_FILE_TOOL if loaded else LOAD_SKILL_TOOL - ) + definition = _function_definition((read_tool if loaded else LOAD_SKILL_TOOL)) if protocol == "openai-completions": return [{"type": "function", "function": definition}] if protocol == "openai-responses": @@ -692,26 +709,42 @@ class ReadOnlySkillRuntime: self.loaded = True return output - if name != READ_SKILL_FILE_TOOL: + if name not in {READ_SKILL_FILE_TOOL, READ_TOOL}: raise SkillExecutionError( "pi_skill_tool_call_invalid", f"Unknown Skill tool: {name}" ) if not self.loaded: raise SkillExecutionError( - "skill_not_loaded", "read_skill_file requires load_skill first" + "skill_not_loaded", "read requires load_skill first" ) - if set(arguments) != {"path"} or not isinstance(arguments.get("path"), str): + allowed = {"path"} if name == READ_SKILL_FILE_TOOL else {"path", "offset", "limit"} + if not set(arguments).issubset(allowed) or "path" not in arguments or not isinstance(arguments.get("path"), str): raise SkillExecutionError( "pi_skill_tool_call_invalid", - "read_skill_file requires only the string 'path' argument", + f"{name} requires a string 'path' argument", ) + if name == READ_TOOL: + for key in ("offset", "limit"): + if key in arguments and ( + isinstance(arguments[key], bool) + or not isinstance(arguments[key], int) + or arguments[key] < 1 + ): + raise SkillExecutionError( + "pi_skill_tool_call_invalid", f"{key} must be a positive integer" + ) + if arguments.get("limit", 2000) > 2000: + raise SkillExecutionError( + "pi_skill_tool_call_invalid", "limit must not exceed 2000" + ) path = arguments["path"] if path == "SKILL.md": raise SkillExecutionError( "pi_skill_tool_call_invalid", "SKILL.md can only be loaded with load_skill" ) - if path in self.cache: - return self.cache[path] + cache_key = path if name == READ_SKILL_FILE_TOOL else (path, arguments.get("offset", 1), arguments.get("limit")) + if cache_key in self.cache: + return self.cache[cache_key] try: result = read_skill_reference(self.snapshot, path) except ValueError as exc: @@ -720,8 +753,25 @@ class ReadOnlySkillRuntime: raise SkillExecutionError( "pi_skill_tool_call_invalid", f"Resource is not in the manifest: {path}" ) + if name == READ_TOOL: + lines = result.get("content", "").splitlines(keepends=True) + offset = arguments.get("offset", 1) + limit = arguments.get("limit", 2000) + start = offset - 1 + if start >= len(lines) and lines: + raise SkillExecutionError( + "pi_skill_tool_call_invalid", f"offset {offset} is beyond end of file" + ) + selected = lines[start : start + limit] + result = dict(result) + result["content"] = "".join(selected) + result["offset"] = offset + result["line_count"] = len(selected) + result["total_lines"] = len(lines) + if start + len(selected) < len(lines): + result["next_offset"] = start + len(selected) + 1 self._account(result) - self.cache[path] = result + self.cache[cache_key] = result return result @@ -1103,7 +1153,7 @@ class SkillToolRegistry: @staticmethod def allowed_name(loaded): - return READ_SKILL_FILE_TOOL if loaded else LOAD_SKILL_TOOL + return READ_TOOL if loaded else LOAD_SKILL_TOOL @classmethod def validate(cls, call, loaded, seen): @@ -1113,10 +1163,12 @@ class SkillToolRegistry: raise SkillExecutionError( "pi_skill_tool_call_invalid", "Tool call_id is missing or duplicated" ) - if name != cls.allowed_name(loaded): + allowed = cls.allowed_name(loaded) + aliases = {READ_TOOL, READ_SKILL_FILE_TOOL} if loaded else {LOAD_SKILL_TOOL} + if name not in aliases: raise SkillExecutionError( "pi_skill_tool_call_invalid", - f"Tool is not registered in the current phase: {name!r}", + f"Tool is not registered in the current phase: {name!r}", ) return call_id, name, _parse_args(call.get("arguments")) @@ -1130,7 +1182,11 @@ class PiSkillAgentLoop: self.session = session self.limits = limits - def run(self, payload, trace, stream=False, on_delta=None, on_activity=None): + def run( + self, payload, trace, stream=False, on_delta=None, on_activity=None, + on_round_start=None, on_round_end=None, on_tool_call_start=None, + on_tool_call_end=None, + ): field = self.adapter.history_field history = deepcopy(payload[field]) base_payload = { @@ -1147,14 +1203,29 @@ class PiSkillAgentLoop: trace["tool_choice_mode"] = "provider_default" seen = set() + final_candidate = [] for round_index in range(self.limits["max_tool_rounds"] + 1): - tools = skill_tool_definitions(self.adapter.protocol, loaded=loaded) + tools = skill_tool_definitions( + self.adapter.protocol, + loaded=loaded, + read_tool=READ_TOOL if stream else READ_SKILL_FILE_TOOL, + ) + if on_round_start is not None: + on_round_start(round_index, bool(tools)) + candidate = [] + + def collect_candidate(value, *_ignored): + candidate.append(value) + if on_delta is not None: + on_delta(value) + data = self.adapter.request( base_payload, history, tools, stream=stream, on_activity=on_activity, + on_delta=collect_candidate if stream else None, ) normalized = self.adapter.normalize(data) self.session.append( @@ -1177,47 +1248,22 @@ class PiSkillAgentLoop: "pi_skill_response_invalid", "Provider response contains no final text", ) + self.adapter.append_native_items(history, normalized) + if stream and on_round_end is not None: + on_round_end(round_index, False) if stream: - # A tools-enabled response cannot be known to be the formal - # answer until its stream has ended. Keep that candidate - # out of history and use a tools-free final phase so the UI - # can safely display its first text delta immediately. - if on_activity is not None: - on_activity("generating") - self.session.append( - "final_phase_started", protocol=self.adapter.protocol - ) - final_data = self.adapter.final_request( - base_payload, - history, - stream=True, - on_delta=on_delta, - on_activity=on_activity, - ) - final_normalized = self.adapter.normalize(final_data) - if final_normalized.tool_calls: - raise SkillExecutionError( - "pi_skill_response_invalid", - "Provider returned a tool call after Skill tools were removed", - ) - if final_normalized.final_text is None: - raise SkillExecutionError( - "pi_skill_response_invalid", - "Provider final phase contains no formal text", - ) - self.adapter.append_native_items(history, final_normalized) - normalized = final_normalized - else: - self.adapter.append_native_items(history, normalized) + final_candidate = candidate trace["skill_loaded"] = True if trace["load_source"] == "none": trace["load_source"] = "tool_call" self.session.commit( history, self.snapshot, self.adapter.protocol ) - return normalized.final_text + return "".join(final_candidate) or normalized.final_text self.adapter.append_native_items(history, normalized) + if stream and on_round_end is not None: + on_round_end(round_index, True) if round_index >= self.limits["max_tool_rounds"]: raise SkillExecutionError( "pi_skill_tool_limit_exceeded", "Skill tool round limit exceeded" @@ -1239,6 +1285,10 @@ class PiSkillAgentLoop: accepted=False, error={"code": exc.code, "message": exc.message}, ) + if on_tool_call_end is not None: + on_tool_call_end( + call_id, name, "error", arguments.get("path"), round_index + ) raise seen.add(call_id) planned.append((call_id, name, arguments)) @@ -1261,6 +1311,10 @@ class PiSkillAgentLoop: on_activity( "loading_skill" if name == LOAD_SKILL_TOOL else "reading_skill" ) + if on_tool_call_start is not None: + on_tool_call_start( + call_id, name, arguments.get("path"), round_index + ) try: output = runtime.execute(name, arguments) except SkillExecutionError as exc: @@ -1274,6 +1328,10 @@ class PiSkillAgentLoop: "tool_call_settled", call_id=call_id, name=name, outcome="success", path=output.get("path"), output=output, ) + if on_tool_call_end is not None: + on_tool_call_end( + call_id, name, "success", output.get("path"), round_index + ) wire_result = self.adapter.serialize_tool_result(call_id, output) history.append(wire_result) self.session.append( @@ -1381,6 +1439,8 @@ class SkillExecutionRouter: protocol, post_json, endpoint, headers, timeout, proxies, payload, snapshot, session_key, persist_context=True, trace=None, stream=False, post_stream=None, on_delta=None, on_activity=None, + on_round_start=None, on_round_end=None, on_tool_call_start=None, + on_tool_call_end=None, ): config = get_skill_config() if not config.get("allow_call", False): @@ -1447,6 +1507,10 @@ class SkillExecutionRouter: stream=stream, on_delta=on_delta, on_activity=on_activity, + on_round_start=on_round_start, + on_round_end=on_round_end, + on_tool_call_start=on_tool_call_start, + on_tool_call_end=on_tool_call_end, ) return result, active_trace, session.conversation(snapshot) except Exception as exc: @@ -1504,7 +1568,8 @@ class SkillRequestContext: def execute( self, post_json, endpoint, headers, timeout, proxies, payload, session_key, persist_context=True, stream=False, post_stream=None, - on_delta=None, on_activity=None, + on_delta=None, on_activity=None, on_round_start=None, on_round_end=None, + on_tool_call_start=None, on_tool_call_end=None, ): if not self.enabled: raise RuntimeError("Skill execution requires selected skill_options") @@ -1525,6 +1590,10 @@ class SkillRequestContext: post_stream=post_stream, on_delta=on_delta, on_activity=on_activity, + on_round_start=on_round_start, + on_round_end=on_round_end, + on_tool_call_start=on_tool_call_start, + on_tool_call_end=on_tool_call_end, ) return result, conversation except Exception as exc: diff --git a/web/js/preview_api_result.js b/web/js/preview_api_result.js index b4fd2c4..9254846 100644 --- a/web/js/preview_api_result.js +++ b/web/js/preview_api_result.js @@ -20,6 +20,14 @@ const MESSAGES = { reading_skill: "Reading Skill files…", reasoning: "Thinking…", generating: "Generating…", + drafting: "Model note…", + toolRunning: "Tool running…", + tool_running: "Tool running…", + promoting: "Preparing answer…", + activityExpand: "Show activity", + activityCollapse: "Hide activity", + processTitle: "Process & status", + intermediate: "Model note", displaying: "Displaying result…", complete: "Complete", copy: "Copy", @@ -37,6 +45,14 @@ const MESSAGES = { reading_skill: "正在读取 Skill 文件…", reasoning: "正在思考…", generating: "正在生成…", + drafting: "模型中间说明…", + toolRunning: "工具执行中…", + tool_running: "工具执行中…", + promoting: "正在整理答案…", + activityExpand: "展开过程", + activityCollapse: "收起过程", + processTitle: "过程与状态", + intermediate: "模型中间说明", displaying: "正在显示结果…", complete: "完成", copy: "复制", @@ -157,7 +173,7 @@ function renderNow(state) { state.host.dataset.hasContent = raw ? "true" : "false"; if (!raw) { state.content.replaceChildren(); - if (!["waiting", "loading_skill", "reading_skill", "reasoning", "error"].includes(state.statusKey)) { + if (!["waiting", "loading_skill", "reading_skill", "reasoning", "drafting", "tool_running", "promoting", "generating", "displaying", "error"].includes(state.statusKey)) { const placeholder = document.createElement("div"); placeholder.className = "empty"; placeholder.textContent = message("empty"); @@ -199,17 +215,64 @@ function setStatus(state, key, detail = "") { state.statusDetail = detail; const visible = [ "waiting", "loading_skill", "reading_skill", "reasoning", - "generating", "displaying", "error", + "drafting", "tool_running", "promoting", "generating", "displaying", "complete", "error", ].includes(key); state.host.dataset.state = key; state.status.textContent = key === "error" && detail ? `${message("error")}: ${detail}` : message(key); state.status.hidden = !visible; + // Status is rendered in the dedicated lower process box, so update its + // visibility whenever the state changes. + renderActivity(state); +} + +function appendActivity(state, entry) { + state.activityLog.push({ ...entry, time: Date.now() }); + if (state.activityLog.length > 20) state.activityLog.splice(0, state.activityLog.length - 20); + state.activityExpanded = true; + renderActivity(state); +} + +function renderActivity(state) { + if (!state.activityPanel) return; + state.activityPanel.replaceChildren(); + const latest = state.currentActivity; + if (latest) { + const current = document.createElement("div"); + current.className = "activity-current"; + current.textContent = latest.detail ? `${latest.label} · ${latest.detail}` : latest.label; + state.activityPanel.append(current); + } + const log = document.createElement("div"); + log.className = "activity-log"; + const entries = state.activityExpanded ? state.activityLog : state.activityLog.slice(-2); + for (const entry of entries) { + const row = document.createElement("div"); + row.className = `activity-entry activity-${entry.type || "info"}`; + row.textContent = entry.text || ""; + log.append(row); + } + if (state.candidateText) { + const candidate = document.createElement("div"); + candidate.className = "activity-candidate"; + candidate.textContent = `${message("intermediate")}: ${state.candidateText}`; + log.append(candidate); + } + state.activityPanel.append(log); + // Keep the process box visible for the complete state as well, so the + // final status and activity history remain available after rendering. + const statusVisible = !state.status.hidden && state.statusKey !== "ready"; + state.activityWrap.hidden = !latest && !state.activityLog.length && !state.candidateText && !statusVisible; + state.activityToggle.hidden = state.activityLog.length < 3 && !state.candidateText; + state.activityToggle.textContent = state.activityExpanded + ? message("activityCollapse") : message("activityExpand"); } function updateCopyAvailability(state) { - state.copyButton.disabled = !(state.finalText || state.receivedText || state.displayedText); + // Only confirmed answer text is copyable. Intermediate candidate text, + // reasoning and activity records must never leak through the copy action. + state.copyButton.disabled = !(state.answerText || state.finalText); } function setCopyState(state, copyState) { @@ -239,7 +302,7 @@ function showCopyToast(success) { } async function copyRawText(state) { - const text = state.sourceEnded ? state.finalText : (state.receivedText || state.displayedText); + const text = state.answerText || state.finalText; if (!text) return; try { if (navigator.clipboard?.writeText && window.isSecureContext) { @@ -293,8 +356,10 @@ function finishTypingIfReady(state) { if (state.terminalError) { setStatus(state, "error", state.terminalError); } else { + state.currentActivity = null; setStatus(state, "complete"); } + renderActivity(state); updateCopyAvailability(state); scheduleRender(state, true); return true; @@ -399,17 +464,30 @@ function createPreview(node) {
- -
- + + +
`; const state = { displayedText: "", + answerText: "", receivedText: "", finalText: "", pendingUnits: [], @@ -491,16 +569,30 @@ function createPreview(node) { copyTimer: null, statusKey: "ready", statusDetail: "", + candidateText: "", + activityLog: [], + currentActivity: null, + activityExpanded: false, root, host: shadow.querySelector(".preview"), content: shadow.querySelector(".content"), - status: shadow.querySelector(".status"), + status: shadow.querySelector(".activity-wrap .status"), copyButton: shadow.querySelector("button"), + activityWrap: shadow.querySelector(".activity-wrap"), + activityTitle: shadow.querySelector(".activity-title"), + activityPanel: shadow.querySelector(".activity-panel"), + activityToggle: shadow.querySelector(".activity-toggle"), }; setStatus(state, "ready"); + state.activityTitle.textContent = message("processTitle"); setCopyState(state, "idle"); updateCopyAvailability(state); state.copyButton.addEventListener("click", () => copyRawText(state)); + state.activityToggle.addEventListener("click", () => { + state.activityExpanded = !state.activityExpanded; + renderActivity(state); + }); + renderActivity(state); root.addEventListener("pointerdown", event => event.stopPropagation()); root.addEventListener("wheel", event => event.stopPropagation(), { passive: true }); @@ -536,6 +628,7 @@ function setFinalText(node, value) { const unchanged = state.displayedText === text && !state.streaming; stopTyping(state); state.displayedText = text; + state.answerText = text; state.receivedText = text; state.finalText = text; state.pendingUnits = []; @@ -577,13 +670,21 @@ function receiveStreamEvent(event) { state.promptId = data.prompt_id ?? null; state.lastSeq = Number.isFinite(sequence) ? sequence : -1; state.displayedText = ""; + state.answerText = ""; state.receivedText = ""; state.finalText = ""; + state.candidateText = ""; + state.activityLog = []; + state.currentActivity = null; + state.activityExpanded = false; + state.currentRound = 0; + state.roundHasToolCall = false; state.pendingUnits = []; state.sourceEnded = false; state.terminalError = ""; state.streaming = true; setStatus(state, "waiting"); + renderActivity(state); updateCopyAvailability(state); scheduleRender(state, true); continue; @@ -592,7 +693,51 @@ function receiveStreamEvent(event) { if (Number.isFinite(sequence) && sequence <= state.lastSeq) continue; if (Number.isFinite(sequence)) state.lastSeq = sequence; - if (data.phase === "activity") { + if (data.phase === "round_start") { + state.currentRound = Number(data.round) || 0; + state.roundHasToolCall = false; + state.candidateText = ""; + state.currentActivity = { label: message("waiting"), detail: `Round ${state.currentRound + 1}` }; + appendActivity(state, { type: "round", text: `Round ${state.currentRound + 1}` }); + setStatus(state, "waiting"); + } else if (data.phase === "candidate_delta") { + state.candidateText += String(data.delta || ""); + state.currentActivity = { label: message("drafting"), detail: `Round ${state.currentRound + 1}` }; + setStatus(state, "drafting"); + renderActivity(state); + } else if (data.phase === "tool_call_start") { + state.roundHasToolCall = true; + const detail = data.path ? `${data.tool} · ${data.path}` : data.tool; + state.currentActivity = { label: message("toolRunning"), detail }; + appendActivity(state, { type: "tool", text: `${message("toolRunning")} ${detail}` }); + setStatus(state, "tool_running"); + } else if (data.phase === "tool_call_end") { + const detail = data.path ? `${data.tool} · ${data.path}` : data.tool; + appendActivity(state, { type: "tool", text: `${detail} · ${data.status || "success"}` }); + renderActivity(state); + } else if (data.phase === "round_end") { + const hasTools = Boolean(data.has_tool_calls); + state.roundHasToolCall = hasTools; + if (hasTools) { + state.candidateText = ""; + state.currentActivity = { label: message("toolRunning"), detail: `Round ${state.currentRound + 1} complete` }; + } else { + state.answerText = state.candidateText; + state.receivedText = state.answerText; + state.finalText = state.answerText; + state.candidateText = ""; + state.currentActivity = { label: message("generating"), detail: "" }; + setStatus(state, state.answerText ? "promoting" : "generating"); + if (state.answerText) { + state.pendingUnits = splitTextUnits(state.answerText); + state.displayedText = ""; + state.sourceEnded = false; + startTyping(state); + } + } + renderActivity(state); + scheduleRender(state, true); + } else if (data.phase === "activity") { const activity = [ "loading_skill", "reading_skill", "reasoning", "generating", ].includes(data.activity) ? data.activity : null; @@ -608,6 +753,10 @@ function receiveStreamEvent(event) { setStatus(state, "displaying"); } enqueueFinalText(state, data.text ?? state.receivedText); + state.answerText = String(data.text ?? state.receivedText); + state.candidateText = ""; + state.currentActivity = null; + renderActivity(state); } else if (data.phase === "error") { state.sourceEnded = true; state.finalText = state.receivedText; @@ -625,6 +774,7 @@ function refreshLocale() { for (const node of app.graph?._nodes || []) { const state = node?._ycyyApiResultPreview; if (!state) continue; + if (state.activityTitle) state.activityTitle.textContent = message("processTitle"); setCopyState(state, state.copyButton.dataset.copyState || "idle"); setStatus(state, state.statusKey, state.statusDetail); scheduleRender(state);