opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: execute LLM tool calls in parallel for faster response (#1070)

Browse source 
When the LLM returns multiple tool calls, they are now executed
concurrently using goroutines + sync.WaitGroup instead of sequentially.
Results are collected in an indexed slice and processed in original order
to preserve message ordering. MessageTool.sentInRound is changed to
atomic.Bool for thread safety.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Guoguo 2026-03-04 17:17:28 +08:00 • committed by GitHub
parent 2a577f7a1d
commit 028605cfd0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 106 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

102
pkg/agent/loop.go
View file

@ -969,62 +969,76 @@ func (al *AgentLoop) runLLMIteration(
// Save assistant message with tool calls to session // Save assistant message with tool calls to session
agent.Sessions.AddFullMessage(opts.SessionKey, assistantMsg) agent.Sessions.AddFullMessage(opts.SessionKey, assistantMsg)
// Execute tool calls // Execute tool calls in parallel
for _, tc := range normalizedToolCalls { type indexedAgentResult struct {
argsJSON, _ := json.Marshal(tc.Arguments) result *tools.ToolResult
argsPreview := utils.Truncate(string(argsJSON), 200) tc providers.ToolCall
logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview), }
map[string]any{
"agent_id": agent.ID,
"tool": tc.Name,
"iteration": iteration,
})
// Create async callback for tools that implement AsyncTool agentResults := make([]indexedAgentResult, len(normalizedToolCalls))
// NOTE: Following openclaw's design, async tools do NOT send results directly to users. var wg sync.WaitGroup
// Instead, they notify the agent via PublishInbound, and the agent decides
// whether to forward the result to the user (in processSystemMessage). for i, tc := range normalizedToolCalls {
asyncCallback := func(callbackCtx context.Context, result *tools.ToolResult) { agentResults[i].tc = tc
// Log the async completion but don't send directly to user
// The agent will handle user notification via processSystemMessage wg.Add(1)
if !result.Silent && result.ForUser != "" { go func(idx int, tc providers.ToolCall) {
logger.InfoCF("agent", "Async tool completed, agent will handle notification", defer wg.Done()
map[string]any{
"tool": tc.Name, argsJSON, _ := json.Marshal(tc.Arguments)
"content_len": len(result.ForUser), argsPreview := utils.Truncate(string(argsJSON), 200)
}) logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview),
map[string]any{
"agent_id": agent.ID,
"tool": tc.Name,
"iteration": iteration,
})
// Create async callback for tools that implement AsyncTool
asyncCallback := func(callbackCtx context.Context, result *tools.ToolResult) {
if !result.Silent && result.ForUser != "" {
logger.InfoCF("agent", "Async tool completed, agent will handle notification",
map[string]any{
"tool": tc.Name,
"content_len": len(result.ForUser),
})
}
} }
}
toolResult := agent.Tools.ExecuteWithContext( toolResult := agent.Tools.ExecuteWithContext(
ctx, ctx,
tc.Name, tc.Name,
tc.Arguments, tc.Arguments,
opts.Channel, opts.Channel,
opts.ChatID, opts.ChatID,
asyncCallback, asyncCallback,
) )
agentResults[idx].result = toolResult
}(i, tc)
}
wg.Wait()
// Process results in original order (send to user, save to session)
for _, r := range agentResults {
// Send ForUser content to user immediately if not Silent // Send ForUser content to user immediately if not Silent
if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse { if !r.result.Silent && r.result.ForUser != "" && opts.SendResponse {
al.bus.PublishOutbound(ctx, bus.OutboundMessage{ al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: opts.Channel, Channel: opts.Channel,
ChatID: opts.ChatID, ChatID: opts.ChatID,
Content: toolResult.ForUser, Content: r.result.ForUser,
}) })
logger.DebugCF("agent", "Sent tool result to user", logger.DebugCF("agent", "Sent tool result to user",
map[string]any{ map[string]any{
"tool": tc.Name, "tool": r.tc.Name,
"content_len": len(toolResult.ForUser), "content_len": len(r.result.ForUser),
}) })
} }
// If tool returned media refs, publish them as outbound media // If tool returned media refs, publish them as outbound media
if len(toolResult.Media) > 0 && opts.SendResponse { if len(r.result.Media) > 0 && opts.SendResponse {
parts := make([]bus.MediaPart, 0, len(toolResult.Media)) parts := make([]bus.MediaPart, 0, len(r.result.Media))
for _, ref := range toolResult.Media { for _, ref := range r.result.Media {
part := bus.MediaPart{Ref: ref} part := bus.MediaPart{Ref: ref}
// Populate metadata from MediaStore when available
if al.mediaStore != nil { if al.mediaStore != nil {
if _, meta, err := al.mediaStore.ResolveWithMeta(ref); err == nil { if _, meta, err := al.mediaStore.ResolveWithMeta(ref); err == nil {
part.Filename = meta.Filename part.Filename = meta.Filename
@ -1042,15 +1056,15 @@ func (al *AgentLoop) runLLMIteration(
} }
// Determine content for LLM based on tool result // Determine content for LLM based on tool result
contentForLLM := toolResult.ForLLM contentForLLM := r.result.ForLLM
if contentForLLM == "" && toolResult.Err != nil { if contentForLLM == "" && r.result.Err != nil {
contentForLLM = toolResult.Err.Error() contentForLLM = r.result.Err.Error()
} }
toolResultMsg := providers.Message{ toolResultMsg := providers.Message{
Role: "tool", Role: "tool",
Content: contentForLLM, Content: contentForLLM,
ToolCallID: tc.ID, ToolCallID: r.tc.ID,
} }
messages = append(messages, toolResultMsg) messages = append(messages, toolResultMsg)

9
pkg/tools/message.go
View file

@ -3,6 +3,7 @@ package tools
import ( import (
"context" "context"
"fmt" "fmt"
"sync/atomic"
) )
type SendCallback func(channel, chatID, content string) error type SendCallback func(channel, chatID, content string) error
@ -11,7 +12,7 @@ type MessageTool struct {
sendCallback SendCallback sendCallback SendCallback
defaultChannel string defaultChannel string
defaultChatID string defaultChatID string
sentInRound bool // Tracks whether a message was sent in the current processing round sentInRound atomic.Bool // Tracks whether a message was sent in the current processing round
} }
func NewMessageTool() *MessageTool { func NewMessageTool() *MessageTool {
@ -50,12 +51,12 @@ func (t *MessageTool) Parameters() map[string]any {
func (t *MessageTool) SetContext(channel, chatID string) { func (t *MessageTool) SetContext(channel, chatID string) {
t.defaultChannel = channel t.defaultChannel = channel
t.defaultChatID = chatID t.defaultChatID = chatID
t.sentInRound = false // Reset send tracking for new processing round t.sentInRound.Store(false) // Reset send tracking for new processing round
} }
// HasSentInRound returns true if the message tool sent a message during the current round. // HasSentInRound returns true if the message tool sent a message during the current round.
func (t *MessageTool) HasSentInRound() bool { func (t *MessageTool) HasSentInRound() bool {
return t.sentInRound return t.sentInRound.Load()
} }
func (t *MessageTool) SetSendCallback(callback SendCallback) { func (t *MessageTool) SetSendCallback(callback SendCallback) {
@ -94,7 +95,7 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]any) *ToolRes
} }
} }
t.sentInRound = true t.sentInRound.Store(true)
// Silent: user already received the message directly // Silent: user already received the message directly
return &ToolResult{ return &ToolResult{
ForLLM: fmt.Sprintf("Message sent to %s:%s", channel, chatID), ForLLM: fmt.Sprintf("Message sent to %s:%s", channel, chatID),

69
pkg/tools/toolloop.go
View file

@ -10,6 +10,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"sync"
"github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
@ -121,37 +122,53 @@ func RunToolLoop(
} }
messages = append(messages, assistantMsg) messages = append(messages, assistantMsg)
// 7. Execute tool calls // 7. Execute tool calls in parallel
for _, tc := range normalizedToolCalls { type indexedResult struct {
argsJSON, _ := json.Marshal(tc.Arguments) result *ToolResult
argsPreview := utils.Truncate(string(argsJSON), 200) tc providers.ToolCall
logger.InfoCF("toolloop", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview), }
map[string]any{
"tool": tc.Name,
"iteration": iteration,
})
// Execute tool (no async callback for subagents - they run independently) results := make([]indexedResult, len(normalizedToolCalls))
var toolResult *ToolResult var wg sync.WaitGroup
if config.Tools != nil {
toolResult = config.Tools.ExecuteWithContext(ctx, tc.Name, tc.Arguments, channel, chatID, nil) for i, tc := range normalizedToolCalls {
} else { results[i].tc = tc
toolResult = ErrorResult("No tools available")
wg.Add(1)
go func(idx int, tc providers.ToolCall) {
defer wg.Done()
argsJSON, _ := json.Marshal(tc.Arguments)
argsPreview := utils.Truncate(string(argsJSON), 200)
logger.InfoCF("toolloop", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview),
map[string]any{
"tool": tc.Name,
"iteration": iteration,
})
var toolResult *ToolResult
if config.Tools != nil {
toolResult = config.Tools.ExecuteWithContext(ctx, tc.Name, tc.Arguments, channel, chatID, nil)
} else {
toolResult = ErrorResult("No tools available")
}
results[idx].result = toolResult
}(i, tc)
}
wg.Wait()
// Append results in original order
for _, r := range results {
contentForLLM := r.result.ForLLM
if contentForLLM == "" && r.result.Err != nil {
contentForLLM = r.result.Err.Error()
} }
// Determine content for LLM messages = append(messages, providers.Message{
contentForLLM := toolResult.ForLLM
if contentForLLM == "" && toolResult.Err != nil {
contentForLLM = toolResult.Err.Error()
}
// Add tool result message
toolResultMsg := providers.Message{
Role: "tool", Role: "tool",
Content: contentForLLM, Content: contentForLLM,
ToolCallID: tc.ID, ToolCallID: r.tc.ID,
} })
messages = append(messages, toolResultMsg)
} }
} }