diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md index 3bec847ba..81f8bcabe 100644 --- a/docs/guides/configuration.md +++ b/docs/guides/configuration.md @@ -136,6 +136,41 @@ Session scope controls how much memory is shared between chats, users, threads, For step-by-step recipes and isolation patterns, see the [Session Guide](session-guide.md). +### Final Turn Render + +`agents.defaults.final_turn_render_mode` controls an experimental final-response render pass for steering-heavy turns. + +When enabled with value `llm`, PicoClaw may do one extra **same-agent** LLM pass after tool execution has already completed: + +- it reuses the accumulated turn context +- it disables tool calling for that final pass +- it asks the same agent to answer the **full accumulated request chain**, not only the latest follow-up + +This is intended for multi-message turns such as: + +- `How much did I eat today?` +- `And yesterday?` +- `And the day before yesterday?` + +Config: + +```json +{ + "agents": { + "defaults": { + "final_turn_render_mode": "llm" + } + } +} +``` + +Notes: + +- omitted or empty: disabled +- `llm`: enable same-agent final no-tools render for eligible steering-heavy turns +- this setting is experimental and is mainly useful when follow-up messages often extend the same in-flight turn +- this is separate from channel/message delivery behavior; it affects only how the final reply text is rendered + ### Routing Routing is configured through `agents.dispatch.rules`. diff --git a/pkg/agent/action_summary.go b/pkg/agent/action_summary.go new file mode 100644 index 000000000..9b4bf027d --- /dev/null +++ b/pkg/agent/action_summary.go @@ -0,0 +1,194 @@ +package agent + +import ( + "context" + "encoding/json" + "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 finalTurnRenderEligible(al *AgentLoop, exec *turnExecution) bool { + if al == nil || exec == nil { + return false + } + if !al.cfg.Agents.Defaults.UseFinalTurnRender() { + return false + } + return exec.sawSteering +} + +func finalTurnRenderModel(ts *turnState, exec *turnExecution) (providers.LLMProvider, string) { + if exec != nil { + if exec.activeProvider != nil && strings.TrimSpace(exec.activeModel) != "" { + return exec.activeProvider, strings.TrimSpace(exec.activeModel) + } + if exec.activeProvider != nil { + return exec.activeProvider, strings.TrimSpace(ts.agent.Model) + } + } + if ts == nil || ts.agent == nil { + return nil, "" + } + return ts.agent.Provider, strings.TrimSpace(ts.agent.Model) +} + +func buildFinalTurnRenderInstruction(exec *turnExecution) string { + var b strings.Builder + b.WriteString("Write the final user-facing reply for this already-completed turn.\n") + b.WriteString("Use the same language and general style as the conversation.\n") + b.WriteString("Do not call tools.\n") + b.WriteString("Answer the full accumulated user request across this turn, not only the latest follow-up.\n") + b.WriteString("If a later follow-up clearly corrected, narrowed, or replaced an earlier request, follow the latest clarified intent.\n") + b.WriteString("If later follow-ups added to earlier requests, include the completed additive results together.\n") + b.WriteString("Use only the facts already present in the conversation and tool results. Do not invent missing results.\n") + b.WriteString("Keep the reply concise and natural.\n") + + if exec == nil || len(exec.actionLog) == 0 { + return b.String() + } + + 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) == 0 { + return b.String() + } + + raw, err := json.MarshalIndent(records, "", " ") + if err != nil { + return b.String() + } + b.WriteString("\nExplicit user-facing outcomes recorded during the turn:\n") + b.WriteString(string(raw)) + return b.String() +} + +func tryRenderFinalTurnReply( + ctx context.Context, + al *AgentLoop, + ts *turnState, + exec *turnExecution, + fallback string, +) (string, bool) { + fallback = strings.TrimSpace(fallback) + if !finalTurnRenderEligible(al, exec) { + return fallback, false + } + if exec == nil || len(exec.messages) == 0 { + return fallback, false + } + + provider, model := finalTurnRenderModel(ts, exec) + if provider == nil || model == "" { + return fallback, false + } + + messages := append([]providers.Message(nil), exec.messages...) + instruction := buildFinalTurnRenderInstruction(exec) + messages = append(messages, providers.Message{ + Role: "user", + Content: instruction, + }) + + opts := map[string]any{ + "max_tokens": min(ts.agent.MaxTokens, 800), + "temperature": 0.2, + "prompt_cache_key": ts.agent.ID, + } + + resp, err := provider.Chat(ctx, messages, nil, model, opts) + if err != nil || resp == nil { + if err != nil { + logger.WarnCF("agent", "Final turn render pass failed", map[string]any{ + "agent_id": ts.agent.ID, + "error": err.Error(), + }) + } + return fallback, false + } + + content := strings.TrimSpace(resp.Content) + if content == "" { + content = strings.TrimSpace(resp.ReasoningContent) + } + if content == "" { + return fallback, false + } + + logger.InfoCF("agent", "Rendered final reply from accumulated turn context", + map[string]any{ + "agent_id": ts.agent.ID, + "session_key": ts.sessionKey, + "messages_count": len(messages), + "action_record_count": len(exec.actionLog), + }) + return content, true +} + +func renderFinalTurnReply( + ctx context.Context, + al *AgentLoop, + ts *turnState, + exec *turnExecution, + fallback string, +) string { + content, ok := tryRenderFinalTurnReply(ctx, al, ts, exec, fallback) + if ok { + return content + } + return strings.TrimSpace(fallback) +} + +func shouldFinalizeAfterToolLoopWithRender(al *AgentLoop, exec *turnExecution) bool { + if !finalTurnRenderEligible(al, exec) { + return false + } + if exec == nil { + return false + } + return !exec.allResponsesHandled +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} diff --git a/pkg/agent/agent_test.go b/pkg/agent/agent_test.go index 7a869ec94..140677bf8 100644 --- a/pkg/agent/agent_test.go +++ b/pkg/agent/agent_test.go @@ -5661,3 +5661,288 @@ func (p *concurrentMockProvider) Chat( func (p *concurrentMockProvider) GetDefaultModel() string { return "test-model" } + +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 && tools == nil { + last := messages[len(messages)-1] + if last.Role == "user" && strings.Contains(last.Content, "already-completed turn") { + 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, + FinalTurnRenderMode: "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) + } +} + +type daySummaryAcrossSteeringProvider struct { + calls int +} + +func (p *daySummaryAcrossSteeringProvider) Chat( + ctx context.Context, + messages []providers.Message, + tools []providers.ToolDefinition, + model string, + opts map[string]any, +) (*providers.LLMResponse, error) { + p.calls++ + if len(messages) > 0 && tools == nil { + last := messages[len(messages)-1] + if last.Role == "user" && strings.Contains(last.Content, "already-completed turn") { + full := flattenMessageContents(messages) + if !strings.Contains(full, "today total: 428 kcal") || + !strings.Contains(full, "yesterday total: 1561 kcal") || + !strings.Contains(full, "day-before total: 1455 kcal") { + return nil, fmt.Errorf("final render pass missing accumulated tool results") + } + return &providers.LLMResponse{ + Content: "Коротко по итогам:\n- сегодня — 428 ккал\n- вчера — 1561 ккал\n- позавчера — 1455 ккал", + }, nil + } + } + + switch p.calls { + case 1: + return &providers.LLMResponse{ + Content: "", + ToolCalls: []providers.ToolCall{{ + ID: "call_day_today", + Type: "function", + Name: "day_summary_with_steering_tool", + Arguments: map[string]any{ + "day": "today", + }, + }}, + }, nil + case 2: + if !messageExists(messages, "А за вчера?") { + return nil, fmt.Errorf("provider did not receive yesterday steering") + } + return &providers.LLMResponse{ + Content: "", + ToolCalls: []providers.ToolCall{{ + ID: "call_day_yesterday", + Type: "function", + Name: "day_summary_with_steering_tool", + Arguments: map[string]any{ + "day": "yesterday", + }, + }}, + }, nil + case 3: + if !messageExists(messages, "И за позавчера?") { + return nil, fmt.Errorf("provider did not receive day-before steering") + } + return &providers.LLMResponse{ + Content: "", + ToolCalls: []providers.ToolCall{{ + ID: "call_day_before", + Type: "function", + Name: "day_summary_with_steering_tool", + Arguments: map[string]any{ + "day": "day_before", + }, + }}, + }, nil + default: + return nil, fmt.Errorf("unexpected provider call count %d", p.calls) + } +} + +func (p *daySummaryAcrossSteeringProvider) GetDefaultModel() string { + return "day-summary-across-steering-model" +} + +type daySummaryWithSteeringTool struct { + loop *AgentLoop +} + +func (t *daySummaryWithSteeringTool) Name() string { return "day_summary_with_steering_tool" } +func (t *daySummaryWithSteeringTool) Description() string { + return "Fetches one day summary and queues the next follow-up question" +} + +func (t *daySummaryWithSteeringTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "day": map[string]any{"type": "string"}, + }, + "required": []string{"day"}, + } +} + +func (t *daySummaryWithSteeringTool) Execute(ctx context.Context, args map[string]any) *tools.ToolResult { + day, _ := args["day"].(string) + switch day { + case "today": + if err := t.loop.Steer(providers.Message{Role: "user", Content: "А за вчера?"}); err != nil { + return tools.ErrorResult(err.Error()).WithError(err) + } + return &tools.ToolResult{ForLLM: "today total: 428 kcal"} + case "yesterday": + if err := t.loop.Steer(providers.Message{Role: "user", Content: "И за позавчера?"}); err != nil { + return tools.ErrorResult(err.Error()).WithError(err) + } + return &tools.ToolResult{ForLLM: "yesterday total: 1561 kcal"} + case "day_before": + return &tools.ToolResult{ForLLM: "day-before total: 1455 kcal"} + default: + return tools.ErrorResult("unknown day") + } +} + +func TestProcessMessage_FinalActionSummaryRendersAcrossInformationalSteering(t *testing.T) { + tmpDir := t.TempDir() + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + FinalTurnRenderMode: "llm", + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &daySummaryAcrossSteeringProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + al.RegisterTool(&daySummaryWithSteeringTool{loop: al}) + + response, err := al.processMessage(context.Background(), testInboundMessage(bus.InboundMessage{ + Channel: "telegram", + ChatID: "chat1", + SenderID: "user1", + Content: "А сколько я за сегодня съел?", + })) + if err != nil { + t.Fatalf("processMessage() error = %v", err) + } + + want := "Коротко по итогам:\n- сегодня — 428 ккал\n- вчера — 1561 ккал\n- позавчера — 1455 ккал" + if response != want { + t.Fatalf("response = %q, want %q", response, want) + } + if provider.calls != 4 { + t.Fatalf("expected 4 LLM calls including final render, got %d", provider.calls) + } +} + +func messageExists(messages []providers.Message, want string) bool { + for _, msg := range messages { + if msg.Role == "user" && msg.Content == want { + return true + } + } + return false +} + +func flattenMessageContents(messages []providers.Message) string { + parts := make([]string, 0, len(messages)) + for _, msg := range messages { + if strings.TrimSpace(msg.Content) == "" { + continue + } + parts = append(parts, msg.Content) + } + return strings.Join(parts, "\n") +} diff --git a/pkg/agent/pipeline_execute.go b/pkg/agent/pipeline_execute.go index 0f71c7432..527748669 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,10 @@ toolLoop: toolResult = tools.ErrorResult("hook returned nil tool result") } + if toolSummary := strings.TrimSpace(toolResult.ForUser); 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 { @@ -678,6 +683,16 @@ toolLoop: } // No pending steering: finalize or break depending on allResponsesHandled + if shouldFinalizeAfterToolLoopWithRender(al, exec) { + logger.InfoCF("agent", "Tool loop completed; rendering terminal reply from accumulated turn context", + map[string]any{ + "agent_id": ts.agent.ID, + "iteration": iteration, + "tool_count": len(normalizedToolCalls), + }) + return ToolControlFinalize + } + if exec.allResponsesHandled { summaryMsg := providers.Message{ Role: "assistant", diff --git a/pkg/agent/pipeline_llm.go b/pkg/agent/pipeline_llm.go index 496fcd7e4..fb9378dd2 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.markSteeringObserved() 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..18a5b5b55 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.markSteeringObserved() pendingMessages = append(pendingMessages, exec.pendingMessages...) exec.pendingMessages = nil } } else if !ts.opts.SkipInitialSteeringPoll { if steerMsgs := al.dequeueSteeringMessagesForScopeWithFallback(ts.sessionKey); len(steerMsgs) > 0 { + exec.markSteeringObserved() 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 = renderFinalTurnReply(turnCtx, al, ts, exec, finalContent) return pipeline.Finalize(ctx, turnCtx, ts, exec, turnStatus, finalContent) case ControlToolLoop: // Execute tools via Pipeline @@ -210,6 +213,25 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel // (added tool results/skipped messages) before returning ControlContinue messages = exec.messages continue + case ToolControlFinalize: + finalContent, rendered := tryRenderFinalTurnReply(turnCtx, al, ts, exec, finalContent) + if !rendered { + messages = exec.messages + continue + } + if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 { + exec.markSteeringObserved() + logger.InfoCF("agent", "Steering arrived during terminal render; continuing turn", + map[string]any{ + "agent_id": ts.agent.ID, + "iteration": iteration, + "steering_count": len(steerMsgs), + }) + exec.pendingMessages = append(exec.pendingMessages, steerMsgs...) + messages = exec.messages + continue + } + return pipeline.Finalize(ctx, turnCtx, ts, exec, turnStatus, finalContent) case ToolControlBreak: // Hard abort: delegate to abortTurn (sets TurnEndStatusAborted) if exec.abortedByHardAbort { @@ -227,6 +249,7 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel if exec.allResponsesHandled { finalContent = "" } + finalContent = renderFinalTurnReply(turnCtx, al, ts, exec, finalContent) return pipeline.Finalize(ctx, turnCtx, ts, exec, turnStatus, finalContent) } } @@ -244,6 +267,7 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel finalContent = ts.opts.DefaultResponse } } + finalContent = renderFinalTurnReply(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..229fef960 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 @@ -147,6 +149,13 @@ type turnExecution struct { abortedByHook bool // true when HookActionAbortTurn triggered } +func (e *turnExecution) markSteeringObserved() { + if e == nil { + return + } + e.sawSteering = true +} + // newTurnExecution creates a turnExecution initialized from turnState and options. func newTurnExecution( agent *AgentInstance, @@ -160,6 +169,7 @@ func newTurnExecution( summary: summary, messages: messages, pendingMessages: append([]providers.Message(nil), opts.InitialSteeringMessages...), + sawSteering: len(opts.InitialSteeringMessages) > 0, iteration: 0, phase: LLMPhaseSetup, } diff --git a/pkg/config/config.go b/pkg/config/config.go index c9d90e0f8..1e0d6b5c4 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -275,6 +275,7 @@ type AgentDefaults struct { 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"` + FinalTurnRenderMode string `json:"final_turn_render_mode,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_FINAL_TURN_RENDER_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"` @@ -311,6 +312,10 @@ func (d *AgentDefaults) IsToolFeedbackSeparateMessagesEnabled() bool { return d.ToolFeedback.SeparateMessages } +func (d *AgentDefaults) UseFinalTurnRender() bool { + return strings.EqualFold(strings.TrimSpace(d.FinalTurnRenderMode), "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 { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 8e2494ae5..a250d059d 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -39,6 +39,7 @@ func DefaultConfig() *Config { MaxArgsLength: 300, SeparateMessages: false, }, + FinalTurnRenderMode: "", SplitOnMarker: false, MaxLLMRetries: 2, LLMRetryBackoffSecs: 2,