diff --git a/internal/agent/api/provider.go b/internal/agent/api/provider.go index 368f331..034eaf0 100644 --- a/internal/agent/api/provider.go +++ b/internal/agent/api/provider.go @@ -7,6 +7,8 @@ import ( "encoding/json" "fmt" "io" + "log" + "os" "net" "net/http" "sort" @@ -191,6 +193,10 @@ type ToolCall struct { Name string `json:"name"` Arguments map[string]interface{} `json:"arguments"` RawArguments string `json:"raw_arguments,omitempty"` // 流式分片原始 JSON 字符串 + // StreamIndex 是上游流式 tool_call 的 OpenAI index 字段（并行多工具调用 + // 时同一轮的分片用它区分归属）。lua 适配器以 stream_index 键透传； + // 仅内核流式累积内部使用，不序列化到对外 API。 + StreamIndex int `json:"stream_index,omitempty"` } type apiToolCall struct { @@ -403,6 +409,14 @@ func (p *LuaAdaptedProvider) Chat(ctx context.Context, req *CompletionRequest) ( return nil, fmt.Errorf("unmarshal unified response: %w (body: %s)", err, unifiedJSON) } + // 诊断：tool_calls 存在但参数为空——上游/适配器丢参数，打印原始响应片段定位 + for _, tc := range result.ToolCalls { + if len(tc.Arguments) == 0 && tc.RawArguments == "" { + log.Printf("[provider:%s] tool_call %s (%s) has empty arguments; raw body head: %s", + p.name, tc.Name, tc.ID, string(rawResp[:min(len(rawResp), 400)])) + } + } + return &result, nil } @@ -551,6 +565,7 @@ func parseOpenAICompatibleSSEBody(raw []byte) (*CompletionResponse, bool) { type openAIToolCall struct { ID string `json:"id"` Type string `json:"type"` + Index int `json:"index"` Function struct { Name string `json:"name"` Arguments interface{} `json:"arguments"` @@ -584,6 +599,7 @@ func normalizeOpenAIToolCalls(raw []openAIToolCall) []ToolCall { Name: name, Arguments: parseToolArguments(argsRaw), RawArguments: rawArgsString(argsRaw), + StreamIndex: tc.Index, }) } return out @@ -629,6 +645,7 @@ func normalizeStreamToolCalls(raw []openAIToolCall) []ToolCall { Type: typ, Name: name, RawArguments: rawArgsString(argsRaw), + StreamIndex: tc.Index, }) } return out @@ -856,6 +873,7 @@ func (p *LuaAdaptedProvider) ChatStream(ctx context.Context, req *CompletionRequ } } + debugSSE := os.Getenv("HOMED_DEBUG_SSE") == "1" for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" || !strings.HasPrefix(line, "data:") { @@ -865,6 +883,9 @@ func (p *LuaAdaptedProvider) ChatStream(ctx context.Context, req *CompletionRequ if data == "" { continue } + if debugSSE && strings.Contains(data, "tool_calls") { + log.Printf("[provider:%s] SSE raw tool_call line: %s", p.name, truncateForLog(data, 400)) + } if data == "[DONE]" { if !doneSent { if !emit(StreamChunk{Done: true}) { @@ -1208,3 +1229,11 @@ func getFloat(m map[string]interface{}, key string) float64 { } return 0 } + +// truncateForLog 诊断日志用截断。 +func truncateForLog(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "..." +} diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index 69d673e..646bb3b 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -86,6 +86,7 @@ type Agent struct { childMu sync.Mutex childNextID int64 childResults map[string]string + childRunning map[string]bool // 运行中的子任务（child_result 查询时区分'运行中'与'不存在'） // 高优先级打断通道：interceptLoop 注入，process() 在工具循环轮次间非阻塞读取 interceptCh chan *agentIO.InputEvent @@ -237,6 +238,7 @@ func New(cfg AgentConfig) *Agent { eventBus: cfg.EventBus, selfInputCh: make(chan selfInputMsg, 64), childResults: make(map[string]string), + childRunning: make(map[string]bool), interceptCh: make(chan *agentIO.InputEvent, 64), pluginHealth: newPluginHealthTracker(), thinkingEnabled: cfg.ThinkingEnabled, diff --git a/internal/agent/core/process.go b/internal/agent/core/process.go index 5c2eca7..ea24617 100644 --- a/internal/agent/core/process.go +++ b/internal/agent/core/process.go @@ -262,6 +262,9 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri toolsUsed = append(toolsUsed, tc.Name) pluginName := a.resolveToolPlugin(tc.Name) log.Printf("[agent] executing tool: %s (plugin=%s, id=%s)", tc.Name, pluginName, tc.ID) + if tc.RawArguments != "" { + log.Printf("[agent] tool %s raw_arguments: %s", tc.Name, truncateStr(tc.RawArguments, 300)) + } sdkTC := sdk.ToolCall{ID: tc.ID, Name: tc.Name, Plugin: pluginName, Arguments: tc.Arguments} stageCtx.ToolCalls = []sdk.ToolCall{sdkTC} @@ -402,13 +405,22 @@ func accumulateStream(ctx context.Context, ch <-chan agentAPI.StreamChunk, a *Ag return } if acc.name == "" { + log.Printf("[agent] stream tool_call idx=%d flushed with EMPTY name (args=%q) — dropped", idx, truncateStr(acc.argsRaw.String(), 120)) delete(accs, idx) return } + args, argsOK := parseToolArgsJSON(acc.argsRaw.String()) + raw := strings.TrimSpace(acc.argsRaw.String()) + // 空参诊断：区分「上游没发分片」(raw="")、「混拼污染」(解析失败) 与「合法空对象」({})。 + if !argsOK { + log.Printf("[agent] stream tool_call %s (idx=%d) argument fragments invalid JSON: %q", acc.name, idx, truncateStr(raw, 200)) + } else if raw == "" { + log.Printf("[agent] stream tool_call %s (idx=%d) received NO argument fragments", acc.name, idx) + } tc := agentAPI.ToolCall{ ID: acc.id, Name: acc.name, - Arguments: parseToolArgsJSON(acc.argsRaw.String()), + Arguments: args, } resp.ToolCalls = append(resp.ToolCalls, tc) delete(accs, idx) @@ -446,9 +458,16 @@ func accumulateStream(ctx context.Context, ch <-chan agentAPI.StreamChunk, a *Ag } } - // 增量 tool call 分片：OpenAI 风格按 index 拼接 id/name/arguments - for i, tc := range ck.ToolCalls { - idx := i + // 增量 tool call 分片：OpenAI 风格按 index 字段拼接 id/name/arguments。 + // 注意必须用分片自带的 StreamIndex（上游 JSON "index"），不能用 Go + // range 序号：每个 SSE chunk 通常只含一个 tool_call 元素，slice 序号 + // 恒为 0，并行多工具调用（index=0,1,2...）的分片会全部污染到同一个桶， + // 导致 name 相互覆盖、args 碎片混拼解析失败（空参数工具调用）。 + for _, tc := range ck.ToolCalls { + idx := tc.StreamIndex + if idx == 0 && tc.Name == "" && tc.RawArguments == "" { + continue + } acc := accs[idx] if acc == nil { acc = &toolCallAcc{} @@ -483,16 +502,17 @@ func accumulateStream(ctx context.Context, ch <-chan agentAPI.StreamChunk, a *Ag } // parseToolArgsJSON 将经过完整拼接的 tool call arguments JSON 字符串解析为 map。 -// 空字符串返回空 map。 -func parseToolArgsJSON(s string) map[string]interface{} { - if s == "" { - return map[string]interface{}{} +// 第二个返回值 ok=false 表示分片拼接结果不是合法 JSON（分片污染/丢失）， +// 与「合法的空对象 {}」相区分。 +func parseToolArgsJSON(s string) (map[string]interface{}, bool) { + if strings.TrimSpace(s) == "" { + return map[string]interface{}{}, true } var m map[string]interface{} if err := json.Unmarshal([]byte(s), &m); err == nil && m != nil { - return m + return m, true } - return map[string]interface{}{} + return map[string]interface{}{}, false } func convertToolCalls(tcs []agentAPI.ToolCall) []sdk.ToolCall { diff --git a/internal/agent/core/spawn.go b/internal/agent/core/spawn.go index b20ec82..e81f157 100644 --- a/internal/agent/core/spawn.go +++ b/internal/agent/core/spawn.go @@ -1,6 +1,7 @@ package core import ( + "encoding/json" "fmt" "log" "strings" @@ -11,6 +12,9 @@ import ( func (a *Agent) executeSpawnChild(tc agentAPI.ToolCall) string { task, _ := tc.Arguments["task"].(string) if task == "" { + if b, _ := json.Marshal(tc.Arguments); len(b) > 2 { + log.Printf("[spawn] task empty but arguments present: %s", truncateStr(string(b), 300)) + } return "请提供 task 参数" } maxTurns := 0 @@ -36,6 +40,9 @@ func (a *Agent) executeSpawnChild(tc agentAPI.ToolCall) string { parentChannel = "cli" } + a.childMu.Lock() + a.childRunning[taskID] = true + a.childMu.Unlock() go a.runChildTask(taskID, task, parentChannel, maxTurns) return fmt.Sprintf("子任务已启动（ID: %s，最多 %d 轮），完成后会自动通知你，届时请使用 child_result 工具查看输出", taskID, maxTurns) @@ -120,6 +127,7 @@ func (a *Agent) runChildTask(taskID, task string, parentChannel string, maxTurns a.childMu.Lock() a.childResults[taskID] = finalResult + delete(a.childRunning, taskID) a.childMu.Unlock() log.Printf("[child] %s done: %s", taskID, truncateStr(finalResult, 100)) @@ -139,14 +147,17 @@ func (a *Agent) executeChildResultTool(tc agentAPI.ToolCall) string { a.childMu.Lock() result, ok := a.childResults[taskID] - if !ok { + if ok { + delete(a.childResults, taskID) a.childMu.Unlock() - return fmt.Sprintf("子任务 %s 不存在或已过期", taskID) + return fmt.Sprintf("【子任务 %s 结果】\n%s", taskID, result) + } + if a.childRunning[taskID] { + a.childMu.Unlock() + return fmt.Sprintf("子任务 %s 仍在运行中，尚未完成。请等待完成通知后再查询。", taskID) } - delete(a.childResults, taskID) a.childMu.Unlock() - - return fmt.Sprintf("【子任务 %s 结果】\n%s", taskID, result) + return fmt.Sprintf("子任务 %s 不存在或已过期", taskID) } func (a *Agent) executeLLMTool(tc agentAPI.ToolCall) string { diff --git a/internal/agent/core/stream_index_test.go b/internal/agent/core/stream_index_test.go new file mode 100644 index 0000000..0e4de4a --- /dev/null +++ b/internal/agent/core/stream_index_test.go @@ -0,0 +1,87 @@ +package core + +import ( + "context" + "encoding/json" + "testing" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" +) + +// 回归：并行多工具调用的流式分片必须按上游 index 字段分桶累积， +// 不能用 Go range 序号（每 chunk 单元素时恒为 0，导致全部污染到同一桶）。 +func TestAccumulateStreamParallelToolCallsByIndex(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + mk := func(idx int, id, name, raw string) agentAPI.StreamChunk { + return agentAPI.StreamChunk{ + ToolCalls: []agentAPI.ToolCall{{ + ID: id, Type: "function", Name: name, RawArguments: raw, StreamIndex: idx, + }}, + } + } + ch := make(chan agentAPI.StreamChunk, 32) + chunks := []agentAPI.StreamChunk{ + {Content: ""}, + // tool_call 0: spawn_child 参数较长，分多片 + mk(0, "call_a", "spawn_child", "{\"task\":"), + mk(0, "", "", "\"调查大模型排名\"}"), + // tool_call 1: browser_render + mk(1, "call_b", "browser_render", "{\"url\":"), + mk(1, "", "", "\"https://example.com\"}"), + // tool_call 2: cmd_run + mk(2, "call_c", "cmd_run", "{\"command\":\"uname -a\"}"), + // tool_call 3: skill_list + mk(3, "call_d", "skill_list", "{}"), + {Done: true, FinishReason: "tool_calls"}, + } + for _, ck := range chunks { + ch <- ck + } + close(ch) + + resp, err := accumulateStream(ctx, ch, nil) + if err != nil { + t.Fatalf("accumulateStream: %v", err) + } + if len(resp.ToolCalls) != 4 { + t.Fatalf("expected 4 tool calls, got %d: %+v", len(resp.ToolCalls), resp.ToolCalls) + } + want := map[string]string{ + "spawn_child": `{"task":"调查大模型排名"}`, + "browser_render": `{"url":"https://example.com"}`, + "cmd_run": `{"command":"uname -a"}`, + "skill_list": `{}`, + } + for _, tc := range resp.ToolCalls { + raw, _ := json.Marshal(tc.Arguments) + got := string(raw) + exp, ok := want[tc.Name] + if !ok { + t.Errorf("unexpected tool %q args=%s", tc.Name, got) + continue + } + delete(want, tc.Name) + if tc.Name == "skill_list" { + // 无参工具的合法空对象 {}，只需确认没被污染成乱码 + continue + } + if len(tc.Arguments) == 0 { + t.Errorf("tool %q has EMPTY arguments (index pollution regression)", tc.Name) + continue + } + var wantMap map[string]interface{} + json.Unmarshal([]byte(exp), &wantMap) + gotB, _ := json.Marshal(wantMap) + if got != string(gotB) { + t.Errorf("tool %q args = %s, want %s", tc.Name, got, exp) + } + } + if len(want) > 0 { + t.Errorf("missing tool calls: %v", want) + } + if resp.FinishReason != "tool_calls" { + t.Errorf("finish reason = %q, want tool_calls", resp.FinishReason) + } +} diff --git a/internal/plugins/webui/plugin.go b/internal/plugins/webui/plugin.go index 33ea14e..319b94b 100644 --- a/internal/plugins/webui/plugin.go +++ b/internal/plugins/webui/plugin.go @@ -165,12 +165,12 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { s.Publish(&sdk.Event{ Type: sdk.EventAgentOutput, Payload: map[string]interface{}{ - "content": payload, - "channel": "webui", - "kind": "channel_output", + "content": payload, + "channel": "webui", + "kind": "channel_output", "output_type": rawType, - "url": url, - "size": size, + "url": url, + "size": size, }, }) return map[string]interface{}{"status": "ok", "url": url, "size": size}, nil