jianf/HomeAgent
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/HomeAgent.git synced 2026-10-03 15:53:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(streaming): token-level delta events + interrupt for CLI/WebUI/GUI

Browse Source 
Expose the LLM token-level streaming deltas (EventReasoningDelta /
EventContentDelta) to every client channel and add user-initiated
interrupt (cancel generation / send interrupt message) to all three
frontends, preserving the existing interrupt-injection semantics.

SDK/events:
  - EventReasoningDelta, EventContentDelta constants exported in the
    public/internal SDK event alias tables.

CLI plugin:
  - handleChat subscribes to both delta events and forwards
    reasoning_delta / content_delta JSON frames (channel-filtered);
    aggregated reasoning/tool_call/response frames still fire as before.
  - New /stop (alias /interrupt) builtin injects an interrupt via
    InjectInterrupt(cliSource, cliChannel) - matches interceptLoop
    semantics: cancels an active stream and re-injects the message as
    a [中断消息] for a restarted turn; with no active LLM it behaves
    as a plain input.

Waiter client (line mode + TUI):
  - streamRender accumulates delta chunks and redraws the current line;
    a reset frame (stream abandoned, e.g. user interrupt) flushes the
    partial buffer so the next turn does not concatenate onto stale
    content. Aggregated frames terminate the delta line and render the
    final text (old servers without deltas behave exactly as before).
  - TUI merges content_delta into the in-flight agent message and seals
    it (final flag) on response/tool_call/error so subsequent deltas
    never append to a finished message.

WebUI:
  - SSE handler subscribes to the two delta events but does NOT record
    them into the replay ring - reconnection replays only aggregated
    events (the final truth), avoiding duplicate delta accumulation.
  - POST /api/v1/chat/interrupt calls InjectInterrupt(webui, webui)
    with optional message; fronted by a Stop button shown only while
    a generation is in flight.

dashboard.html / GUI app.js:
  - Stop button next to Send (hidden until chatLoading); interruptChat
    POSTs /chat/interrupt. Delta listeners append incrementally;
    agent_output (aggregated) now replaces (not appends) the in-flight
    content and marks _final; reset frames finalize the partial message.

process.go:
  - chatStreamWithFallback preserves the context.Canceled/
    DeadlineExceeded contract: a user interrupt returns the canceled
    error (never a partial-content success) so the existing continue
    branch restarts the turn with the [中断消息]. A reset
    EventContentDelta is published so connected clients drop stale
    partial renderings before the new turn begins.

Verified: /stop 'msg' via waiter triggers 'interrupt from cli/cli' in
interceptLoop; unit TestChatStreamCancelPreservesInterrupt confirms the
canceled error propagates instead of being swallowed.
This commit is contained in:
JianFeeeee
2026-08-25 10:50:37 +08:00
parent 28a6d3f09c
commit 061d2ae320
11 changed files with 423 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
internal/agent/core/process.go
View File

