feat(agent): handle /btw as a side question outside session history
This commit is contained in:
parent
748ac58dd1
commit
89e4570126
7 changed files with 393 additions and 1 deletions
|
|
@ -641,6 +641,13 @@ func (al *AgentLoop) drainBusToSteering(ctx context.Context, activeScope, active
|
|||
// Transcribe audio if needed before steering, so the agent sees text.
|
||||
msg, _ = al.transcribeAudioInMessage(ctx, msg)
|
||||
|
||||
if handled, response := al.tryHandlePriorityCommand(ctx, msg); handled {
|
||||
if response != "" {
|
||||
al.PublishResponseIfNeeded(ctx, msg.Channel, msg.ChatID, response)
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
logger.InfoCF("agent", "Redirecting inbound message to steering queue",
|
||||
map[string]any{
|
||||
"channel": msg.Channel,
|
||||
|
|
@ -1339,6 +1346,95 @@ func (al *AgentLoop) ProcessHeartbeat(
|
|||
})
|
||||
}
|
||||
|
||||
func (al *AgentLoop) askSideQuestion(
|
||||
ctx context.Context,
|
||||
agent *AgentInstance,
|
||||
opts *processOptions,
|
||||
question string,
|
||||
) (string, error) {
|
||||
if agent == nil {
|
||||
return "", fmt.Errorf("no agent available for /btw")
|
||||
}
|
||||
|
||||
question = strings.TrimSpace(question)
|
||||
if question == "" {
|
||||
return "", fmt.Errorf("Usage: /btw <question>")
|
||||
}
|
||||
|
||||
var channel, chatID, senderID, senderDisplayName string
|
||||
var activeSkills []string
|
||||
if opts != nil {
|
||||
channel = opts.Channel
|
||||
chatID = opts.ChatID
|
||||
senderID = opts.SenderID
|
||||
senderDisplayName = opts.SenderDisplayName
|
||||
activeSkills = activeSkillNames(agent, *opts)
|
||||
}
|
||||
|
||||
messages := agent.ContextBuilder.BuildMessages(
|
||||
nil,
|
||||
"",
|
||||
question,
|
||||
nil,
|
||||
channel,
|
||||
chatID,
|
||||
senderID,
|
||||
senderDisplayName,
|
||||
activeSkills...,
|
||||
)
|
||||
|
||||
activeCandidates, activeModel, usedLight := al.selectCandidates(agent, question, messages)
|
||||
activeProvider := agent.Provider
|
||||
if usedLight && agent.LightProvider != nil {
|
||||
activeProvider = agent.LightProvider
|
||||
}
|
||||
|
||||
llmOpts := map[string]any{
|
||||
"max_tokens": agent.MaxTokens,
|
||||
"temperature": agent.Temperature,
|
||||
"prompt_cache_key": agent.ID + ":btw",
|
||||
}
|
||||
if agent.ThinkingLevel != ThinkingOff {
|
||||
if tc, ok := activeProvider.(providers.ThinkingCapable); ok && tc.SupportsThinking() {
|
||||
llmOpts["thinking_level"] = string(agent.ThinkingLevel)
|
||||
}
|
||||
}
|
||||
|
||||
callProvider := func(ctx context.Context, provider providers.LLMProvider, model string) (*providers.LLMResponse, error) {
|
||||
return provider.Chat(ctx, messages, nil, model, llmOpts)
|
||||
}
|
||||
|
||||
if len(activeCandidates) > 1 && al.fallback != nil {
|
||||
fbResult, err := al.fallback.Execute(
|
||||
ctx,
|
||||
activeCandidates,
|
||||
func(ctx context.Context, providerName, model string) (*providers.LLMResponse, error) {
|
||||
candidateProvider := activeProvider
|
||||
if cp, ok := agent.CandidateProviders[providers.ModelKey(providerName, model)]; ok {
|
||||
candidateProvider = cp
|
||||
}
|
||||
return callProvider(ctx, candidateProvider, model)
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if fbResult.Response == nil {
|
||||
return "", nil
|
||||
}
|
||||
return fbResult.Response.Content, nil
|
||||
}
|
||||
|
||||
resp, err := callProvider(ctx, activeProvider, activeModel)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if resp == nil {
|
||||
return "", nil
|
||||
}
|
||||
return resp.Content, nil
|
||||
}
|
||||
|
||||
func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (string, error) {
|
||||
// Add message preview to log (show full content for error messages)
|
||||
var logContent string
|
||||
|
|
@ -3490,6 +3586,14 @@ func (al *AgentLoop) buildCommandsRuntime(agent *AgentInstance, opts *processOpt
|
|||
if agent.ContextBuilder != nil {
|
||||
rt.ListSkillNames = agent.ContextBuilder.ListSkillNames
|
||||
}
|
||||
rt.AskSideQuestion = func(ctx context.Context, question string) (string, error) {
|
||||
question = strings.TrimSpace(question)
|
||||
if question == "" {
|
||||
return "", fmt.Errorf("Usage: /btw <question>")
|
||||
}
|
||||
|
||||
return al.askSideQuestion(ctx, agent, opts, question)
|
||||
}
|
||||
rt.GetModelInfo = func() (string, string) {
|
||||
return agent.Model, resolvedCandidateProvider(agent.Candidates, cfg.Agents.Defaults.Provider)
|
||||
}
|
||||
|
|
@ -3616,6 +3720,40 @@ func mapCommandError(result commands.ExecuteResult) string {
|
|||
return fmt.Sprintf("Failed to execute /%s: %v", result.Command, result.Err)
|
||||
}
|
||||
|
||||
func (al *AgentLoop) tryHandlePriorityCommand(ctx context.Context, msg bus.InboundMessage) (bool, string) {
|
||||
cmdName, ok := commands.CommandName(msg.Content)
|
||||
if !ok || cmdName != "btw" {
|
||||
return false, ""
|
||||
}
|
||||
|
||||
route, agent, err := al.resolveMessageRoute(msg)
|
||||
if err != nil || agent == nil {
|
||||
if err != nil {
|
||||
return true, fmt.Sprintf("Error processing message: %v", err)
|
||||
}
|
||||
return true, "Command unavailable in current context."
|
||||
}
|
||||
|
||||
opts := processOptions{
|
||||
SessionKey: resolveScopeKey(route, msg.SessionKey),
|
||||
Channel: msg.Channel,
|
||||
ChatID: msg.ChatID,
|
||||
MessageID: msg.MessageID,
|
||||
ReplyToMessageID: inboundMetadata(msg, metadataKeyReplyToMessage),
|
||||
SenderID: msg.SenderID,
|
||||
SenderDisplayName: msg.Sender.DisplayName,
|
||||
UserMessage: msg.Content,
|
||||
Media: msg.Media,
|
||||
DefaultResponse: defaultResponse,
|
||||
}
|
||||
|
||||
response, handled := al.handleCommand(ctx, msg, agent, &opts)
|
||||
if !handled {
|
||||
return false, ""
|
||||
}
|
||||
return true, response
|
||||
}
|
||||
|
||||
// extractPeer extracts the routing peer from the inbound message's structured Peer field.
|
||||
func extractPeer(msg bus.InboundMessage) *routing.RoutePeer {
|
||||
if msg.Peer.Kind == "" {
|
||||
|
|
|
|||
|
|
@ -228,6 +228,57 @@ func TestProcessMessage_UseCommandLoadsRequestedSkill(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestProcessMessage_BtwCommandRunsWithoutPersistingHistory(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
cfg := &config.Config{
|
||||
Agents: config.AgentsConfig{
|
||||
Defaults: config.AgentDefaults{
|
||||
Workspace: tmpDir,
|
||||
ModelName: "test-model",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &recordingProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
|
||||
msg := bus.InboundMessage{
|
||||
Channel: "telegram",
|
||||
SenderID: "telegram:123",
|
||||
ChatID: "chat-1",
|
||||
Content: "/btw explain side effects",
|
||||
}
|
||||
|
||||
response, err := al.processMessage(context.Background(), msg)
|
||||
if err != nil {
|
||||
t.Fatalf("processMessage() error = %v", err)
|
||||
}
|
||||
if response != "Mock response" {
|
||||
t.Fatalf("processMessage() response = %q, want %q", response, "Mock response")
|
||||
}
|
||||
if len(provider.lastMessages) == 0 {
|
||||
t.Fatal("provider did not receive any messages")
|
||||
}
|
||||
|
||||
lastMessage := provider.lastMessages[len(provider.lastMessages)-1]
|
||||
if lastMessage.Role != "user" || lastMessage.Content != "explain side effects" {
|
||||
t.Fatalf("last provider message = %+v, want stripped /btw question", lastMessage)
|
||||
}
|
||||
|
||||
route, _, err := al.resolveMessageRoute(msg)
|
||||
if err != nil {
|
||||
t.Fatalf("resolveMessageRoute() error = %v", err)
|
||||
}
|
||||
sessionKey := resolveScopeKey(route, msg.SessionKey)
|
||||
history := al.GetRegistry().GetDefaultAgent().Sessions.GetHistory(sessionKey)
|
||||
if len(history) != 0 {
|
||||
t.Fatalf("session history len = %d, want 0 for /btw", len(history))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCommand_UseCommandRejectsUnknownSkill(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
cfg := &config.Config{
|
||||
|
|
|
|||
|
|
@ -1013,6 +1013,114 @@ func TestAgentLoop_Steering_DirectResponseContinuesWithQueuedMessage(t *testing.
|
|||
}
|
||||
}
|
||||
|
||||
func TestAgentLoop_Steering_BtwCommandBypassesQueuedTurn(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-test-*")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create temp dir: %v", err)
|
||||
}
|
||||
defer os.RemoveAll(tmpDir)
|
||||
|
||||
cfg := &config.Config{
|
||||
Agents: config.AgentsConfig{
|
||||
Defaults: config.AgentDefaults{
|
||||
Workspace: tmpDir,
|
||||
ModelName: "test-model",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
provider := &blockingDirectProvider{
|
||||
firstStarted: make(chan struct{}),
|
||||
releaseFirst: make(chan struct{}),
|
||||
firstResp: "long turn finished",
|
||||
finalResp: "btw immediate reply",
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
|
||||
runCtx, cancelRun := context.WithCancel(context.Background())
|
||||
defer cancelRun()
|
||||
runErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
runErrCh <- al.Run(runCtx)
|
||||
}()
|
||||
|
||||
first := bus.InboundMessage{
|
||||
Channel: "test",
|
||||
SenderID: "user1",
|
||||
ChatID: "chat1",
|
||||
Content: "execute sleep 60, then send OK",
|
||||
Peer: bus.Peer{
|
||||
Kind: "direct",
|
||||
ID: "user1",
|
||||
},
|
||||
}
|
||||
btw := bus.InboundMessage{
|
||||
Channel: "test",
|
||||
SenderID: "user1",
|
||||
ChatID: "chat1",
|
||||
Content: "/btw 你说283928",
|
||||
Peer: bus.Peer{
|
||||
Kind: "direct",
|
||||
ID: "user1",
|
||||
},
|
||||
}
|
||||
|
||||
pubCtx, pubCancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer pubCancel()
|
||||
if err := msgBus.PublishInbound(pubCtx, first); err != nil {
|
||||
t.Fatalf("publish first inbound: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-provider.firstStarted:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timeout waiting for first LLM call to start")
|
||||
}
|
||||
|
||||
if err := msgBus.PublishInbound(pubCtx, btw); err != nil {
|
||||
t.Fatalf("publish /btw inbound: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case outbound := <-msgBus.OutboundChan():
|
||||
if outbound.Content != "btw immediate reply" {
|
||||
t.Fatalf("expected /btw reply before long turn completion, got %q", outbound.Content)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timeout waiting for /btw outbound response")
|
||||
}
|
||||
|
||||
sessionKey := routing.BuildAgentMainSessionKey(routing.DefaultAgentID)
|
||||
if msgs := al.dequeueSteeringMessagesForScope(sessionKey); len(msgs) != 0 {
|
||||
t.Fatalf("expected /btw to bypass steering queue, got %v", msgs)
|
||||
}
|
||||
|
||||
close(provider.releaseFirst)
|
||||
|
||||
select {
|
||||
case outbound := <-msgBus.OutboundChan():
|
||||
if outbound.Content != "long turn finished" {
|
||||
t.Fatalf("expected original turn response after release, got %q", outbound.Content)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timeout waiting for original turn response")
|
||||
}
|
||||
|
||||
cancelRun()
|
||||
select {
|
||||
case err := <-runErrCh:
|
||||
if err != nil {
|
||||
t.Fatalf("Run returned error: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timeout waiting for Run to stop")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentLoop_Continue_PreservesSteeringMedia(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-test-*")
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ func BuiltinDefinitions() []Definition {
|
|||
showCommand(),
|
||||
listCommand(),
|
||||
useCommand(),
|
||||
btwCommand(),
|
||||
switchCommand(),
|
||||
checkCommand(),
|
||||
clearCommand(),
|
||||
|
|
|
|||
|
|
@ -188,3 +188,55 @@ func TestBuiltinUseCommand_PassthroughsToAgentLogic(t *testing.T) {
|
|||
t.Fatalf("/use command=%q, want=%q", res.Command, "use")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuiltinBtwCommand_UsesSideQuestionRuntime(t *testing.T) {
|
||||
rt := &Runtime{
|
||||
AskSideQuestion: func(ctx context.Context, question string) (string, error) {
|
||||
if question != "what is 2+2?" {
|
||||
t.Fatalf("question=%q, want %q", question, "what is 2+2?")
|
||||
}
|
||||
return "4", nil
|
||||
},
|
||||
}
|
||||
defs := BuiltinDefinitions()
|
||||
ex := NewExecutor(NewRegistry(defs), rt)
|
||||
|
||||
var reply string
|
||||
res := ex.Execute(context.Background(), Request{
|
||||
Text: "/btw what is 2+2?",
|
||||
Reply: func(text string) error {
|
||||
reply = text
|
||||
return nil
|
||||
},
|
||||
})
|
||||
if res.Outcome != OutcomeHandled {
|
||||
t.Fatalf("/btw outcome=%v, want=%v", res.Outcome, OutcomeHandled)
|
||||
}
|
||||
if reply != "4" {
|
||||
t.Fatalf("/btw reply=%q, want=%q", reply, "4")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuiltinBtwCommand_MissingQuestion(t *testing.T) {
|
||||
defs := BuiltinDefinitions()
|
||||
ex := NewExecutor(NewRegistry(defs), &Runtime{
|
||||
AskSideQuestion: func(context.Context, string) (string, error) {
|
||||
return "", nil
|
||||
},
|
||||
})
|
||||
|
||||
var reply string
|
||||
res := ex.Execute(context.Background(), Request{
|
||||
Text: "/btw",
|
||||
Reply: func(text string) error {
|
||||
reply = text
|
||||
return nil
|
||||
},
|
||||
})
|
||||
if res.Outcome != OutcomeHandled {
|
||||
t.Fatalf("/btw outcome=%v, want=%v", res.Outcome, OutcomeHandled)
|
||||
}
|
||||
if reply != "Usage: /btw <question>" {
|
||||
t.Fatalf("/btw reply=%q, want usage message", reply)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
37
pkg/commands/cmd_btw.go
Normal file
37
pkg/commands/cmd_btw.go
Normal file
|
|
@ -0,0 +1,37 @@
|
|||
package commands
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func btwCommand() Definition {
|
||||
return Definition{
|
||||
Name: "btw",
|
||||
Description: "Ask a side question without changing session history",
|
||||
Usage: "/btw <question>",
|
||||
Handler: func(ctx context.Context, req Request, rt *Runtime) error {
|
||||
const emptyAnswerMsg = "The model returned an empty response. This may indicate a provider error or token limit."
|
||||
|
||||
if rt == nil || rt.AskSideQuestion == nil {
|
||||
return req.Reply(unavailableMsg)
|
||||
}
|
||||
|
||||
question := strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(req.Text), nthToken(req.Text, 0)))
|
||||
if question == "" {
|
||||
return req.Reply("Usage: /btw <question>")
|
||||
}
|
||||
|
||||
answer, err := rt.AskSideQuestion(ctx, question)
|
||||
if err != nil {
|
||||
return req.Reply(err.Error())
|
||||
}
|
||||
if strings.TrimSpace(answer) == "" {
|
||||
return req.Reply(emptyAnswerMsg)
|
||||
}
|
||||
|
||||
return req.Reply(answer)
|
||||
|
||||
},
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +1,10 @@
|
|||
package commands
|
||||
|
||||
import "github.com/sipeed/picoclaw/pkg/config"
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
)
|
||||
|
||||
// Runtime provides runtime dependencies to command handlers. It is constructed
|
||||
// per-request by the agent loop so that per-request state (like session scope)
|
||||
|
|
@ -8,6 +12,7 @@ import "github.com/sipeed/picoclaw/pkg/config"
|
|||
type Runtime struct {
|
||||
Config *config.Config
|
||||
GetModelInfo func() (name, provider string)
|
||||
AskSideQuestion func(ctx context.Context, question string) (string, error)
|
||||
ListAgentIDs func() []string
|
||||
ListDefinitions func() []Definition
|
||||
ListSkillNames func() []string
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue