From 056d9bf73d410ee9e9ed6f7981c28da88278e7d6 Mon Sep 17 00:00:00 2001 From: Anton Bogdanovich <27antonb@gmail.com> Date: Sat, 9 May 2026 16:10:47 -0700 Subject: [PATCH] feat(agent): synthesize steering final replies from action log --- pkg/agent/action_summary.go | 157 ++++++++++++++++++++++++++++++++++ pkg/agent/agent_test.go | 107 +++++++++++++++++++++++ pkg/agent/pipeline_execute.go | 8 ++ pkg/agent/pipeline_llm.go | 2 + pkg/agent/turn_coord.go | 5 ++ pkg/agent/turn_state.go | 2 + pkg/config/config.go | 53 +++++++----- pkg/config/defaults.go | 13 ++- 8 files changed, 321 insertions(+), 26 deletions(-) create mode 100644 pkg/agent/action_summary.go diff --git a/pkg/agent/action_summary.go b/pkg/agent/action_summary.go new file mode 100644 index 000000000..46f6e2175 --- /dev/null +++ b/pkg/agent/action_summary.go @@ -0,0 +1,157 @@ +package agent + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/providers" +) + +type TurnActionRecord struct { + Source string `json:"source"` + Tool string `json:"tool,omitempty"` + Text string `json:"text"` + Error bool `json:"error,omitempty"` +} + +func appendTurnActionRecord( + records []TurnActionRecord, + source, tool, text string, + isError bool, +) []TurnActionRecord { + text = strings.TrimSpace(text) + if text == "" { + return records + } + rec := TurnActionRecord{ + Source: source, + Tool: strings.TrimSpace(tool), + Text: text, + Error: isError, + } + if n := len(records); n > 0 { + prev := records[n-1] + if prev.Source == rec.Source && prev.Tool == rec.Tool && prev.Text == rec.Text && prev.Error == rec.Error { + return records + } + } + return append(records, rec) +} + +func actionSummaryEligible(al *AgentLoop, ts *turnState, exec *turnExecution) bool { + if al == nil || ts == nil || exec == nil { + return false + } + if !al.cfg.Agents.Defaults.UseFinalActionSummary() { + return false + } + if !exec.sawSteering { + return false + } + return len(exec.actionLog) >= 2 +} + +func joinActionTexts(records []TurnActionRecord) string { + parts := make([]string, 0, len(records)) + for _, rec := range records { + text := strings.TrimSpace(rec.Text) + if text == "" { + continue + } + if len(parts) > 0 && parts[len(parts)-1] == text { + continue + } + parts = append(parts, text) + } + return strings.Join(parts, "\n\n") +} + +func synthesizeFinalActionSummary( + ctx context.Context, + al *AgentLoop, + ts *turnState, + exec *turnExecution, + fallback string, +) string { + fallback = strings.TrimSpace(fallback) + if !actionSummaryEligible(al, ts, exec) { + return fallback + } + + records := make([]TurnActionRecord, 0, len(exec.actionLog)) + for _, rec := range exec.actionLog { + if strings.TrimSpace(rec.Text) == "" { + continue + } + records = append(records, rec) + } + if len(records) < 2 { + return fallback + } + + if fallback == "" { + fallback = joinActionTexts(records) + } + + provider := exec.activeProvider + model := strings.TrimSpace(exec.activeModel) + if provider == nil { + provider = ts.agent.Provider + } + if model == "" { + model = ts.agent.Model + } + if provider == nil || model == "" { + return fallback + } + + raw, err := json.MarshalIndent(records, "", " ") + if err != nil { + return fallback + } + + messages := []providers.Message{ + { + Role: "system", + Content: "You are formatting the final user-facing reply for a completed assistant turn.\n" + + "Use the same language as the action records.\n" + + "Summarize only the factual completed outcomes listed below.\n" + + "Omit pure progress chatter unless it states an actual completed outcome.\n" + + "Do not invent actions. Do not omit completed actions unless they are exact duplicates.\n" + + "Keep the reply concise and natural.", + }, + { + Role: "user", + Content: fmt.Sprintf( + "Action records for this completed turn:\n\n%s\n\nWrite one concise final reply for the user.", + string(raw), + ), + }, + } + + resp, err := provider.Chat(ctx, messages, nil, model, map[string]any{ + "max_tokens": 300, + "temperature": 0.2, + }) + if err != nil || resp == nil { + if err != nil { + logger.WarnCF("agent", "Final action summary synthesis failed", map[string]any{ + "agent_id": ts.agent.ID, + "error": err.Error(), + }) + } + return fallback + } + + content := strings.TrimSpace(resp.Content) + if content == "" { + content = strings.TrimSpace(resp.ReasoningContent) + } + if content == "" { + return fallback + } + return content +} diff --git a/pkg/agent/agent_test.go b/pkg/agent/agent_test.go index 7a869ec94..36d0d631b 100644 --- a/pkg/agent/agent_test.go +++ b/pkg/agent/agent_test.go @@ -1233,6 +1233,113 @@ func TestProcessMessage_HandledToolProcessesQueuedSteeringBeforeReturning(t *tes } } +type activitySummaryWithSteeringProvider struct { + calls int +} + +func (m *activitySummaryWithSteeringProvider) Chat( + ctx context.Context, + messages []providers.Message, + tools []providers.ToolDefinition, + model string, + opts map[string]any, +) (*providers.LLMResponse, error) { + m.calls++ + if len(messages) > 0 && messages[0].Role == "system" && + strings.Contains(messages[0].Content, "formatting the final user-facing reply") { + return &providers.LLMResponse{ + Content: "Записал.\n\nДобавил активности:\n- yoga — 30 мин\n- squats — 20 повторений", + }, nil + } + if m.calls == 1 { + return &providers.LLMResponse{ + Content: "Записал yoga — 30 мин.", + ToolCalls: []providers.ToolCall{{ + ID: "call_activity_steering", + Type: "function", + Name: "activity_with_steering_tool", + Arguments: map[string]any{}, + }}, + }, nil + } + + for _, msg := range messages { + if msg.Role == "user" && msg.Content == "и еще 20 приседаний" { + return &providers.LLMResponse{Content: "Записал squats — 20 повторений."}, nil + } + } + + return nil, fmt.Errorf("provider did not receive steering or synthesis prompt") +} + +func (m *activitySummaryWithSteeringProvider) GetDefaultModel() string { + return "activity-summary-with-steering-model" +} + +type activityWithSteeringTool struct { + loop *AgentLoop +} + +func (m *activityWithSteeringTool) Name() string { return "activity_with_steering_tool" } +func (m *activityWithSteeringTool) Description() string { + return "Queues a follow-up steering message after recording an activity" +} + +func (m *activityWithSteeringTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{}, + } +} + +func (m *activityWithSteeringTool) Execute(ctx context.Context, args map[string]any) *tools.ToolResult { + if err := m.loop.Steer(providers.Message{Role: "user", Content: "и еще 20 приседаний"}); err != nil { + return tools.ErrorResult(err.Error()).WithError(err) + } + return &tools.ToolResult{ + ForLLM: "activity recorded", + ForUser: "Записал yoga — 30 мин.", + } +} + +func TestProcessMessage_FinalActionSummarySynthesizesAcrossSteering(t *testing.T) { + tmpDir := t.TempDir() + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + FinalActionSummaryMode: "llm", + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &activitySummaryWithSteeringProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + al.RegisterTool(&activityWithSteeringTool{loop: al}) + + response, err := al.processMessage(context.Background(), testInboundMessage(bus.InboundMessage{ + Channel: "telegram", + ChatID: "chat1", + SenderID: "user1", + Content: "я позанимался йогой 30 минут", + })) + if err != nil { + t.Fatalf("processMessage() error = %v", err) + } + + want := "Записал.\n\nДобавил активности:\n- yoga — 30 мин\n- squats — 20 повторений" + if response != want { + t.Fatalf("response = %q, want %q", response, want) + } + if provider.calls != 3 { + t.Fatalf("expected 3 LLM calls including final synthesis, got %d", provider.calls) + } +} + func TestRunAgentLoop_ResponseHandledToolPublishesForUserWhenSendResponseDisabled(t *testing.T) { tmpDir := t.TempDir() cfg := config.DefaultConfig() diff --git a/pkg/agent/pipeline_execute.go b/pkg/agent/pipeline_execute.go index 0f71c7432..ef873629b 100644 --- a/pkg/agent/pipeline_execute.go +++ b/pkg/agent/pipeline_execute.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "time" "github.com/sipeed/picoclaw/pkg/bus" @@ -484,6 +485,11 @@ toolLoop: toolResult = tools.ErrorResult("hook returned nil tool result") } + toolSummary := strings.TrimSpace(toolResult.ForUser) + if toolSummary != "" { + exec.actionLog = appendTurnActionRecord(exec.actionLog, "tool_result", toolName, toolSummary, toolResult.IsError) + } + if len(toolResult.Media) > 0 && toolResult.ResponseHandled { parts := make([]bus.MediaPart, 0, len(toolResult.Media)) for _, ref := range toolResult.Media { @@ -655,6 +661,7 @@ toolLoop: // This covers the case where tools were partially executed and skipped due to steering, // but one tool had ResponseHandled=false (so allResponsesHandled=false). if len(exec.pendingMessages) > 0 { + exec.sawSteering = true logger.InfoCF("agent", "Pending steering after partial tool execution; continuing turn", map[string]any{ "agent_id": ts.agent.ID, @@ -667,6 +674,7 @@ toolLoop: // Poll for newly arrived steering if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 { + exec.sawSteering = true logger.InfoCF("agent", "Steering arrived after tool delivery; continuing turn", map[string]any{ "agent_id": ts.agent.ID, diff --git a/pkg/agent/pipeline_llm.go b/pkg/agent/pipeline_llm.go index 496fcd7e4..0f13f3b46 100644 --- a/pkg/agent/pipeline_llm.go +++ b/pkg/agent/pipeline_llm.go @@ -474,7 +474,9 @@ func (p *Pipeline) CallLLM( if responseContent == "" && exec.response.ReasoningContent != "" && ts.channel != "pico" { responseContent = exec.response.ReasoningContent } + exec.actionLog = appendTurnActionRecord(exec.actionLog, "assistant_direct", "", responseContent, false) if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 { + exec.sawSteering = true logger.InfoCF("agent", "Steering arrived after direct LLM response; continuing turn", map[string]any{ "agent_id": ts.agent.ID, diff --git a/pkg/agent/turn_coord.go b/pkg/agent/turn_coord.go index 2826e662c..72455f251 100644 --- a/pkg/agent/turn_coord.go +++ b/pkg/agent/turn_coord.go @@ -89,11 +89,13 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel // We do NOT call dequeueSteeringMessagesForScope here because // steering was already consumed from al.steering by ExecuteTools. if len(exec.pendingMessages) > 0 { + exec.sawSteering = true pendingMessages = append(pendingMessages, exec.pendingMessages...) exec.pendingMessages = nil } } else if !ts.opts.SkipInitialSteeringPoll { if steerMsgs := al.dequeueSteeringMessagesForScopeWithFallback(ts.sessionKey); len(steerMsgs) > 0 { + exec.sawSteering = true pendingMessages = append(pendingMessages, steerMsgs...) } } @@ -200,6 +202,7 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel if finalContent == "" { finalContent = ts.opts.DefaultResponse } + finalContent = synthesizeFinalActionSummary(turnCtx, al, ts, exec, finalContent) return pipeline.Finalize(ctx, turnCtx, ts, exec, turnStatus, finalContent) case ControlToolLoop: // Execute tools via Pipeline @@ -227,6 +230,7 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel if exec.allResponsesHandled { finalContent = "" } + finalContent = synthesizeFinalActionSummary(turnCtx, al, ts, exec, finalContent) return pipeline.Finalize(ctx, turnCtx, ts, exec, turnStatus, finalContent) } } @@ -244,6 +248,7 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel finalContent = ts.opts.DefaultResponse } } + finalContent = synthesizeFinalActionSummary(turnCtx, al, ts, exec, finalContent) // Check hard abort before finalizing (may have been set during tool execution) if ts.hardAbortRequested() { diff --git a/pkg/agent/turn_state.go b/pkg/agent/turn_state.go index b769ebcd0..33b61c3ad 100644 --- a/pkg/agent/turn_state.go +++ b/pkg/agent/turn_state.go @@ -118,6 +118,8 @@ type turnExecution struct { // Turn output finalContent string + actionLog []TurnActionRecord + sawSteering bool // Iteration tracking iteration int diff --git a/pkg/config/config.go b/pkg/config/config.go index c9d90e0f8..9211dbd1d 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -255,31 +255,32 @@ type ToolFeedbackConfig struct { } type AgentDefaults struct { - Workspace string `json:"workspace" env:"PICOCLAW_AGENTS_DEFAULTS_WORKSPACE"` - RestrictToWorkspace bool `json:"restrict_to_workspace" env:"PICOCLAW_AGENTS_DEFAULTS_RESTRICT_TO_WORKSPACE"` - AllowReadOutsideWorkspace bool `json:"allow_read_outside_workspace" env:"PICOCLAW_AGENTS_DEFAULTS_ALLOW_READ_OUTSIDE_WORKSPACE"` - Provider string `json:"provider" env:"PICOCLAW_AGENTS_DEFAULTS_PROVIDER"` - ModelName string `json:"model_name" env:"PICOCLAW_AGENTS_DEFAULTS_MODEL_NAME"` + Workspace string `json:"workspace" env:"PICOCLAW_AGENTS_DEFAULTS_WORKSPACE"` + RestrictToWorkspace bool `json:"restrict_to_workspace" env:"PICOCLAW_AGENTS_DEFAULTS_RESTRICT_TO_WORKSPACE"` + AllowReadOutsideWorkspace bool `json:"allow_read_outside_workspace" env:"PICOCLAW_AGENTS_DEFAULTS_ALLOW_READ_OUTSIDE_WORKSPACE"` + Provider string `json:"provider" env:"PICOCLAW_AGENTS_DEFAULTS_PROVIDER"` + ModelName string `json:"model_name" env:"PICOCLAW_AGENTS_DEFAULTS_MODEL_NAME"` ModelFallbacks []string `json:"model_fallbacks,omitempty"` - ImageModel string `json:"image_model,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_IMAGE_MODEL"` + ImageModel string `json:"image_model,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_IMAGE_MODEL"` ImageModelFallbacks []string `json:"image_model_fallbacks,omitempty"` - MaxTokens int `json:"max_tokens" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_TOKENS"` - ContextWindow int `json:"context_window,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_WINDOW"` - Temperature *float64 `json:"temperature,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_TEMPERATURE"` - MaxToolIterations int `json:"max_tool_iterations" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_TOOL_ITERATIONS"` - SummarizeMessageThreshold int `json:"summarize_message_threshold" env:"PICOCLAW_AGENTS_DEFAULTS_SUMMARIZE_MESSAGE_THRESHOLD"` - SummarizeTokenPercent int `json:"summarize_token_percent" env:"PICOCLAW_AGENTS_DEFAULTS_SUMMARIZE_TOKEN_PERCENT"` - MaxMediaSize int `json:"max_media_size,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_MEDIA_SIZE"` + MaxTokens int `json:"max_tokens" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_TOKENS"` + ContextWindow int `json:"context_window,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_WINDOW"` + Temperature *float64 `json:"temperature,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_TEMPERATURE"` + MaxToolIterations int `json:"max_tool_iterations" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_TOOL_ITERATIONS"` + SummarizeMessageThreshold int `json:"summarize_message_threshold" env:"PICOCLAW_AGENTS_DEFAULTS_SUMMARIZE_MESSAGE_THRESHOLD"` + SummarizeTokenPercent int `json:"summarize_token_percent" env:"PICOCLAW_AGENTS_DEFAULTS_SUMMARIZE_TOKEN_PERCENT"` + MaxMediaSize int `json:"max_media_size,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_MEDIA_SIZE"` Routing *RoutingConfig `json:"routing,omitempty"` - SteeringMode string `json:"steering_mode,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_STEERING_MODE"` // "one-at-a-time" (default) or "all" - MaxParallelTurns int `json:"max_parallel_turns,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_PARALLEL_TURNS"` // Max concurrent turns (0 or 1 = sequential) - SubTurn SubTurnConfig `json:"subturn" envPrefix:"PICOCLAW_AGENTS_DEFAULTS_SUBTURN_"` + SteeringMode string `json:"steering_mode,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_STEERING_MODE"` // "one-at-a-time" (default) or "all" + MaxParallelTurns int `json:"max_parallel_turns,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_PARALLEL_TURNS"` // Max concurrent turns (0 or 1 = sequential) + SubTurn SubTurnConfig `json:"subturn" envPrefix:"PICOCLAW_AGENTS_DEFAULTS_SUBTURN_"` ToolFeedback ToolFeedbackConfig `json:"tool_feedback,omitempty"` - SplitOnMarker bool `json:"split_on_marker" env:"PICOCLAW_AGENTS_DEFAULTS_SPLIT_ON_MARKER"` // split messages on <|[SPLIT]|> marker - ContextManager string `json:"context_manager,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER"` - ContextManagerConfig json.RawMessage `json:"context_manager_config,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER_CONFIG"` - MaxLLMRetries int `json:"max_llm_retries,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_LLM_RETRIES"` - LLMRetryBackoffSecs int `json:"llm_retry_backoff_secs,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_LLM_RETRY_BACKOFF_SECS"` + FinalActionSummaryMode string `json:"final_action_summary_mode,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_FINAL_ACTION_SUMMARY_MODE"` + SplitOnMarker bool `json:"split_on_marker" env:"PICOCLAW_AGENTS_DEFAULTS_SPLIT_ON_MARKER"` // split messages on <|[SPLIT]|> marker + ContextManager string `json:"context_manager,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER"` + ContextManagerConfig json.RawMessage `json:"context_manager_config,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_CONTEXT_MANAGER_CONFIG"` + MaxLLMRetries int `json:"max_llm_retries,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_LLM_RETRIES"` + LLMRetryBackoffSecs int `json:"llm_retry_backoff_secs,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_LLM_RETRY_BACKOFF_SECS"` } const DefaultMaxMediaSize = 20 * 1024 * 1024 // 20 MB @@ -311,6 +312,10 @@ func (d *AgentDefaults) IsToolFeedbackSeparateMessagesEnabled() bool { return d.ToolFeedback.SeparateMessages } +func (d *AgentDefaults) UseFinalActionSummary() bool { + return strings.EqualFold(strings.TrimSpace(d.FinalActionSummaryMode), "llm") +} + // GetModelName returns the effective model name for the agent defaults. // It prefers the new "model_name" field but falls back to "model" for backward compatibility. func (d *AgentDefaults) GetModelName() string { @@ -1017,7 +1022,11 @@ func LoadConfig(path string) (*Config, error) { } if e := json.Unmarshal(data, &versionInfo); e != nil { e = wrapJSONError(data, e, "config.json") - logger.ErrorCF("config", formatDiagnosticLogMessage("Malformed config file", e), map[string]any{"path": path}) + logger.ErrorCF( + "config", + formatDiagnosticLogMessage("Malformed config file", e), + map[string]any{"path": path}, + ) return nil, e } if len(data) <= 10 { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 8e2494ae5..4f0560081 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -39,9 +39,10 @@ func DefaultConfig() *Config { MaxArgsLength: 300, SeparateMessages: false, }, - SplitOnMarker: false, - MaxLLMRetries: 2, - LLMRetryBackoffSecs: 2, + FinalActionSummaryMode: "", + SplitOnMarker: false, + MaxLLMRetries: 2, + LLMRetryBackoffSecs: 2, }, }, Session: SessionConfig{ @@ -496,7 +497,11 @@ func defaultChannels() ChannelsConfig { "typing": map[string]any{"enabled": true}, "placeholder": map[string]any{"enabled": true, "text": []string{"Thinking... 💭"}}, "settings": map[string]any{ - "streaming": map[string]any{"enabled": true, "throttle_seconds": 3, "min_growth_chars": 200}, + "streaming": map[string]any{ + "enabled": true, + "throttle_seconds": 3, + "min_growth_chars": 200, + }, "use_markdown_v2": false, }, },