@ -315,11 +315,28 @@ func chatStreamWithFallback(ctx context.Context, p agentAPI.Provider, req *agent
}
resp, accErr := accumulateStream(ctx, ch, a)
// 中断/超时取消必须保持取消语义传给调用方（与原 Chat() 行为一致：
// 被 cancel 时丢弃已收内容返回 err），让 process() 的 continue 分支
// 重启轮次并以 [中断消息] 注入打断内容。绝不能把部分内容当成功返回，
// 否则用户打断会被无视、继续执行工具/输出。
if errors.Is(accErr, context.Canceled) || errors.Is(accErr, context.DeadlineExceeded) {
// 通知客户端：本轮流式作废，清空 delta 累积并定格已显示内容
if a != nil {
a.publishEvent(events.EventContentDelta, map[string]interface{}{
"content": "",
"channel": a.currentOutputChannel,
"reset": true,
})
}
return resp, accErr
}
if accErr == nil {
return resp, nil
}
// 流中途错误：若已累积到内容则返回部分结果，否则回退非流式
// 其他错误（网络中断等）：已累积到实质内容则返回部分结果，否则回退非流式
if resp != nil && (resp.Content != "" || len(resp.ToolCalls) > 0) {
log.Printf("[agent] stream interrupted mid-way (%v), returning partial result", accErr)
return resp, nil

69
internal/agent/core/stream_accumulate_test.go Normal file
View File

@ -0,0 +1,69 @@
package core
import (
"context"
"testing"
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
)
// 验证流式 tool call 分片累积：模拟 llmsproxy/big-pickle 的分片序列
func TestAccumulateStreamToolCalls(t *testing.T) {
ch := make(chan agentAPI.StreamChunk, 10)
go func() {
// 分片1: name + id + arguments 开头
ch <- agentAPI.StreamChunk{ToolCalls: []agentAPI.ToolCall{
{ID: "call_1", Name: "cmd_run", RawArguments: "{\""},
}}
// 分片2-3: 只有 arguments 分片
ch <- agentAPI.StreamChunk{ToolCalls: []agentAPI.ToolCall{
{RawArguments: "command\""},
}}
ch <- agentAPI.StreamChunk{ToolCalls: []agentAPI.ToolCall{
{RawArguments: ":\"date\"}"},
}}
ch <- agentAPI.StreamChunk{Done: true, FinishReason: "tool_calls"}
close(ch)
}()
resp, err := accumulateStream(context.Background(), ch, nil)
if err != nil {
t.Fatalf("accumulateStream: %v", err)
}
if len(resp.ToolCalls) != 1 {
t.Fatalf("want 1 tool call, got %d", len(resp.ToolCalls))
}
tc := resp.ToolCalls[0]
if tc.Name != "cmd_run" || tc.ID != "call_1" {
t.Fatalf("bad name/id: %s/%s", tc.ID, tc.Name)
}
cmd, _ := tc.Arguments["command"].(string)
if cmd != "date" {
t.Fatalf("arguments not merged, got: %v", tc.Arguments)
}
if resp.FinishReason != "tool_calls" {
t.Fatalf("finish reason: %q", resp.FinishReason)
}
}
// 验证 content/reasoning 增量累积
func TestAccumulateStreamContent(t *testing.T) {
ch := make(chan agentAPI.StreamChunk, 5)
go func() {
ch <- agentAPI.StreamChunk{ReasoningContent: "think "}
ch <- agentAPI.StreamChunk{Content: "你"}
ch <- agentAPI.StreamChunk{Content: "好"}
ch <- agentAPI.StreamChunk{Done: true, FinishReason: "stop"}
close(ch)
}()
resp, err := accumulateStream(context.Background(), ch, nil)
if err != nil {
t.Fatalf("accumulateStream: %v", err)
}
if resp.Content != "你好" {
t.Fatalf("content: %q", resp.Content)
}
if resp.ReasoningContent != "think " {
t.Fatalf("reasoning: %q", resp.ReasoningContent)
}
}

58
internal/plugins/cli/plugin.go
View File

@ -211,8 +211,33 @@ func (p *Plugin) handleChat(w *connWriter, line string, s *sdk.PluginSDK) {
"result": truncateOneLine(result, 160),
})
})
// token 级流式增量帧：客户端可选订做逐 token 渲染。
// 旧客户端收到未知 type 会忽略；聚合 reasoning/response 帧仍照常发送，
// 保证旧/新客户端最终都能看到完整文本。
unsubReasoningDelta := s.Subscribe(sdk.EventReasoningDelta, func(evt *sdk.Event) {
if ch, _ := evt.Payload["channel"].(string); ch != cliChannel {
return
}
content, _ := evt.Payload["content"].(string)
if content == "" {
return
}
w.writeLine(map[string]interface{}{"type": "reasoning_delta", "content": content})
})
unsubContentDelta := s.Subscribe(sdk.EventContentDelta, func(evt *sdk.Event) {
if ch, _ := evt.Payload["channel"].(string); ch != cliChannel {
return
}
content, _ := evt.Payload["content"].(string)
if content == "" {
return
}
w.writeLine(map[string]interface{}{"type": "content_delta", "content": content})
})
defer unsubReasoning()
defer unsubToolCall()
defer unsubReasoningDelta()
defer unsubContentDelta()
resp := s.InjectTextSync(cliSource, cliChannel, line)
if resp != nil {
@ -260,6 +285,8 @@ func (p *Plugin) handleBuiltin(conn net.Conn, line string, s *sdk.PluginSDK) boo
switch parts[0] {
case "/help":
p.cmdHelp(conn)
case "/stop", "/interrupt":
p.cmdInterrupt(conn, parts, s)
case "/status":
p.cmdStatus(conn, s)
case "/kernel":
@ -285,6 +312,7 @@ func (p *Plugin) cmdHelp(conn net.Conn) {
"type": "response",
"content": `内置命令（直接对话内核，不依赖网络）:
/help 显示此帮助
/stop [消息] 停止当前生成/发送中断消息（别名 /interrupt）
/status 系统运行状态
/kernel 内核状态（插件、工具、LLM、记忆）
/settings 列出所有配置
@ -304,6 +332,36 @@ func (p *Plugin) cmdHelp(conn net.Conn) {
})
}
// ======== /stop ========
// cmdInterrupt 注入用户中断。核心拦截语义（interceptLoop）：
// - 有 LLM 在跑：cancelLLM 取消当前流式请求，中断入队，process() 以
// [中断消息] 重启轮次（模型看到被打断的上下文 + 用户新输入）；
// - 无 LLM 在跑：作为普通输入处理（等同发了一条消息）。
//
// 可选附带消息：/stop 换个话题（空参数 = 纯取消）。
func (p *Plugin) cmdInterrupt(conn net.Conn, parts []string, s *sdk.PluginSDK) {
msg := strings.TrimSpace(strings.TrimPrefix(line2(parts), "/stop"))
if alias := strings.TrimSpace(strings.TrimPrefix(line2(parts), "/interrupt")); alias != "" {
msg = alias
}
s.InjectInterrupt(cliSource, cliChannel, "text", map[string]interface{}{
"content": msg,
})
writeLine(conn, map[string]interface{}{
"type": "response",
"content": "已发送中断信号",
})
}
// line2 将命令行参数重组为原始字符串（保留词间空格，去掉首 token）。
func line2(parts []string) string {
if len(parts) < 2 {
return ""
}
return strings.Join(parts[1:], " ")
}
// ======== /status ========
func (p *Plugin) cmdStatus(conn net.Conn, s *sdk.PluginSDK) {

96
internal/plugins/webui/dashboard.html
View File

@ -2774,6 +2774,9 @@ background:
'<input id="chat-input" placeholder="' +
__("输入消息...", "Type a message...") +
'" onkeydown="if(event.key==\'Enter\')sendChat()">' +
'<button class="btn" onclick="interruptChat()" id="chat-stop-btn" style="display:none;background:var(--danger, #d1383d);color:#fff">' +
__("停止", "Stop") +
"</button>" +
'<button class="btn btn-primary" onclick="sendChat()" id="chat-send-btn">' +
__("发送", "Send") +
"</button>" +
@ -3633,6 +3636,7 @@ background:
async function sendChat() {
var inp = document.getElementById("chat-input");
var btn = document.getElementById("chat-send-btn");
var stopBtn = document.getElementById("chat-stop-btn");
var text = inp.value.trim();
if (!text || state.chatLoading) return;
state.chatStick = true;
@ -3644,6 +3648,7 @@ background:
state.chatStage = __("等待AI回复...", "Waiting for AI...");
btn.disabled = true;
btn.textContent = "";
if (stopBtn) stopBtn.style.display = ""; // 生成期间可停止
rerenderChat(true);
// 触发式 POST：短超时仅确认受理；回复靠 SSE 流式渲染（对齐 GUI 行为）。
try {
@ -3715,10 +3720,26 @@ background:
state.chatStage = "";
btn.disabled = false;
btn.textContent = __("发送", "Send");
var sb2 = document.getElementById("chat-stop-btn");
if (sb2) sb2.style.display = "none"; // 回复完成/失败，隐藏停止按钮
rerenderChat(true);
}
}
// 停止生成 / 发送中断消息。核心拦截语义：有 LLM 在跑则取消当前
// 请求并以 [中断消息] 重启轮次；无则在跑则作为普通消息处理。
async function interruptChat() {
try {
await api("/chat/interrupt", {
method: "POST",
body: JSON.stringify({}),
});
toast(__("已发送中断信号", "Interrupt signal sent"));
} catch (e) {
toast(__("中断失败: ", "Interrupt failed: ") + e.message, true);
}
}
async function queryMemoryChat() {
var q = document.getElementById("mem-query")?.value;
var r = document.getElementById("mem-result-chat");
@ -4119,8 +4140,11 @@ background:
lastM2.role === "assistant" &&
!lastM2._final
) {
// 聚合最终响应：覆盖 delta 累积的中间内容（以聚合为准，
// 含 stage 插件改写后的最终文本），并置 final 结束本轮流式。
lastM2._grow = true;
lastM2.content += p.content;
lastM2.content = p.content;
lastM2._final = true;
rerenderChat();
return;
}
@ -4137,12 +4161,50 @@ background:
content: p.content,
_streaming: true,
_grow: true,
_final: true,
});
rerenderChat();
} catch (ex) {
console.error("[SSE] agent_output error", ex);
}
});
// token 级流式增量：逐块追加到当前回复内容（流式生成中）；
// reset 帧表示轮次作废（用户中断）：定格已显示的部分内容，置 final。
es.addEventListener("content_delta", function (e) {
try {
var ev = JSON.parse(e.data);
var p = ev.payload || {};
if (p.channel === "_consolidation_") return;
if (p.reset) {
var lm = state.messages.length
? state.messages[state.messages.length - 1]
: null;
if (lm && lm.role === "assistant" && !lm._final) {
lm._final = true;
rerenderChat();
}
return;
}
if (!p.content) return;
state.chatStage = __("AI 回复中...", "AI replying...");
var last =
state.messages.length > 0
? state.messages[state.messages.length - 1]
: null;
if (!last || last.role !== "assistant" || last._final) {
state.messages.push({
role: "assistant",
content: "",
tool_calls: [],
_streaming: true,
_grow: true,
});
last = state.messages[state.messages.length - 1];
}
last.content += p.content;
rerenderChat();
} catch (ex) {}
});
es.addEventListener("terminal_output", function (e) {
try {
var ev = JSON.parse(e.data);
@ -4170,6 +4232,7 @@ background:
try {
var ev = JSON.parse(e.data);
var p = ev.payload || {};
if (p.channel === "_consolidation_") return;
if (p.content) {
state.chatStage = __("AI 思考中...", "AI thinking...");
var last =
@ -4186,12 +4249,39 @@ background:
});
last = state.messages[state.messages.length - 1];
}
last.reasoning_content =
(last.reasoning_content || "") + p.content;
// 聚合 reasoning 帧携带全文：直接覆盖（若已有 delta 累积则等价）
last.reasoning_content = p.content;
rerenderChat();
}
} catch (ex) {}
});
// token 级流式增量：逐块追加到当前思考内容；reset 帧表示轮次作废
es.addEventListener("reasoning_delta", function (e) {
try {
var ev = JSON.parse(e.data);
var p = ev.payload || {};
if (p.channel === "_consolidation_") return;
if (p.reset) return; // 轮次作废（用户中断）：清空累积中的思考
if (!p.content) return;
state.chatStage = __("AI 思考中...", "AI thinking...");
var last =
state.messages.length > 0
? state.messages[state.messages.length - 1]
: null;
if (!last || last.role !== "assistant" || last._final) {
state.messages.push({
role: "assistant",
content: "",
reasoning_content: "",
tool_calls: [],
_streaming: true,
});
last = state.messages[state.messages.length - 1];
}
last.reasoning_content = (last.reasoning_content || "") + p.content;
rerenderChat();
} catch (ex) {}
});
es.addEventListener("tool_call", function (e) {
try {
var ev = JSON.parse(e.data);

53
internal/plugins/webui/handler.go
View File

@ -694,6 +694,7 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
mux.HandleFunc("/api/v1/tracker/", h.requireAPI(h.handleTracker))
mux.HandleFunc("/api/v1/chat", h.requireAPI(h.handleChat))
mux.HandleFunc("/api/v1/chat/history", h.requireAPI(h.handleChatHistory))
mux.HandleFunc("/api/v1/chat/interrupt", h.requireAPI(h.handleChatInterrupt))
mux.HandleFunc("/api/v1/chat/events", h.requireAPI(h.handleChatEvents))
mux.HandleFunc("/api/v1/terminals", h.requireAPI(h.handleTerminals))
mux.HandleFunc("/api/v1/cmd/history", h.requireAPI(h.handleCmdHistory))
@ -1221,6 +1222,39 @@ func (h *Handler) handleChatHistory(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]interface{}{"messages": result})
}
// handleChatInterrupt 注入用户中断：取消正在进行的 LLM 生成并/或发送打断消息。
// 核心拦截语义（interceptLoop）：
// - 有 LLM 在跑：cancelLLM 取消当前请求 + 中断入队，process() 以
// [中断消息] 重启轮次，模型看到被打断的上下文和用户新输入；
// - 无 LLM 在跑：作为普通输入处理（等同发了一条消息）。
// message 可选：空则纯取消（仍会注入空内容中断触发取消）。
func (h *Handler) handleChatInterrupt(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if h.sdk == nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "agent unavailable"})
return
}
var body struct {
Message string `json:"message"`
DeviceID string `json:"device_id"`
}
if r.Body != nil {
_ = json.NewDecoder(r.Body).Decode(&body) // body 可选
}
source := "webui"
if body.DeviceID != "" {
source = "webui/" + body.DeviceID
}
h.sdk.InjectInterrupt(source, "webui", "text", map[string]interface{}{
"content": body.Message,
})
writeJSON(w, http.StatusOK, map[string]string{"status": "interrupted"})
}
func (h *Handler) handleTerminals(w http.ResponseWriter, r *http.Request) {
h.termMu.Lock()
terms := make([]*termState, 0, len(h.termStates))
@ -1413,8 +1447,27 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
}
subTypes := []string{"agent_output", "reasoning", "agent_error", "tool_call", "stage", "agent_llm_chain", "terminal_output"}
// token 级流式增量事件：实时转发给浏览器做逐 token 渲染。
// 不进 sseEventRing —— 断线重连只重放聚合事件（最终真相），
// 避免重放 delta 与聚合内容重复追加。
var unsubs []func()
var seq int64
appendDeltaSub := func(evtType sdk.EventType) {
unsub := h.sdk.Subscribe(evtType, func(evt *sdk.Event) {
data, _ := json.Marshal(evt)
seq++
id := fmt.Sprintf("%d-%d", evt.Timestamp, seq)
select {
case writeCh <- fmt.Sprintf("id: %s\nevent: %s\ndata: %s\n", id, evt.Type, string(data)):
default:
log.Printf("[SSE] DROPPED %s (writeCh full, len=%d)", evt.Type, len(writeCh))
}
})
unsubs = append(unsubs, unsub)
}
appendDeltaSub(sdk.EventReasoningDelta)
appendDeltaSub(sdk.EventContentDelta)
for _, t := range subTypes {
t2 := t
unsub := h.sdk.Subscribe(sdk.EventType(t2), func(evt *sdk.Event) {

5
internal/sdk/events.go
View File

@ -18,4 +18,9 @@ const (
EventSystem = events.EventSystem
EventTerminalOutput = events.EventTerminalOutput
EventAll = events.EventAll
// 流式增量事件（token 级）：核心 process() 流式化后每收到一个增量块发布。
// 客户端可选订做真逐 token 渲染；聚合事件仍照常发布，旧订阅者不受影响。
EventReasoningDelta = events.EventReasoningDelta
EventContentDelta = events.EventContentDelta
)