mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 09:28:14 +00:00
fix: 流式渲染回合生命周期 + LLM 瞬断重试与 SSE body 兜底
问题一(webui 不是真流式): - sendChat 的 finally 在 POST 结束(15s ackTimer abort)时就复位 chatLoading,但 agent 生成窗口 15~190s,后续 SSE delta 全部走 全量重建路径、停止按钮提前消失、用户误发重复消息。 - GUI app.js 完全没有 content_delta/reasoning_delta 监听器, 只能等聚合帧一次性显示。 修复:三端统一回合生命周期——POST 只是触发,收尾由 SSE 驱动: - dashboard/GUI 新增 endChatTurn/armTurnWatchdog;拿到同步兜底 响应立即收尾,否则保持回合打开等 agent_output final / reset 帧 / 120s watchdog 兜底 - GUI 补齐 delta 监听器;agent_output 聚合分支 += 改覆盖; reasoning 聚合帧改覆盖(多轮工具调用时旧逻辑会重复累加) - agent_output 误杀分支(final 无 source 即 return 丢弃新输出) 改为内容比较去重,多轮连发时新一轮回复不再被吞 - waiter reasoning_delta reset 从清空全部消息改为 sealLastAgent 问题二(三条只成功一条): - handleChat 60s ctx 含排队时间,agent 串行处理下第 N 条必超时 (实测第 3 条 62s 超时 504);放宽到 300s(客户端 abort 时立即取消) - LLM 单 provider 瞬断无重试:process.go provider 循环内加同源 重试(2 次、退避 2s),401/403 凭证错误与用户中断不重试 - llmsproxy auto 链在非流式请求下可能返回 SSE body(上游恢复后 吐已生成的 chunk 流),非流式解析报 invalid character 'd' 丢掉 整段回复;新增 parseOpenAICompatibleSSEBody 拼接为完整响应 - 顺带修 normalizeStreamToolCalls 分片续传 bug:name 不重发时 argsRaw 被顶层 Arguments(nil) 覆盖丢失 function.arguments 验证: - 连发 3 条 + 单条共 4 条全部成功(首条 190s 重试扛住瞬断) - sse_body_test.go 锁定 SSE body 解析契约(content/usage/tool call 分片)
This commit is contained in:
@ -384,6 +384,11 @@ func (p *LuaAdaptedProvider) Chat(ctx context.Context, req *CompletionRequest) (
|
||||
if parsed, perr := parseOpenAICompatibleResponse(rawResp); perr == nil {
|
||||
return parsed, nil
|
||||
}
|
||||
// 网关在非流式请求下返回了 SSE 流 body(上游恢复后吐 chunk 流),
|
||||
// 拼接为完整响应,避免丢掉已生成的整段回复
|
||||
if parsed, ok := parseOpenAICompatibleSSEBody(rawResp); ok {
|
||||
return parsed, nil
|
||||
}
|
||||
return nil, fmt.Errorf("lua transform_response: %w", err)
|
||||
}
|
||||
|
||||
@ -392,6 +397,9 @@ func (p *LuaAdaptedProvider) Chat(ctx context.Context, req *CompletionRequest) (
|
||||
if parsed, perr := parseOpenAICompatibleResponse(rawResp); perr == nil {
|
||||
return parsed, nil
|
||||
}
|
||||
if parsed, ok := parseOpenAICompatibleSSEBody(rawResp); ok {
|
||||
return parsed, nil
|
||||
}
|
||||
return nil, fmt.Errorf("unmarshal unified response: %w (body: %s)", err, unifiedJSON)
|
||||
}
|
||||
|
||||
@ -460,6 +468,86 @@ func parseOpenAICompatibleResponse(raw []byte) (*CompletionResponse, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// parseOpenAICompatibleSSEBody 将 SSE 格式的响应体("data: {...}" 多行)
|
||||
// 拼接为完整 CompletionResponse。场景:网关(llmsproxy auto 链等)在非流式
|
||||
// 请求下也可能返回流式 body——上游恢复后吐出的是已生成的 chunk 流,若按
|
||||
// 普通 JSON 解析会报 "invalid character 'd'" 而丢掉整段完整回复。
|
||||
// 返回 false 表示 body 不是 SSE 格式,调用方继续走原有解析路径。
|
||||
func parseOpenAICompatibleSSEBody(raw []byte) (*CompletionResponse, bool) {
|
||||
body := strings.TrimSpace(string(raw))
|
||||
if !strings.HasPrefix(body, "data:") && !strings.Contains(body, "\ndata:") {
|
||||
return nil, false
|
||||
}
|
||||
type sseAcc struct {
|
||||
id string
|
||||
name string
|
||||
argsRaw strings.Builder
|
||||
}
|
||||
var out CompletionResponse
|
||||
var contentBuf, reasoningBuf strings.Builder
|
||||
accs := map[int]*sseAcc{}
|
||||
toolOrder := []int{}
|
||||
finish := ""
|
||||
found := false
|
||||
|
||||
for _, line := range strings.Split(body, "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if !strings.HasPrefix(line, "data:") {
|
||||
continue
|
||||
}
|
||||
payload := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
|
||||
if payload == "" || payload == "[DONE]" {
|
||||
continue
|
||||
}
|
||||
ck, ok := parseOpenAICompatibleStreamChunkFull(payload)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
contentBuf.WriteString(ck.Content)
|
||||
reasoningBuf.WriteString(ck.ReasoningContent)
|
||||
for i, tc := range ck.ToolCalls {
|
||||
acc := accs[i]
|
||||
if acc == nil {
|
||||
acc = &sseAcc{}
|
||||
accs[i] = acc
|
||||
toolOrder = append(toolOrder, i)
|
||||
}
|
||||
if tc.ID != "" {
|
||||
acc.id = tc.ID
|
||||
}
|
||||
if tc.Name != "" {
|
||||
acc.name = tc.Name
|
||||
}
|
||||
acc.argsRaw.WriteString(tc.RawArguments)
|
||||
}
|
||||
if ck.Done && ck.FinishReason != "" {
|
||||
finish = ck.FinishReason
|
||||
}
|
||||
if ck.Usage != nil {
|
||||
out.TokenUsage = *ck.Usage
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return nil, false
|
||||
}
|
||||
out.Content = contentBuf.String()
|
||||
out.ReasoningContent = reasoningBuf.String()
|
||||
out.FinishReason = finish
|
||||
for _, i := range toolOrder {
|
||||
acc := accs[i]
|
||||
name := strings.TrimSpace(acc.name)
|
||||
argsStr := strings.TrimSpace(acc.argsRaw.String())
|
||||
if name == "" && argsStr == "" && acc.id == "" {
|
||||
continue
|
||||
}
|
||||
tc := ToolCall{ID: acc.id, Type: "function", Name: name, RawArguments: argsStr}
|
||||
tc.Arguments = parseToolArguments(argsStr)
|
||||
out.ToolCalls = append(out.ToolCalls, tc)
|
||||
}
|
||||
return &out, true
|
||||
}
|
||||
|
||||
type openAIToolCall struct {
|
||||
ID string `json:"id"`
|
||||
Type string `json:"type"`
|
||||
@ -526,7 +614,11 @@ func normalizeStreamToolCalls(raw []openAIToolCall) []ToolCall {
|
||||
argsRaw := tc.Function.Arguments
|
||||
if name == "" {
|
||||
name = tc.Name
|
||||
argsRaw = tc.Arguments
|
||||
// 仅当顶层 Arguments 存在才用扁平格式;否则保留 function.arguments 嵌套值
|
||||
// (OpenAI 流式续传 chunk:name 不重发但 function.arguments 继续)
|
||||
if tc.Arguments != nil {
|
||||
argsRaw = tc.Arguments
|
||||
}
|
||||
}
|
||||
typ := tc.Type
|
||||
if typ == "" && (tc.ID != "" || name != "" || argsRaw != nil) {
|
||||
|
||||
121
internal/agent/api/sse_body_test.go
Normal file
121
internal/agent/api/sse_body_test.go
Normal file
@ -0,0 +1,121 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 锁定契约:网关(llmsproxy auto 链等)在非流式请求下返回 SSE 流 body 时,
|
||||
// 必须拼接为完整响应而不是报 "invalid character 'd'" 丢掉已生成的回复。
|
||||
// 事故样本取自 2026-08-25 生产日志:上游恢复后吐出完整 chunk 流被非流式解析器丢弃。
|
||||
func TestParseOpenAICompatibleSSEBody(t *testing.T) {
|
||||
body := "data: {\"id\":\"chatcmpl-572\",\"object\":\"chat.completion.chunk\",\"created\":1787630289,\"model\":\"x-preview-f-free\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"你好\"},\"finish_reason\":null}]}\n" +
|
||||
"\n" +
|
||||
"data: {\"id\":\"chatcmpl-572\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\",世界\"},\"finish_reason\":null}]}\n" +
|
||||
"\n" +
|
||||
"data: {\"id\":\"chatcmpl-572\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"\"},\"finish_reason\":\"stop\"}]}\n" +
|
||||
"data: {\"id\":\"chatcmpl-572\",\"choices\":[],\"usage\":{\"prompt_tokens\":100,\"completion_tokens\":7,\"total_tokens\":107}}\n" +
|
||||
"data: [DONE]\n"
|
||||
|
||||
resp, ok := parseOpenAICompatibleSSEBody([]byte(body))
|
||||
if !ok {
|
||||
t.Fatal("expected SSE body to be recognized")
|
||||
}
|
||||
if resp.Content != "你好,世界" {
|
||||
t.Errorf("content = %q, want %q", resp.Content, "你好,世界")
|
||||
}
|
||||
if resp.FinishReason != "stop" {
|
||||
t.Errorf("finish_reason = %q, want stop", resp.FinishReason)
|
||||
}
|
||||
if resp.TokenUsage.Total != 107 || resp.TokenUsage.Prompt != 100 || resp.TokenUsage.Completion != 7 {
|
||||
t.Errorf("usage = %+v, want prompt=100 completion=7 total=107", resp.TokenUsage)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseOpenAICompatibleSSEBodyRejectsPlainJSON(t *testing.T) {
|
||||
plain := `{"choices":[{"message":{"content":"hi"}}]}`
|
||||
if _, ok := parseOpenAICompatibleSSEBody([]byte(plain)); ok {
|
||||
t.Fatal("plain JSON body must not be treated as SSE")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseOpenAICompatibleSSEBodyToolCallShards(t *testing.T) {
|
||||
// 用 json.Marshal 构建测试数据,避免 Go 字面量转义错误
|
||||
chunk1 := map[string]interface{}{
|
||||
"choices": []map[string]interface{}{{
|
||||
"index": 0,
|
||||
"delta": map[string]interface{}{
|
||||
"tool_calls": []map[string]interface{}{{
|
||||
"index": 0,
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": map[string]interface{}{
|
||||
"name": "exec",
|
||||
"arguments": `{"command":`,
|
||||
},
|
||||
}},
|
||||
},
|
||||
}},
|
||||
}
|
||||
chunk2 := map[string]interface{}{
|
||||
"choices": []map[string]interface{}{{
|
||||
"index": 0,
|
||||
"delta": map[string]interface{}{
|
||||
"tool_calls": []map[string]interface{}{{
|
||||
"index": 0,
|
||||
"function": map[string]interface{}{
|
||||
"arguments": `"date"}`,
|
||||
},
|
||||
}},
|
||||
},
|
||||
}},
|
||||
}
|
||||
chunk3 := map[string]interface{}{
|
||||
"choices": []map[string]interface{}{{
|
||||
"index": 0,
|
||||
"delta": map[string]interface{}{},
|
||||
"finish_reason": "tool_calls",
|
||||
}},
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
for _, c := range []map[string]interface{}{chunk1, chunk2, chunk3} {
|
||||
b, _ := json.Marshal(c)
|
||||
sb.WriteString("data: ")
|
||||
sb.Write(b)
|
||||
sb.WriteString("\n")
|
||||
}
|
||||
sb.WriteString("data: [DONE]\n")
|
||||
|
||||
t.Logf("SSE body:\n%s", sb.String())
|
||||
|
||||
resp, ok := parseOpenAICompatibleSSEBody([]byte(sb.String()))
|
||||
if !ok {
|
||||
t.Fatal("expected SSE body to be recognized")
|
||||
}
|
||||
if len(resp.ToolCalls) != 1 {
|
||||
t.Fatalf("got %d tool calls, want 1", len(resp.ToolCalls))
|
||||
}
|
||||
tc := resp.ToolCalls[0]
|
||||
if tc.Name != "exec" || tc.ID != "call_1" {
|
||||
t.Errorf("tool call name/id = %q/%q, want exec/call_1", tc.Name, tc.ID)
|
||||
}
|
||||
args := tc.RawArguments
|
||||
if args != `{"command":"date"}` {
|
||||
t.Errorf("raw args = %q", args)
|
||||
}
|
||||
if tc.Arguments["command"] != "date" {
|
||||
t.Errorf("parsed args = %v, want command=date", tc.Arguments)
|
||||
}
|
||||
if resp.FinishReason != "tool_calls" {
|
||||
t.Errorf("finish_reason = %q, want tool_calls", resp.FinishReason)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseOpenAICompatibleSSEBodyEmptyStream(t *testing.T) {
|
||||
body := "data: \ndata: \n"
|
||||
if resp, ok := parseOpenAICompatibleSSEBody([]byte(body)); ok && strings.TrimSpace(resp.Content) != "" {
|
||||
t.Fatalf("empty stream should not parse into non-empty response")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user