refactor(agent): integrate Fantasy SDK, streaming, and MemGPT memory
Core agent loop migration to Fantasy SDK: - Replace providers.LLMProvider with fantasy.LanguageModel interface - Add streaming agent loop (runAgentLoopStreaming) with real-time token delivery via bus StreamDelta messages - Inject MemGPT working context into system prompt automatically - Register memory tool for agent self-managed persistent memory - Add MemGPT tool implementation (search/write/status operations) - Exponential backoff retry with jitter for transient LLM failures Supporting changes: - bus: add PublishStreamDelta for real-time token streaming - bus: add StreamDelta message type - channels: migrate from providers.Message to messages.Message - config: add ProgressiveDisclosure and memory configuration fields - heartbeat: update imports for messages package - cmd/main: wire Fantasy provider and language model creation Includes integration test scaffolding for the full agent loop.
This commit is contained in:
ZanzyTHEbar
parent
d982302fa6
commit
dd53be5640
12 changed files with 1261 additions and 323 deletions
|
|
@ -26,10 +26,10 @@ import (
|
|||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
"github.com/sipeed/picoclaw/pkg/cron"
|
||||
"github.com/sipeed/picoclaw/pkg/devices"
|
||||
picofantasy "github.com/sipeed/picoclaw/pkg/fantasy"
|
||||
"github.com/sipeed/picoclaw/pkg/heartbeat"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/migrate"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/skills"
|
||||
"github.com/sipeed/picoclaw/pkg/state"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
|
|
@ -508,14 +508,21 @@ func agentCmd() {
|
|||
os.Exit(1)
|
||||
}
|
||||
|
||||
provider, err := providers.CreateProvider(cfg)
|
||||
fantasyProvider, err := picofantasy.CreateProvider(cfg)
|
||||
if err != nil {
|
||||
fmt.Printf("Error creating provider: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
languageModel, err := fantasyProvider.LanguageModel(ctx, picofantasy.ModelID(cfg))
|
||||
if err != nil {
|
||||
fmt.Printf("Error creating language model: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
agentLoop := agent.NewAgentLoop(cfg, msgBus, provider)
|
||||
agentLoop := agent.NewAgentLoop(cfg, msgBus, languageModel)
|
||||
|
||||
// Print agent startup info (only for interactive mode)
|
||||
startupInfo := agentLoop.GetStartupInfo()
|
||||
|
|
@ -527,7 +534,6 @@ func agentCmd() {
|
|||
})
|
||||
|
||||
if message != "" {
|
||||
ctx := context.Background()
|
||||
response, err := agentLoop.ProcessDirect(ctx, message, sessionKey)
|
||||
if err != nil {
|
||||
fmt.Printf("Error: %v\n", err)
|
||||
|
|
@ -643,14 +649,20 @@ func gatewayCmd() {
|
|||
os.Exit(1)
|
||||
}
|
||||
|
||||
provider, err := providers.CreateProvider(cfg)
|
||||
fantasyProvider, err := picofantasy.CreateProvider(cfg)
|
||||
if err != nil {
|
||||
fmt.Printf("Error creating provider: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
languageModel, err := fantasyProvider.LanguageModel(context.Background(), picofantasy.ModelID(cfg))
|
||||
if err != nil {
|
||||
fmt.Printf("Error creating language model: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
agentLoop := agent.NewAgentLoop(cfg, msgBus, provider)
|
||||
agentLoop := agent.NewAgentLoop(cfg, msgBus, languageModel)
|
||||
|
||||
// Print agent startup info
|
||||
fmt.Println("\n📦 Agent Status:")
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
|
@ -9,7 +10,8 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/memory"
|
||||
"github.com/sipeed/picoclaw/pkg/messages"
|
||||
"github.com/sipeed/picoclaw/pkg/skills"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
)
|
||||
|
|
@ -17,7 +19,8 @@ import (
|
|||
type ContextBuilder struct {
|
||||
workspace string
|
||||
skillsLoader *skills.SkillsLoader
|
||||
memory *MemoryStore
|
||||
memory *MemoryStore // Legacy file-based memory
|
||||
memoryStore memory.Memory // New 3-tier MemGPT memory (may be nil)
|
||||
tools *tools.ToolRegistry // Direct reference to tool registry
|
||||
}
|
||||
|
||||
|
|
@ -48,6 +51,11 @@ func (cb *ContextBuilder) SetToolsRegistry(registry *tools.ToolRegistry) {
|
|||
cb.tools = registry
|
||||
}
|
||||
|
||||
// SetMemoryStore sets the 3-tier MemGPT memory store for working context injection.
|
||||
func (cb *ContextBuilder) SetMemoryStore(ms memory.Memory) {
|
||||
cb.memoryStore = ms
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) getIdentity() string {
|
||||
now := time.Now().Format("2006-01-02 15:04 (Monday)")
|
||||
workspacePath, _ := filepath.Abs(filepath.Join(cb.workspace))
|
||||
|
|
@ -128,12 +136,20 @@ The following skills extend your capabilities. To use a skill, read its SKILL.md
|
|||
%s`, skillsSummary))
|
||||
}
|
||||
|
||||
// Memory context
|
||||
// Legacy file-based memory context
|
||||
memoryContext := cb.memory.GetMemoryContext()
|
||||
if memoryContext != "" {
|
||||
parts = append(parts, "# Memory\n\n"+memoryContext)
|
||||
}
|
||||
|
||||
// 3-tier MemGPT working context injection
|
||||
if cb.memoryStore != nil {
|
||||
wcSection := cb.buildWorkingContextSection()
|
||||
if wcSection != "" {
|
||||
parts = append(parts, wcSection)
|
||||
}
|
||||
}
|
||||
|
||||
// Join with "---" separator
|
||||
return strings.Join(parts, "\n\n---\n\n")
|
||||
}
|
||||
|
|
@ -157,8 +173,46 @@ func (cb *ContextBuilder) LoadBootstrapFiles() string {
|
|||
return result
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) BuildMessages(history []providers.Message, summary string, currentMessage string, media []string, channel, chatID string) []providers.Message {
|
||||
messages := []providers.Message{}
|
||||
// buildWorkingContextSection returns the working context section for the system prompt.
|
||||
// It includes the hot-tier working context buffer and memory usage instructions.
|
||||
func (cb *ContextBuilder) buildWorkingContextSection() string {
|
||||
if cb.memoryStore == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
// Use a background context for system prompt building (non-blocking)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
|
||||
var parts []string
|
||||
|
||||
// Inject working context (hot tier)
|
||||
wc, err := cb.memoryStore.GetWorkingContext(ctx, "picoclaw", "default")
|
||||
if err == nil && wc != "" {
|
||||
parts = append(parts, "## Working Context\n\n"+wc)
|
||||
}
|
||||
|
||||
// Memory system instructions
|
||||
parts = append(parts, `## Memory System
|
||||
|
||||
You have a 3-tier memory system accessible via the **memory** tool:
|
||||
|
||||
- **Working Context** (hot): Always loaded. Use memory tool to update it with durable facts.
|
||||
- **Recall** (warm): Scored memory items. Search returns the most relevant.
|
||||
- **Archival** (cold): Large documents chunked and embedded for retrieval.
|
||||
|
||||
Use the memory tool to:
|
||||
- **search**: Find relevant memories across all tiers.
|
||||
- **write**: Store important facts, preferences, decisions.
|
||||
- **status**: Check memory pressure and usage.
|
||||
|
||||
Store important user preferences, key decisions, and facts you want to remember long-term.`)
|
||||
|
||||
return strings.Join(parts, "\n\n")
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) BuildMessages(history []messages.Message, summary string, currentMessage string, media []string, channel, chatID string) []messages.Message {
|
||||
msgs := []messages.Message{}
|
||||
|
||||
systemPrompt := cb.BuildSystemPrompt()
|
||||
|
||||
|
|
@ -189,49 +243,41 @@ func (cb *ContextBuilder) BuildMessages(history []providers.Message, summary str
|
|||
systemPrompt += "\n\n## Summary of Previous Conversation\n\n" + summary
|
||||
}
|
||||
|
||||
//This fix prevents the session memory from LLM failure due to elimination of toolu_IDs required from LLM
|
||||
// --- INICIO DEL FIX ---
|
||||
//Diegox-17
|
||||
for len(history) > 0 && (history[0].Role == "tool") {
|
||||
logger.DebugCF("agent", "Removing orphaned tool message from history to prevent LLM error",
|
||||
map[string]interface{}{"role": history[0].Role})
|
||||
history = history[1:]
|
||||
}
|
||||
//Diegox-17
|
||||
// --- FIN DEL FIX ---
|
||||
// Note: Orphaned tool messages are now prevented at the source by
|
||||
// tool-call-aware truncation in SessionManager.TruncateHistory().
|
||||
|
||||
messages = append(messages, providers.Message{
|
||||
msgs = append(msgs, messages.Message{
|
||||
Role: "system",
|
||||
Content: systemPrompt,
|
||||
})
|
||||
|
||||
messages = append(messages, history...)
|
||||
msgs = append(msgs, history...)
|
||||
|
||||
messages = append(messages, providers.Message{
|
||||
msgs = append(msgs, messages.Message{
|
||||
Role: "user",
|
||||
Content: currentMessage,
|
||||
})
|
||||
|
||||
return messages
|
||||
return msgs
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) AddToolResult(messages []providers.Message, toolCallID, toolName, result string) []providers.Message {
|
||||
messages = append(messages, providers.Message{
|
||||
func (cb *ContextBuilder) AddToolResult(msgs []messages.Message, toolCallID, toolName, result string) []messages.Message {
|
||||
msgs = append(msgs, messages.Message{
|
||||
Role: "tool",
|
||||
Content: result,
|
||||
ToolCallID: toolCallID,
|
||||
})
|
||||
return messages
|
||||
return msgs
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) AddAssistantMessage(messages []providers.Message, content string, toolCalls []map[string]interface{}) []providers.Message {
|
||||
msg := providers.Message{
|
||||
func (cb *ContextBuilder) AddAssistantMessage(msgs []messages.Message, content string, toolCalls []map[string]interface{}) []messages.Message {
|
||||
msg := messages.Message{
|
||||
Role: "assistant",
|
||||
Content: content,
|
||||
}
|
||||
// Always add assistant message, whether or not it has tool calls
|
||||
messages = append(messages, msg)
|
||||
return messages
|
||||
msgs = append(msgs, msg)
|
||||
return msgs
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) loadSkills() string {
|
||||
|
|
|
|||
560
pkg/agent/integration_test.go
Normal file
560
pkg/agent/integration_test.go
Normal file
|
|
@ -0,0 +1,560 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
fantasy "charm.land/fantasy"
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
)
|
||||
|
||||
// --- Mock language model that simulates tool calls ---
|
||||
|
||||
// toolCallingModel simulates an LLM that requests tool calls on first round,
|
||||
// then produces a final text response incorporating tool results.
|
||||
type toolCallingModel struct {
|
||||
callCount int
|
||||
}
|
||||
|
||||
func (m *toolCallingModel) Generate(_ context.Context, call fantasy.Call) (*fantasy.Response, error) {
|
||||
m.callCount++
|
||||
|
||||
// First call: check if any tool results already exist in the prompt.
|
||||
// If no tool results found, request a tool call.
|
||||
hasToolResults := false
|
||||
for _, msg := range call.Prompt {
|
||||
for _, part := range msg.Content {
|
||||
if part.GetType() == fantasy.ContentTypeToolResult {
|
||||
hasToolResults = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !hasToolResults && len(call.Tools) > 0 {
|
||||
// Request a tool call
|
||||
return &fantasy.Response{
|
||||
Content: fantasy.ResponseContent{
|
||||
fantasy.ToolCallContent{
|
||||
ToolCallID: "call-1",
|
||||
ToolName: "echo",
|
||||
Input: `{"text": "hello from tool"}`,
|
||||
},
|
||||
},
|
||||
FinishReason: fantasy.FinishReasonToolCalls,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// After tool results: produce final response
|
||||
return &fantasy.Response{
|
||||
Content: fantasy.ResponseContent{
|
||||
fantasy.TextContent{Text: "Integration test response with tool output"},
|
||||
},
|
||||
FinishReason: fantasy.FinishReasonStop,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *toolCallingModel) Stream(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
resp, err := m.Generate(context.Background(), call)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return func(yield func(fantasy.StreamPart) bool) {
|
||||
// Check if response has tool calls
|
||||
hasToolCalls := false
|
||||
for _, c := range resp.Content {
|
||||
if c.GetType() == fantasy.ContentTypeToolCall {
|
||||
hasToolCalls = true
|
||||
}
|
||||
}
|
||||
|
||||
if hasToolCalls {
|
||||
// Emit tool calls as stream parts
|
||||
for _, c := range resp.Content {
|
||||
if tc, ok := c.(fantasy.ToolCallContent); ok {
|
||||
if !yield(fantasy.StreamPart{
|
||||
Type: fantasy.StreamPartTypeToolCall,
|
||||
ID: tc.ToolCallID,
|
||||
ToolCallName: tc.ToolName,
|
||||
ToolCallInput: tc.Input,
|
||||
}) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonToolCalls}) {
|
||||
return
|
||||
}
|
||||
} else {
|
||||
// Emit text as proper stream sequence
|
||||
text := resp.Content.Text()
|
||||
if text != "" {
|
||||
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text-0"}) {
|
||||
return
|
||||
}
|
||||
// Emit text in chunks to simulate real streaming
|
||||
for i := 0; i < len(text); i += 10 {
|
||||
end := i + 10
|
||||
if end > len(text) {
|
||||
end = len(text)
|
||||
}
|
||||
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text-0", Delta: text[i:end]}) {
|
||||
return
|
||||
}
|
||||
}
|
||||
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text-0"}) {
|
||||
return
|
||||
}
|
||||
}
|
||||
yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop})
|
||||
}
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *toolCallingModel) GenerateObject(_ context.Context, _ fantasy.ObjectCall) (*fantasy.ObjectResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
func (m *toolCallingModel) StreamObject(_ context.Context, _ fantasy.ObjectCall) (fantasy.ObjectStreamResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
func (m *toolCallingModel) Provider() string { return "mock" }
|
||||
func (m *toolCallingModel) Model() string { return "mock-tool-model" }
|
||||
|
||||
// --- Simple echo tool for integration testing ---
|
||||
|
||||
type echoTool struct{}
|
||||
|
||||
func (t *echoTool) Name() string { return "echo" }
|
||||
func (t *echoTool) Description() string { return "Echo the given text back" }
|
||||
func (t *echoTool) Parameters() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
"type": "object",
|
||||
"properties": map[string]interface{}{
|
||||
"text": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Text to echo",
|
||||
},
|
||||
},
|
||||
"required": []string{"text"},
|
||||
}
|
||||
}
|
||||
|
||||
func (t *echoTool) Execute(_ context.Context, args map[string]interface{}) *tools.ToolResult {
|
||||
text, _ := args["text"].(string)
|
||||
return &tools.ToolResult{
|
||||
ForLLM: "Echo: " + text,
|
||||
ForUser: "",
|
||||
Silent: true,
|
||||
IsError: false,
|
||||
}
|
||||
}
|
||||
|
||||
// --- Integration Tests ---
|
||||
|
||||
// TestIntegration_FullAgentLoop_SimpleResponse tests the full agent loop
|
||||
// with a simple mock model that returns text directly (no tool calls).
|
||||
func TestIntegration_FullAgentLoop_SimpleResponse(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-integration-*")
|
||||
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,
|
||||
Model: "test-model",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("Hello from Fantasy agent")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
msg := bus.InboundMessage{
|
||||
Channel: "test",
|
||||
SenderID: "user1",
|
||||
ChatID: "chat1",
|
||||
Content: "Say hello",
|
||||
SessionKey: "test-session-simple",
|
||||
}
|
||||
|
||||
response, err := al.processMessage(ctx, msg)
|
||||
if err != nil {
|
||||
t.Fatalf("processMessage failed: %v", err)
|
||||
}
|
||||
|
||||
if response != "Hello from Fantasy agent" {
|
||||
t.Errorf("Expected 'Hello from Fantasy agent', got: %s", response)
|
||||
}
|
||||
|
||||
// Verify session has messages saved
|
||||
history := al.sessions.GetHistory("test-session-simple")
|
||||
if len(history) == 0 {
|
||||
t.Error("Expected session history to have messages")
|
||||
}
|
||||
|
||||
// First message should be the user's
|
||||
foundUser := false
|
||||
for _, m := range history {
|
||||
if m.Role == "user" && m.Content == "Say hello" {
|
||||
foundUser = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !foundUser {
|
||||
t.Error("Expected user message in session history")
|
||||
}
|
||||
}
|
||||
|
||||
// TestIntegration_FullAgentLoop_WithToolCalls tests the full agent loop
|
||||
// including tool call execution and response incorporation.
|
||||
func TestIntegration_FullAgentLoop_WithToolCalls(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-integration-tools-*")
|
||||
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,
|
||||
Model: "mock-tool-model",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := &toolCallingModel{}
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Register the echo tool
|
||||
al.RegisterTool(&echoTool{})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
msg := bus.InboundMessage{
|
||||
Channel: "test",
|
||||
SenderID: "user1",
|
||||
ChatID: "chat1",
|
||||
Content: "Use the echo tool",
|
||||
SessionKey: "test-session-tools",
|
||||
}
|
||||
|
||||
response, err := al.processMessage(ctx, msg)
|
||||
if err != nil {
|
||||
t.Fatalf("processMessage failed: %v", err)
|
||||
}
|
||||
|
||||
// The model returns "Integration test response with tool output" after tool execution
|
||||
if !strings.Contains(response, "Integration test response") {
|
||||
t.Errorf("Expected response to contain 'Integration test response', got: %s", response)
|
||||
}
|
||||
|
||||
// Model should have been called at least twice (tool call + final response)
|
||||
if model.callCount < 2 {
|
||||
t.Errorf("Expected model to be called at least 2 times, got: %d", model.callCount)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIntegration_ProcessDirect tests the ProcessDirect method
|
||||
// which is used by CLI mode for one-shot message processing.
|
||||
func TestIntegration_ProcessDirect(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-integration-direct-*")
|
||||
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,
|
||||
Model: "test-model",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("Direct CLI response")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
response, err := al.ProcessDirect(ctx, "Direct message", "direct-session")
|
||||
if err != nil {
|
||||
t.Fatalf("ProcessDirect failed: %v", err)
|
||||
}
|
||||
|
||||
if response != "Direct CLI response" {
|
||||
t.Errorf("Expected 'Direct CLI response', got: %s", response)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Streaming-specific mock ---
|
||||
|
||||
// streamingModel simulates an LLM with proper streaming token emission.
|
||||
// It emits tokens one word at a time via Stream() and also supports Generate()
|
||||
// for non-streaming fallback.
|
||||
type streamingModel struct {
|
||||
words []string
|
||||
}
|
||||
|
||||
func newStreamingModel(text string) *streamingModel {
|
||||
return &streamingModel{words: strings.Fields(text)}
|
||||
}
|
||||
|
||||
func (m *streamingModel) Generate(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
fullText := strings.Join(m.words, " ")
|
||||
return &fantasy.Response{
|
||||
Content: fantasy.ResponseContent{fantasy.TextContent{Text: fullText}},
|
||||
FinishReason: fantasy.FinishReasonStop,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *streamingModel) Stream(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
words := m.words
|
||||
return func(yield func(fantasy.StreamPart) bool) {
|
||||
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "s-0"}) {
|
||||
return
|
||||
}
|
||||
for i, w := range words {
|
||||
delta := w
|
||||
if i < len(words)-1 {
|
||||
delta += " "
|
||||
}
|
||||
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "s-0", Delta: delta}) {
|
||||
return
|
||||
}
|
||||
}
|
||||
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "s-0"}) {
|
||||
return
|
||||
}
|
||||
yield(fantasy.StreamPart{
|
||||
Type: fantasy.StreamPartTypeFinish,
|
||||
FinishReason: fantasy.FinishReasonStop,
|
||||
Usage: fantasy.Usage{InputTokens: 10, OutputTokens: int64(len(words)), TotalTokens: 10 + int64(len(words))},
|
||||
})
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *streamingModel) GenerateObject(_ context.Context, _ fantasy.ObjectCall) (*fantasy.ObjectResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
func (m *streamingModel) StreamObject(_ context.Context, _ fantasy.ObjectCall) (fantasy.ObjectStreamResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
func (m *streamingModel) Provider() string { return "mock" }
|
||||
func (m *streamingModel) Model() string { return "streaming-mock" }
|
||||
|
||||
// TestIntegration_Streaming_TextDeltas tests that the streaming agent loop
|
||||
// publishes text deltas to the bus and returns the complete text.
|
||||
func TestIntegration_Streaming_TextDeltas(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-integration-stream-*")
|
||||
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,
|
||||
Model: "streaming-mock",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newStreamingModel("Hello from streaming agent response")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Collect stream deltas in background
|
||||
var deltas []string
|
||||
var deltaDone = make(chan struct{})
|
||||
go func() {
|
||||
defer close(deltaDone)
|
||||
for {
|
||||
msg, ok := msgBus.SubscribeOutbound(ctx)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if msg.StreamDelta {
|
||||
deltas = append(deltas, msg.Content)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Process with streaming
|
||||
response, err := al.ProcessDirectStreaming(ctx, "Stream me", "stream-session", "test", "chat-1")
|
||||
if err != nil {
|
||||
t.Fatalf("ProcessDirectStreaming failed: %v", err)
|
||||
}
|
||||
|
||||
// Cancel to stop delta collector
|
||||
cancel()
|
||||
<-deltaDone
|
||||
|
||||
// Verify complete response
|
||||
if response != "Hello from streaming agent response" {
|
||||
t.Errorf("Expected 'Hello from streaming agent response', got: %s", response)
|
||||
}
|
||||
|
||||
// Verify stream deltas were published
|
||||
if len(deltas) == 0 {
|
||||
t.Error("Expected stream deltas to be published to bus")
|
||||
}
|
||||
|
||||
// Reconstruct full text from deltas
|
||||
fullFromDeltas := strings.Join(deltas, "")
|
||||
if fullFromDeltas != "Hello from streaming agent response" {
|
||||
t.Errorf("Delta reconstruction mismatch: got '%s'", fullFromDeltas)
|
||||
}
|
||||
|
||||
// Verify session was saved
|
||||
history := al.sessions.GetHistory("stream-session")
|
||||
if len(history) < 2 { // user + assistant
|
||||
t.Errorf("Expected at least 2 messages in session, got %d", len(history))
|
||||
}
|
||||
}
|
||||
|
||||
// TestIntegration_Streaming_WithToolCalls tests streaming with a model
|
||||
// that requests tool calls before producing a final streamed response.
|
||||
func TestIntegration_Streaming_WithToolCalls(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-integration-stream-tools-*")
|
||||
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,
|
||||
Model: "mock-tool-model",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := &toolCallingModel{}
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al.RegisterTool(&echoTool{})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Collect deltas
|
||||
var deltas []string
|
||||
var deltaDone = make(chan struct{})
|
||||
go func() {
|
||||
defer close(deltaDone)
|
||||
for {
|
||||
msg, ok := msgBus.SubscribeOutbound(ctx)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if msg.StreamDelta {
|
||||
deltas = append(deltas, msg.Content)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
response, err := al.ProcessDirectStreaming(ctx, "Use the echo tool (streaming)", "stream-tools-session", "test", "chat-1")
|
||||
cancel()
|
||||
<-deltaDone
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("ProcessDirectStreaming with tools failed: %v", err)
|
||||
}
|
||||
|
||||
if !strings.Contains(response, "Integration test response") {
|
||||
t.Errorf("Expected response containing 'Integration test response', got: %s", response)
|
||||
}
|
||||
|
||||
// Model should have been called at least twice (tool call step + final response step)
|
||||
if model.callCount < 2 {
|
||||
t.Errorf("Expected at least 2 model calls, got %d", model.callCount)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIntegration_MultipleMessages tests sequential message processing
|
||||
// to verify session history accumulation.
|
||||
func TestIntegration_MultipleMessages(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "agent-integration-multi-*")
|
||||
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,
|
||||
Model: "test-model",
|
||||
MaxTokens: 4096,
|
||||
MaxToolIterations: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("Response")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
sessionKey := "multi-msg-session"
|
||||
ctx := context.Background()
|
||||
|
||||
// Send 3 messages
|
||||
for i := 0; i < 3; i++ {
|
||||
msg := bus.InboundMessage{
|
||||
Channel: "test",
|
||||
SenderID: "user1",
|
||||
ChatID: "chat1",
|
||||
Content: fmt.Sprintf("Message %d", i+1),
|
||||
SessionKey: sessionKey,
|
||||
}
|
||||
|
||||
_, err := al.processMessage(ctx, msg)
|
||||
if err != nil {
|
||||
t.Fatalf("processMessage #%d failed: %v", i+1, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify session history grew
|
||||
history := al.sessions.GetHistory(sessionKey)
|
||||
if len(history) < 6 { // At least 3 user messages + 3 assistant messages
|
||||
t.Errorf("Expected at least 6 messages in history, got: %d", len(history))
|
||||
}
|
||||
}
|
||||
|
|
@ -8,7 +8,6 @@ package agent
|
|||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
|
@ -17,11 +16,15 @@ import (
|
|||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
fantasy "charm.land/fantasy"
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
"github.com/sipeed/picoclaw/pkg/constants"
|
||||
picofantasy "github.com/sipeed/picoclaw/pkg/fantasy"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/memory/delegate"
|
||||
memstore "github.com/sipeed/picoclaw/pkg/memory/store"
|
||||
"github.com/sipeed/picoclaw/pkg/messages"
|
||||
"github.com/sipeed/picoclaw/pkg/session"
|
||||
"github.com/sipeed/picoclaw/pkg/state"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
|
|
@ -30,7 +33,7 @@ import (
|
|||
|
||||
type AgentLoop struct {
|
||||
bus *bus.MessageBus
|
||||
provider providers.LLMProvider
|
||||
languageModel fantasy.LanguageModel
|
||||
workspace string
|
||||
model string
|
||||
contextWindow int // Maximum context window size in tokens
|
||||
|
|
@ -39,8 +42,11 @@ type AgentLoop struct {
|
|||
state *state.Manager
|
||||
contextBuilder *ContextBuilder
|
||||
tools *tools.ToolRegistry
|
||||
memoryStore *memstore.MemoryStore // 3-tier MemGPT memory (nil if init failed)
|
||||
running atomic.Bool
|
||||
summarizing sync.Map // Tracks which sessions are currently being summarized
|
||||
summarizeFailures sync.Map // Tracks consecutive summarization failures per session (string -> int)
|
||||
cfg *config.Config // Stored for subagent factory access
|
||||
}
|
||||
|
||||
// processOptions configures how a message is processed
|
||||
|
|
@ -53,6 +59,7 @@ type processOptions struct {
|
|||
EnableSummary bool // Whether to trigger summarization
|
||||
SendResponse bool // Whether to send response via bus
|
||||
NoHistory bool // If true, don't load session history (for heartbeat)
|
||||
Streaming bool // If true, stream token deltas to bus via OnTextDelta
|
||||
}
|
||||
|
||||
// createToolRegistry creates a tool registry with common tools.
|
||||
|
|
@ -101,7 +108,7 @@ func createToolRegistry(workspace string, restrict bool, cfg *config.Config, msg
|
|||
return registry
|
||||
}
|
||||
|
||||
func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers.LLMProvider) *AgentLoop {
|
||||
func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, model fantasy.LanguageModel) *AgentLoop {
|
||||
workspace := cfg.WorkspacePath()
|
||||
os.MkdirAll(workspace, 0755)
|
||||
|
||||
|
|
@ -111,7 +118,7 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers
|
|||
toolsRegistry := createToolRegistry(workspace, restrict, cfg, msgBus)
|
||||
|
||||
// Create subagent manager with its own tool registry
|
||||
subagentManager := tools.NewSubagentManager(provider, cfg.Agents.Defaults.Model, workspace, msgBus)
|
||||
subagentManager := tools.NewSubagentManager(model, cfg.Agents.Defaults.Model, workspace, msgBus)
|
||||
subagentTools := createToolRegistry(workspace, restrict, cfg, msgBus)
|
||||
// Subagent doesn't need spawn/subagent tools to avoid recursion
|
||||
subagentManager.SetTools(subagentTools)
|
||||
|
|
@ -133,18 +140,66 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers
|
|||
contextBuilder := NewContextBuilder(workspace)
|
||||
contextBuilder.SetToolsRegistry(toolsRegistry)
|
||||
|
||||
// Initialize 3-tier MemGPT memory system
|
||||
var ms *memstore.MemoryStore
|
||||
memDBPath := filepath.Join(workspace, "memory", "picoclaw.db")
|
||||
os.MkdirAll(filepath.Dir(memDBPath), 0755)
|
||||
|
||||
del, err := delegate.NewLibSQLDelegate(memDBPath)
|
||||
if err != nil {
|
||||
logger.WarnCF("agent", "Failed to create memory delegate, memory system disabled",
|
||||
map[string]interface{}{"error": err.Error()})
|
||||
} else {
|
||||
if err := del.Init(context.Background()); err != nil {
|
||||
logger.WarnCF("agent", "Failed to init memory schema, memory system disabled",
|
||||
map[string]interface{}{"error": err.Error()})
|
||||
del.Close()
|
||||
} else {
|
||||
chunker := memstore.NewMarkdownChunker(memstore.DefaultMarkdownChunkerConfig())
|
||||
ms = memstore.New(del, chunker, nil, memstore.Config{
|
||||
ContextWindowTokens: cfg.Agents.Defaults.MaxTokens,
|
||||
OffloadThresholdTokens: 4000,
|
||||
})
|
||||
contextBuilder.SetMemoryStore(ms)
|
||||
|
||||
// Register the memory tool
|
||||
memTool := NewMemGPTTool(ms, "picoclaw", "default")
|
||||
toolsRegistry.Register(memTool)
|
||||
|
||||
logger.InfoCF("agent", "3-tier memory system initialized",
|
||||
map[string]interface{}{"db_path": memDBPath})
|
||||
}
|
||||
}
|
||||
|
||||
// Register meta-tools for progressive disclosure (tool_search + tool_call)
|
||||
toolsRegistry.RegisterMetaTools()
|
||||
|
||||
// If memory tool is a gateway, mark it visible in progressive mode
|
||||
if ms != nil {
|
||||
toolsRegistry.MarkGateway("memory")
|
||||
}
|
||||
|
||||
// Apply progressive disclosure config
|
||||
if cfg.Tools.ProgressiveDisclosure {
|
||||
toolsRegistry.SetProgressiveDisclosure(true)
|
||||
logger.InfoCF("agent", "Progressive tool disclosure enabled",
|
||||
map[string]interface{}{"gateway_tools": toolsRegistry.ListVisible()})
|
||||
}
|
||||
|
||||
return &AgentLoop{
|
||||
bus: msgBus,
|
||||
provider: provider,
|
||||
languageModel: model,
|
||||
workspace: workspace,
|
||||
model: cfg.Agents.Defaults.Model,
|
||||
contextWindow: cfg.Agents.Defaults.MaxTokens, // Restore context window for summarization
|
||||
contextWindow: cfg.Agents.Defaults.MaxTokens,
|
||||
maxIterations: cfg.Agents.Defaults.MaxToolIterations,
|
||||
sessions: sessionsManager,
|
||||
state: stateManager,
|
||||
contextBuilder: contextBuilder,
|
||||
tools: toolsRegistry,
|
||||
memoryStore: ms,
|
||||
summarizing: sync.Map{},
|
||||
cfg: cfg,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -192,6 +247,9 @@ func (al *AgentLoop) Run(ctx context.Context) error {
|
|||
|
||||
func (al *AgentLoop) Stop() {
|
||||
al.running.Store(false)
|
||||
if al.memoryStore != nil {
|
||||
al.memoryStore.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (al *AgentLoop) RegisterTool(tool tools.Tool) {
|
||||
|
|
@ -226,6 +284,29 @@ func (al *AgentLoop) ProcessDirectWithChannel(ctx context.Context, content, sess
|
|||
return al.processMessage(ctx, msg)
|
||||
}
|
||||
|
||||
// ProcessDirectStreaming processes a message with streaming token delivery.
|
||||
// Text deltas are published to the bus as StreamDelta messages in real time.
|
||||
func (al *AgentLoop) ProcessDirectStreaming(ctx context.Context, content, sessionKey, channel, chatID string) (string, error) {
|
||||
msg := bus.InboundMessage{
|
||||
Channel: channel,
|
||||
SenderID: "user",
|
||||
ChatID: chatID,
|
||||
Content: content,
|
||||
SessionKey: sessionKey,
|
||||
}
|
||||
|
||||
return al.runAgentLoop(ctx, processOptions{
|
||||
SessionKey: msg.SessionKey,
|
||||
Channel: msg.Channel,
|
||||
ChatID: msg.ChatID,
|
||||
UserMessage: msg.Content,
|
||||
DefaultResponse: "I've completed processing but have no response to give.",
|
||||
EnableSummary: true,
|
||||
SendResponse: false,
|
||||
Streaming: true,
|
||||
})
|
||||
}
|
||||
|
||||
// ProcessHeartbeat processes a heartbeat request without session history.
|
||||
// Each heartbeat is independent and doesn't accumulate context.
|
||||
func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, chatID string) (string, error) {
|
||||
|
|
@ -327,11 +408,14 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe
|
|||
}
|
||||
|
||||
// runAgentLoop is the core message processing logic.
|
||||
// It handles context building, LLM calls, tool execution, and response handling.
|
||||
// It handles context building, Fantasy agent creation, tool execution, and response handling.
|
||||
// When opts.Streaming is true, delegates to runAgentLoopStreaming for real-time token delivery.
|
||||
func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (string, error) {
|
||||
if opts.Streaming {
|
||||
return al.runAgentLoopStreaming(ctx, opts)
|
||||
}
|
||||
// 0. Record last channel for heartbeat notifications (skip internal channels)
|
||||
if opts.Channel != "" && opts.ChatID != "" {
|
||||
// Don't record internal channels (cli, system, subagent)
|
||||
if !constants.IsInternalChannel(opts.Channel) {
|
||||
channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID)
|
||||
if err := al.RecordLastChannel(channelKey); err != nil {
|
||||
|
|
@ -344,13 +428,13 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
al.updateToolContexts(opts.Channel, opts.ChatID)
|
||||
|
||||
// 2. Build messages (skip history for heartbeat)
|
||||
var history []providers.Message
|
||||
var history []messages.Message
|
||||
var summary string
|
||||
if !opts.NoHistory {
|
||||
history = al.sessions.GetHistory(opts.SessionKey)
|
||||
summary = al.sessions.GetSummary(opts.SessionKey)
|
||||
}
|
||||
messages := al.contextBuilder.BuildMessages(
|
||||
builtMsgs := al.contextBuilder.BuildMessages(
|
||||
history,
|
||||
summary,
|
||||
opts.UserMessage,
|
||||
|
|
@ -362,30 +446,89 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
// 3. Save user message to session
|
||||
al.sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage)
|
||||
|
||||
// 4. Run LLM iteration loop
|
||||
finalContent, iteration, err := al.runLLMIteration(ctx, messages, opts)
|
||||
if err != nil {
|
||||
return "", err
|
||||
// 4. Split built messages into system prompt, conversation history, and current user prompt.
|
||||
// BuildMessages returns: [system, ...history, user]
|
||||
systemPrompt := ""
|
||||
var historyMsgs []messages.Message
|
||||
userPrompt := opts.UserMessage
|
||||
|
||||
if len(builtMsgs) > 0 && builtMsgs[0].Role == "system" {
|
||||
systemPrompt = builtMsgs[0].Content
|
||||
// History is everything between system and last user message.
|
||||
if len(builtMsgs) > 2 {
|
||||
historyMsgs = builtMsgs[1 : len(builtMsgs)-1]
|
||||
}
|
||||
}
|
||||
|
||||
// If last tool had ForUser content and we already sent it, we might not need to send final response
|
||||
// This is controlled by the tool's Silent flag and ForUser content
|
||||
// 5. Convert history to Fantasy message format
|
||||
fantasyHistory := picofantasy.MessagesToFantasy(historyMsgs)
|
||||
|
||||
// 5. Handle empty response
|
||||
// 6. Build adapted tools from PicoClaw registry (with optional offloading)
|
||||
adaptCfg := picofantasy.AdaptedToolsConfig{
|
||||
MemStore: al.memoryStore,
|
||||
AgentID: "picoclaw",
|
||||
SessionKey: opts.SessionKey,
|
||||
}
|
||||
adaptedTools := picofantasy.BuildAdaptedTools(al.tools, al.bus, opts.Channel, opts.ChatID, adaptCfg)
|
||||
|
||||
// 7. Create Fantasy agent with tools and configuration
|
||||
agentOpts := []fantasy.AgentOption{
|
||||
fantasy.WithTools(adaptedTools...),
|
||||
fantasy.WithStopConditions(fantasy.StepCountIs(al.maxIterations)),
|
||||
}
|
||||
if systemPrompt != "" {
|
||||
agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt))
|
||||
}
|
||||
agent := fantasy.NewAgent(al.languageModel, agentOpts...)
|
||||
|
||||
logger.DebugCF("agent", "Fantasy agent created",
|
||||
map[string]interface{}{
|
||||
"model": al.model,
|
||||
"tools_count": len(adaptedTools),
|
||||
"history_count": len(historyMsgs),
|
||||
"max_iterations": al.maxIterations,
|
||||
"memory_enabled": al.memoryStore != nil,
|
||||
})
|
||||
|
||||
// 8. Call Fantasy agent.Generate()
|
||||
result, err := agent.Generate(ctx, fantasy.AgentCall{
|
||||
Prompt: userPrompt,
|
||||
Messages: fantasyHistory,
|
||||
})
|
||||
if err != nil {
|
||||
logger.ErrorCF("agent", "Fantasy Generate failed",
|
||||
map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", fmt.Errorf("agent Generate failed: %w", err)
|
||||
}
|
||||
|
||||
// 9. Save all step messages to session
|
||||
stepCount := len(result.Steps)
|
||||
for _, step := range result.Steps {
|
||||
stepMsgs := picofantasy.StepToMessages(step)
|
||||
for _, m := range stepMsgs {
|
||||
al.sessions.AddFullMessage(opts.SessionKey, m)
|
||||
}
|
||||
}
|
||||
|
||||
// 10. Extract final text
|
||||
finalContent := result.Response.Content.Text()
|
||||
|
||||
// 11. Handle empty response
|
||||
if finalContent == "" {
|
||||
finalContent = opts.DefaultResponse
|
||||
}
|
||||
|
||||
// 6. Save final assistant message to session
|
||||
al.sessions.AddMessage(opts.SessionKey, "assistant", finalContent)
|
||||
// 12. Save session
|
||||
al.sessions.Save(opts.SessionKey)
|
||||
|
||||
// 7. Optional: summarization
|
||||
// 13. Optional: summarization
|
||||
if opts.EnableSummary {
|
||||
al.maybeSummarize(opts.SessionKey)
|
||||
}
|
||||
|
||||
// 8. Optional: send response via bus
|
||||
// 14. Optional: send response via bus
|
||||
if opts.SendResponse {
|
||||
al.bus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: opts.Channel,
|
||||
|
|
@ -394,179 +537,165 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
})
|
||||
}
|
||||
|
||||
// 9. Log response
|
||||
// 15. Log response
|
||||
responsePreview := utils.Truncate(finalContent, 120)
|
||||
logger.InfoCF("agent", fmt.Sprintf("Response: %s", responsePreview),
|
||||
map[string]interface{}{
|
||||
"session_key": opts.SessionKey,
|
||||
"iterations": iteration,
|
||||
"steps": stepCount,
|
||||
"final_length": len(finalContent),
|
||||
})
|
||||
|
||||
return finalContent, nil
|
||||
}
|
||||
|
||||
// runLLMIteration executes the LLM call loop with tool handling.
|
||||
// Returns the final content, iteration count, and any error.
|
||||
func (al *AgentLoop) runLLMIteration(ctx context.Context, messages []providers.Message, opts processOptions) (string, int, error) {
|
||||
iteration := 0
|
||||
var finalContent string
|
||||
// runAgentLoopStreaming uses Fantasy's agent.Stream() to stream token deltas
|
||||
// to the bus in real time. Structure mirrors runAgentLoop but uses AgentStreamCall
|
||||
// with OnTextDelta, OnStepFinish, and OnToolCall callbacks.
|
||||
func (al *AgentLoop) runAgentLoopStreaming(ctx context.Context, opts processOptions) (string, error) {
|
||||
// 0. Record last channel
|
||||
if opts.Channel != "" && opts.ChatID != "" {
|
||||
if !constants.IsInternalChannel(opts.Channel) {
|
||||
channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID)
|
||||
if err := al.RecordLastChannel(channelKey); err != nil {
|
||||
logger.WarnCF("agent", "Failed to record last channel: %v", map[string]interface{}{"error": err.Error()})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for iteration < al.maxIterations {
|
||||
iteration++
|
||||
// 1. Update tool contexts
|
||||
al.updateToolContexts(opts.Channel, opts.ChatID)
|
||||
|
||||
logger.DebugCF("agent", "LLM iteration",
|
||||
// 2. Build messages
|
||||
var history []messages.Message
|
||||
var summary string
|
||||
if !opts.NoHistory {
|
||||
history = al.sessions.GetHistory(opts.SessionKey)
|
||||
summary = al.sessions.GetSummary(opts.SessionKey)
|
||||
}
|
||||
builtMsgs := al.contextBuilder.BuildMessages(history, summary, opts.UserMessage, nil, opts.Channel, opts.ChatID)
|
||||
|
||||
// 3. Save user message
|
||||
al.sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage)
|
||||
|
||||
// 4. Split into system/history/user
|
||||
systemPrompt := ""
|
||||
var historyMsgs []messages.Message
|
||||
userPrompt := opts.UserMessage
|
||||
|
||||
if len(builtMsgs) > 0 && builtMsgs[0].Role == "system" {
|
||||
systemPrompt = builtMsgs[0].Content
|
||||
if len(builtMsgs) > 2 {
|
||||
historyMsgs = builtMsgs[1 : len(builtMsgs)-1]
|
||||
}
|
||||
}
|
||||
|
||||
// 5. Convert history
|
||||
fantasyHistory := picofantasy.MessagesToFantasy(historyMsgs)
|
||||
|
||||
// 6. Build adapted tools (with optional offloading)
|
||||
streamAdaptCfg := picofantasy.AdaptedToolsConfig{
|
||||
MemStore: al.memoryStore,
|
||||
AgentID: "picoclaw",
|
||||
SessionKey: opts.SessionKey,
|
||||
}
|
||||
adaptedTools := picofantasy.BuildAdaptedTools(al.tools, al.bus, opts.Channel, opts.ChatID, streamAdaptCfg)
|
||||
|
||||
// 7. Create Fantasy agent
|
||||
agentOpts := []fantasy.AgentOption{
|
||||
fantasy.WithTools(adaptedTools...),
|
||||
fantasy.WithStopConditions(fantasy.StepCountIs(al.maxIterations)),
|
||||
}
|
||||
if systemPrompt != "" {
|
||||
agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt))
|
||||
}
|
||||
fantasyAgent := fantasy.NewAgent(al.languageModel, agentOpts...)
|
||||
|
||||
logger.DebugCF("agent", "Fantasy streaming agent created",
|
||||
map[string]interface{}{
|
||||
"iteration": iteration,
|
||||
"max": al.maxIterations,
|
||||
})
|
||||
|
||||
// Build tool definitions
|
||||
providerToolDefs := al.tools.ToProviderDefs()
|
||||
|
||||
// Log LLM request details
|
||||
logger.DebugCF("agent", "LLM request",
|
||||
map[string]interface{}{
|
||||
"iteration": iteration,
|
||||
"model": al.model,
|
||||
"messages_count": len(messages),
|
||||
"tools_count": len(providerToolDefs),
|
||||
"max_tokens": 8192,
|
||||
"temperature": 0.7,
|
||||
"system_prompt_len": len(messages[0].Content),
|
||||
"tools_count": len(adaptedTools),
|
||||
"history_count": len(historyMsgs),
|
||||
"max_iterations": al.maxIterations,
|
||||
"memory_enabled": al.memoryStore != nil,
|
||||
})
|
||||
|
||||
// Log full messages (detailed)
|
||||
logger.DebugCF("agent", "Full LLM request",
|
||||
map[string]interface{}{
|
||||
"iteration": iteration,
|
||||
"messages_json": formatMessagesForLog(messages),
|
||||
"tools_json": formatToolsForLog(providerToolDefs),
|
||||
})
|
||||
// 8. Build streaming call with callbacks
|
||||
streamCall := fantasy.AgentStreamCall{
|
||||
Prompt: userPrompt,
|
||||
Messages: fantasyHistory,
|
||||
|
||||
// Call LLM
|
||||
response, err := al.provider.Chat(ctx, messages, providerToolDefs, al.model, map[string]interface{}{
|
||||
"max_tokens": 8192,
|
||||
"temperature": 0.7,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
logger.ErrorCF("agent", "LLM call failed",
|
||||
map[string]interface{}{
|
||||
"iteration": iteration,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", iteration, fmt.Errorf("LLM call failed: %w", err)
|
||||
}
|
||||
|
||||
// Check if no tool calls - we're done
|
||||
if len(response.ToolCalls) == 0 {
|
||||
finalContent = response.Content
|
||||
logger.InfoCF("agent", "LLM response without tool calls (direct answer)",
|
||||
map[string]interface{}{
|
||||
"iteration": iteration,
|
||||
"content_chars": len(finalContent),
|
||||
})
|
||||
break
|
||||
}
|
||||
|
||||
// Log tool calls
|
||||
toolNames := make([]string, 0, len(response.ToolCalls))
|
||||
for _, tc := range response.ToolCalls {
|
||||
toolNames = append(toolNames, tc.Name)
|
||||
}
|
||||
logger.InfoCF("agent", "LLM requested tool calls",
|
||||
map[string]interface{}{
|
||||
"tools": toolNames,
|
||||
"count": len(response.ToolCalls),
|
||||
"iteration": iteration,
|
||||
})
|
||||
|
||||
// Build assistant message with tool calls
|
||||
assistantMsg := providers.Message{
|
||||
Role: "assistant",
|
||||
Content: response.Content,
|
||||
}
|
||||
for _, tc := range response.ToolCalls {
|
||||
argumentsJSON, _ := json.Marshal(tc.Arguments)
|
||||
assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, providers.ToolCall{
|
||||
ID: tc.ID,
|
||||
Type: "function",
|
||||
Function: &providers.FunctionCall{
|
||||
Name: tc.Name,
|
||||
Arguments: string(argumentsJSON),
|
||||
},
|
||||
})
|
||||
}
|
||||
messages = append(messages, assistantMsg)
|
||||
|
||||
// Save assistant message with tool calls to session
|
||||
al.sessions.AddFullMessage(opts.SessionKey, assistantMsg)
|
||||
|
||||
// Execute tool calls
|
||||
for _, tc := range response.ToolCalls {
|
||||
// Log tool call with arguments preview
|
||||
argsJSON, _ := json.Marshal(tc.Arguments)
|
||||
argsPreview := utils.Truncate(string(argsJSON), 200)
|
||||
logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview),
|
||||
map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"iteration": iteration,
|
||||
})
|
||||
|
||||
// Create async callback for tools that implement AsyncTool
|
||||
// NOTE: Following openclaw's design, async tools do NOT send results directly to users.
|
||||
// Instead, they notify the agent via PublishInbound, and the agent decides
|
||||
// whether to forward the result to the user (in processSystemMessage).
|
||||
asyncCallback := func(callbackCtx context.Context, result *tools.ToolResult) {
|
||||
// Log the async completion but don't send directly to user
|
||||
// The agent will handle user notification via processSystemMessage
|
||||
if !result.Silent && result.ForUser != "" {
|
||||
logger.InfoCF("agent", "Async tool completed, agent will handle notification",
|
||||
map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"content_len": len(result.ForUser),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
toolResult := al.tools.ExecuteWithContext(ctx, tc.Name, tc.Arguments, opts.Channel, opts.ChatID, asyncCallback)
|
||||
|
||||
// Send ForUser content to user immediately if not Silent
|
||||
if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse {
|
||||
// Stream text deltas to bus in real time
|
||||
OnTextDelta: func(id, text string) error {
|
||||
if opts.Channel != "" && opts.ChatID != "" {
|
||||
al.bus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: opts.Channel,
|
||||
ChatID: opts.ChatID,
|
||||
Content: toolResult.ForUser,
|
||||
Content: text,
|
||||
StreamDelta: true,
|
||||
})
|
||||
logger.DebugCF("agent", "Sent tool result to user",
|
||||
}
|
||||
return nil
|
||||
},
|
||||
|
||||
// Save each step's messages to session as they complete
|
||||
OnStepFinish: func(step fantasy.StepResult) error {
|
||||
stepMsgs := picofantasy.StepToMessages(step)
|
||||
for _, m := range stepMsgs {
|
||||
al.sessions.AddFullMessage(opts.SessionKey, m)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
|
||||
// Log tool calls as they happen
|
||||
OnToolCall: func(tc fantasy.ToolCallContent) error {
|
||||
logger.DebugCF("agent", "Streaming tool call",
|
||||
map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"content_len": len(toolResult.ForUser),
|
||||
"tool": tc.ToolName,
|
||||
"id": tc.ToolCallID,
|
||||
})
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// Determine content for LLM based on tool result
|
||||
contentForLLM := toolResult.ForLLM
|
||||
if contentForLLM == "" && toolResult.Err != nil {
|
||||
contentForLLM = toolResult.Err.Error()
|
||||
// 9. Call Fantasy agent.Stream()
|
||||
result, err := fantasyAgent.Stream(ctx, streamCall)
|
||||
if err != nil {
|
||||
logger.ErrorCF("agent", "Fantasy Stream failed",
|
||||
map[string]interface{}{"error": err.Error()})
|
||||
return "", fmt.Errorf("agent Stream failed: %w", err)
|
||||
}
|
||||
|
||||
toolResultMsg := providers.Message{
|
||||
Role: "tool",
|
||||
Content: contentForLLM,
|
||||
ToolCallID: tc.ID,
|
||||
}
|
||||
messages = append(messages, toolResultMsg)
|
||||
|
||||
// Save tool result message to session
|
||||
al.sessions.AddFullMessage(opts.SessionKey, toolResultMsg)
|
||||
}
|
||||
// 10. Extract final text
|
||||
finalContent := result.Response.Content.Text()
|
||||
if finalContent == "" {
|
||||
finalContent = opts.DefaultResponse
|
||||
}
|
||||
|
||||
return finalContent, iteration, nil
|
||||
// 11. Save session
|
||||
al.sessions.Save(opts.SessionKey)
|
||||
|
||||
// 12. Summarization
|
||||
if opts.EnableSummary {
|
||||
al.maybeSummarize(opts.SessionKey)
|
||||
}
|
||||
|
||||
// 13. Log response
|
||||
stepCount := len(result.Steps)
|
||||
responsePreview := utils.Truncate(finalContent, 120)
|
||||
logger.InfoCF("agent", fmt.Sprintf("Streaming response: %s", responsePreview),
|
||||
map[string]interface{}{
|
||||
"session_key": opts.SessionKey,
|
||||
"steps": stepCount,
|
||||
"final_length": len(finalContent),
|
||||
"total_tokens": result.TotalUsage.TotalTokens,
|
||||
})
|
||||
|
||||
return finalContent, nil
|
||||
}
|
||||
|
||||
// runLLMIteration — DELETED. Replaced by Fantasy's internal agent loop.
|
||||
|
||||
// updateToolContexts updates the context for tools that need channel/chatID info.
|
||||
func (al *AgentLoop) updateToolContexts(channel, chatID string) {
|
||||
// Use ContextualTool interface instead of type assertions
|
||||
|
|
@ -621,16 +750,16 @@ func (al *AgentLoop) GetStartupInfo() map[string]interface{} {
|
|||
}
|
||||
|
||||
// formatMessagesForLog formats messages for logging
|
||||
func formatMessagesForLog(messages []providers.Message) string {
|
||||
if len(messages) == 0 {
|
||||
func formatMessagesForLog(msgs []messages.Message) string {
|
||||
if len(msgs) == 0 {
|
||||
return "[]"
|
||||
}
|
||||
|
||||
var result string
|
||||
result += "[\n"
|
||||
for i, msg := range messages {
|
||||
for i, msg := range msgs {
|
||||
result += fmt.Sprintf(" [%d] Role: %s\n", i, msg.Role)
|
||||
if msg.ToolCalls != nil && len(msg.ToolCalls) > 0 {
|
||||
if len(msg.ToolCalls) > 0 {
|
||||
result += " ToolCalls:\n"
|
||||
for _, tc := range msg.ToolCalls {
|
||||
result += fmt.Sprintf(" - ID: %s, Type: %s, Name: %s\n", tc.ID, tc.Type, tc.Name)
|
||||
|
|
@ -652,25 +781,6 @@ func formatMessagesForLog(messages []providers.Message) string {
|
|||
return result
|
||||
}
|
||||
|
||||
// formatToolsForLog formats tool definitions for logging
|
||||
func formatToolsForLog(tools []providers.ToolDefinition) string {
|
||||
if len(tools) == 0 {
|
||||
return "[]"
|
||||
}
|
||||
|
||||
var result string
|
||||
result += "[\n"
|
||||
for i, tool := range tools {
|
||||
result += fmt.Sprintf(" [%d] Type: %s, Name: %s\n", i, tool.Type, tool.Function.Name)
|
||||
result += fmt.Sprintf(" Description: %s\n", tool.Function.Description)
|
||||
if len(tool.Function.Parameters) > 0 {
|
||||
result += fmt.Sprintf(" Parameters: %s\n", utils.Truncate(fmt.Sprintf("%v", tool.Function.Parameters), 200))
|
||||
}
|
||||
}
|
||||
result += "]"
|
||||
return result
|
||||
}
|
||||
|
||||
// summarizeSession summarizes the conversation history for a session.
|
||||
func (al *AgentLoop) summarizeSession(sessionKey string) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
|
||||
|
|
@ -689,14 +799,13 @@ func (al *AgentLoop) summarizeSession(sessionKey string) {
|
|||
// Oversized Message Guard
|
||||
// Skip messages larger than 50% of context window to prevent summarizer overflow
|
||||
maxMessageTokens := al.contextWindow / 2
|
||||
validMessages := make([]providers.Message, 0)
|
||||
validMessages := make([]messages.Message, 0)
|
||||
omitted := false
|
||||
|
||||
for _, m := range toSummarize {
|
||||
if m.Role != "user" && m.Role != "assistant" {
|
||||
continue
|
||||
}
|
||||
// Estimate tokens for this message
|
||||
msgTokens := len(m.Content) / 4
|
||||
if msgTokens > maxMessageTokens {
|
||||
omitted = true
|
||||
|
|
@ -710,7 +819,6 @@ func (al *AgentLoop) summarizeSession(sessionKey string) {
|
|||
}
|
||||
|
||||
// Multi-Part Summarization
|
||||
// Split into two parts if history is significant
|
||||
var finalSummary string
|
||||
if len(validMessages) > 10 {
|
||||
mid := len(validMessages) / 2
|
||||
|
|
@ -722,12 +830,9 @@ func (al *AgentLoop) summarizeSession(sessionKey string) {
|
|||
|
||||
// Merge them
|
||||
mergePrompt := fmt.Sprintf("Merge these two conversation summaries into one cohesive summary:\n\n1: %s\n\n2: %s", s1, s2)
|
||||
resp, err := al.provider.Chat(ctx, []providers.Message{{Role: "user", Content: mergePrompt}}, nil, al.model, map[string]interface{}{
|
||||
"max_tokens": 1024,
|
||||
"temperature": 0.3,
|
||||
})
|
||||
resp, err := al.callModel(ctx, mergePrompt)
|
||||
if err == nil {
|
||||
finalSummary = resp.Content
|
||||
finalSummary = resp
|
||||
} else {
|
||||
finalSummary = s1 + " " + s2
|
||||
}
|
||||
|
|
@ -743,11 +848,33 @@ func (al *AgentLoop) summarizeSession(sessionKey string) {
|
|||
al.sessions.SetSummary(sessionKey, finalSummary)
|
||||
al.sessions.TruncateHistory(sessionKey, 4)
|
||||
al.sessions.Save(sessionKey)
|
||||
al.summarizeFailures.Delete(sessionKey)
|
||||
} else {
|
||||
var count int
|
||||
if v, ok := al.summarizeFailures.Load(sessionKey); ok {
|
||||
count = v.(int)
|
||||
}
|
||||
count++
|
||||
al.summarizeFailures.Store(sessionKey, count)
|
||||
|
||||
const maxSummarizeFailures = 3
|
||||
const emergencyKeep = 10
|
||||
if count >= maxSummarizeFailures {
|
||||
logger.ErrorCF("agent", "Summarization failed repeatedly, force-truncating session",
|
||||
map[string]interface{}{
|
||||
"session": sessionKey,
|
||||
"consecutive_failures": count,
|
||||
"keep": emergencyKeep,
|
||||
})
|
||||
al.sessions.TruncateHistory(sessionKey, emergencyKeep)
|
||||
al.sessions.Save(sessionKey)
|
||||
al.summarizeFailures.Delete(sessionKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// summarizeBatch summarizes a batch of messages.
|
||||
func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []providers.Message, existingSummary string) (string, error) {
|
||||
// summarizeBatch summarizes a batch of messages using the Fantasy LanguageModel directly.
|
||||
func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []messages.Message, existingSummary string) (string, error) {
|
||||
prompt := "Provide a concise summary of this conversation segment, preserving core context and key points.\n"
|
||||
if existingSummary != "" {
|
||||
prompt += "Existing context: " + existingSummary + "\n"
|
||||
|
|
@ -757,20 +884,32 @@ func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []providers.Messa
|
|||
prompt += fmt.Sprintf("%s: %s\n", m.Role, m.Content)
|
||||
}
|
||||
|
||||
response, err := al.provider.Chat(ctx, []providers.Message{{Role: "user", Content: prompt}}, nil, al.model, map[string]interface{}{
|
||||
"max_tokens": 1024,
|
||||
"temperature": 0.3,
|
||||
return al.callModel(ctx, prompt)
|
||||
}
|
||||
|
||||
// callModel makes a direct call to the Fantasy LanguageModel (no tools, no agent loop).
|
||||
// Used for summarization and other simple generation tasks.
|
||||
func (al *AgentLoop) callModel(ctx context.Context, prompt string) (string, error) {
|
||||
temp := 0.3
|
||||
maxTokens := int64(1024)
|
||||
|
||||
resp, err := al.languageModel.Generate(ctx, fantasy.Call{
|
||||
Prompt: fantasy.Prompt{
|
||||
fantasy.NewUserMessage(prompt),
|
||||
},
|
||||
Temperature: &temp,
|
||||
MaxOutputTokens: &maxTokens,
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return response.Content, nil
|
||||
return resp.Content.Text(), nil
|
||||
}
|
||||
|
||||
// estimateTokens estimates the number of tokens in a message list.
|
||||
func (al *AgentLoop) estimateTokens(messages []providers.Message) int {
|
||||
func (al *AgentLoop) estimateTokens(msgs []messages.Message) int {
|
||||
total := 0
|
||||
for _, m := range messages {
|
||||
for _, m := range msgs {
|
||||
total += len(m.Content) / 4 // Simple heuristic: 4 chars per token
|
||||
}
|
||||
return total
|
||||
|
|
|
|||
|
|
@ -2,31 +2,55 @@ package agent
|
|||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
fantasy "charm.land/fantasy"
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
)
|
||||
|
||||
// mockProvider is a simple mock LLM provider for testing
|
||||
type mockProvider struct{}
|
||||
// mockLanguageModel is a simple mock fantasy.LanguageModel for testing
|
||||
type mockLanguageModel struct {
|
||||
response string
|
||||
}
|
||||
|
||||
func (m *mockProvider) Chat(ctx context.Context, messages []providers.Message, tools []providers.ToolDefinition, model string, opts map[string]interface{}) (*providers.LLMResponse, error) {
|
||||
return &providers.LLMResponse{
|
||||
Content: "Mock response",
|
||||
ToolCalls: []providers.ToolCall{},
|
||||
func newMockLanguageModel(response string) *mockLanguageModel {
|
||||
if response == "" {
|
||||
response = "Mock response"
|
||||
}
|
||||
return &mockLanguageModel{response: response}
|
||||
}
|
||||
|
||||
func (m *mockLanguageModel) Generate(_ context.Context, call fantasy.Call) (*fantasy.Response, error) {
|
||||
return &fantasy.Response{
|
||||
Content: fantasy.ResponseContent{fantasy.TextContent{Text: m.response}},
|
||||
FinishReason: fantasy.FinishReasonStop,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *mockProvider) GetDefaultModel() string {
|
||||
return "mock-model"
|
||||
func (m *mockLanguageModel) Stream(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return func(yield func(fantasy.StreamPart) bool) {
|
||||
yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, Delta: m.response})
|
||||
yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop})
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *mockLanguageModel) GenerateObject(_ context.Context, _ fantasy.ObjectCall) (*fantasy.ObjectResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
func (m *mockLanguageModel) StreamObject(_ context.Context, _ fantasy.ObjectCall) (fantasy.ObjectStreamResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
func (m *mockLanguageModel) Provider() string { return "mock" }
|
||||
func (m *mockLanguageModel) Model() string { return "mock-model" }
|
||||
|
||||
func TestRecordLastChannel(t *testing.T) {
|
||||
// Create temp workspace
|
||||
tmpDir, err := os.MkdirTemp("", "agent-test-*")
|
||||
|
|
@ -49,8 +73,8 @@ func TestRecordLastChannel(t *testing.T) {
|
|||
|
||||
// Create agent loop
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &mockProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Test RecordLastChannel
|
||||
testChannel := "test-channel"
|
||||
|
|
@ -66,7 +90,7 @@ func TestRecordLastChannel(t *testing.T) {
|
|||
}
|
||||
|
||||
// Verify persistence by creating a new agent loop
|
||||
al2 := NewAgentLoop(cfg, msgBus, provider)
|
||||
al2 := NewAgentLoop(cfg, msgBus, model)
|
||||
if al2.state.GetLastChannel() != testChannel {
|
||||
t.Errorf("Expected persistent channel '%s', got '%s'", testChannel, al2.state.GetLastChannel())
|
||||
}
|
||||
|
|
@ -94,8 +118,8 @@ func TestRecordLastChatID(t *testing.T) {
|
|||
|
||||
// Create agent loop
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &mockProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Test RecordLastChatID
|
||||
testChatID := "test-chat-id-123"
|
||||
|
|
@ -111,7 +135,7 @@ func TestRecordLastChatID(t *testing.T) {
|
|||
}
|
||||
|
||||
// Verify persistence by creating a new agent loop
|
||||
al2 := NewAgentLoop(cfg, msgBus, provider)
|
||||
al2 := NewAgentLoop(cfg, msgBus, model)
|
||||
if al2.state.GetLastChatID() != testChatID {
|
||||
t.Errorf("Expected persistent chat ID '%s', got '%s'", testChatID, al2.state.GetLastChatID())
|
||||
}
|
||||
|
|
@ -139,8 +163,8 @@ func TestNewAgentLoop_StateInitialized(t *testing.T) {
|
|||
|
||||
// Create agent loop
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &mockProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Verify state manager is initialized
|
||||
if al.state == nil {
|
||||
|
|
@ -174,8 +198,8 @@ func TestToolRegistry_ToolRegistration(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &mockProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Register a custom tool
|
||||
customTool := &mockCustomTool{}
|
||||
|
|
@ -220,8 +244,8 @@ func TestToolContext_Updates(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &simpleMockProvider{response: "OK"}
|
||||
_ = NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("OK")
|
||||
_ = NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Verify that ContextualTool interface is defined and can be implemented
|
||||
// This test validates the interface contract exists
|
||||
|
|
@ -251,8 +275,8 @@ func TestToolRegistry_GetDefinitions(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &mockProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Register a test tool and verify it shows up in startup info
|
||||
testTool := &mockCustomTool{}
|
||||
|
|
@ -295,8 +319,8 @@ func TestAgentLoop_GetStartupInfo(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &mockProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
info := al.GetStartupInfo()
|
||||
|
||||
|
|
@ -342,8 +366,8 @@ func TestAgentLoop_Stop(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &mockProvider{}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
|
||||
// Note: running is only set to true when Run() is called
|
||||
// We can't test that without starting the event loop
|
||||
|
|
@ -358,21 +382,6 @@ func TestAgentLoop_Stop(t *testing.T) {
|
|||
|
||||
// Mock implementations for testing
|
||||
|
||||
type simpleMockProvider struct {
|
||||
response string
|
||||
}
|
||||
|
||||
func (m *simpleMockProvider) Chat(ctx context.Context, messages []providers.Message, tools []providers.ToolDefinition, model string, opts map[string]interface{}) (*providers.LLMResponse, error) {
|
||||
return &providers.LLMResponse{
|
||||
Content: m.response,
|
||||
ToolCalls: []providers.ToolCall{},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *simpleMockProvider) GetDefaultModel() string {
|
||||
return "mock-model"
|
||||
}
|
||||
|
||||
// mockCustomTool is a simple mock tool for registration testing
|
||||
type mockCustomTool struct{}
|
||||
|
||||
|
|
@ -464,8 +473,8 @@ func TestToolResult_SilentToolDoesNotSendUserMessage(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &simpleMockProvider{response: "File operation complete"}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("File operation complete")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
helper := testHelper{al: al}
|
||||
|
||||
// ReadFileTool returns SilentResult, which should not send user message
|
||||
|
|
@ -506,8 +515,8 @@ func TestToolResult_UserFacingToolDoesSendMessage(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
provider := &simpleMockProvider{response: "Command output: hello world"}
|
||||
al := NewAgentLoop(cfg, msgBus, provider)
|
||||
model := newMockLanguageModel("Command output: hello world")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
helper := testHelper{al: al}
|
||||
|
||||
// ExecTool returns UserResult, which should send user message
|
||||
|
|
|
|||
106
pkg/agent/memgpt_tool.go
Normal file
106
pkg/agent/memgpt_tool.go
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
// PicoClaw - Ultra-lightweight personal AI agent
|
||||
// License: MIT
|
||||
//
|
||||
// Copyright (c) 2026 PicoClaw contributors
|
||||
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
memstore "github.com/sipeed/picoclaw/pkg/memory/store"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
)
|
||||
|
||||
// MemGPTTool wraps store.MemoryTool as a PicoClaw tools.Tool so it can be
|
||||
// registered in the ToolRegistry and executed by the Fantasy agent loop.
|
||||
type MemGPTTool struct {
|
||||
inner *memstore.MemoryTool
|
||||
}
|
||||
|
||||
var _ tools.Tool = (*MemGPTTool)(nil)
|
||||
|
||||
// NewMemGPTTool creates a PicoClaw tool wrapper around a MemoryTool.
|
||||
func NewMemGPTTool(store *memstore.MemoryStore, agentID, session string) *MemGPTTool {
|
||||
return &MemGPTTool{
|
||||
inner: memstore.NewMemoryTool(store, agentID, session),
|
||||
}
|
||||
}
|
||||
|
||||
func (t *MemGPTTool) Name() string { return "memory" }
|
||||
|
||||
func (t *MemGPTTool) Description() string {
|
||||
return "Manage the agent's 3-tier memory system. Actions: search (hybrid keyword+vector), read (by ID), write (to recall or archival), update, delete, status (context pressure). All memory persists across sessions."
|
||||
}
|
||||
|
||||
func (t *MemGPTTool) Parameters() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
"type": "object",
|
||||
"properties": map[string]interface{}{
|
||||
"action": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "The memory operation to perform.",
|
||||
"enum": []string{"search", "read", "write", "update", "delete", "status"},
|
||||
},
|
||||
"query": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Search query (for action=search).",
|
||||
},
|
||||
"id": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Memory ID (for action=read/update/delete).",
|
||||
},
|
||||
"content": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Content to store or update (for action=write/update).",
|
||||
},
|
||||
"source": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Source label for archival writes.",
|
||||
},
|
||||
"sector": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Memory sector: episodic, semantic, procedural, reflective.",
|
||||
"enum": []string{"episodic", "semantic", "procedural", "reflective"},
|
||||
},
|
||||
"tags": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Comma-separated tags for the memory entry.",
|
||||
},
|
||||
"tier": map[string]interface{}{
|
||||
"type": "string",
|
||||
"description": "Storage tier: recall (warm, default) or archival (cold, chunked+embedded).",
|
||||
"enum": []string{"recall", "archival"},
|
||||
},
|
||||
"limit": map[string]interface{}{
|
||||
"type": "integer",
|
||||
"description": "Max results for search. Default: 5.",
|
||||
},
|
||||
},
|
||||
"required": []string{"action"},
|
||||
}
|
||||
}
|
||||
|
||||
func (t *MemGPTTool) Execute(ctx context.Context, args map[string]interface{}) *tools.ToolResult {
|
||||
// Marshal the args back to JSON for the inner MemoryTool.Execute()
|
||||
input, err := json.Marshal(args)
|
||||
if err != nil {
|
||||
return tools.ErrorResult("invalid arguments: " + err.Error())
|
||||
}
|
||||
|
||||
result, err := t.inner.Execute(ctx, string(input))
|
||||
if err != nil {
|
||||
return tools.ErrorResult("memory tool error: " + err.Error())
|
||||
}
|
||||
|
||||
return &tools.ToolResult{
|
||||
ForLLM: result,
|
||||
}
|
||||
}
|
||||
|
||||
// UpdateSession rebinds the inner MemoryTool to a new session.
|
||||
// Called when the agent switches sessions.
|
||||
func (t *MemGPTTool) UpdateSession(store *memstore.MemoryStore, agentID, session string) {
|
||||
t.inner = memstore.NewMemoryTool(store, agentID, session)
|
||||
}
|
||||
|
|
@ -3,6 +3,8 @@ package bus
|
|||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
)
|
||||
|
||||
type MessageBus struct {
|
||||
|
|
@ -24,6 +26,22 @@ func (mb *MessageBus) PublishInbound(msg InboundMessage) {
|
|||
mb.inbound <- msg
|
||||
}
|
||||
|
||||
// TryPublishInbound attempts a non-blocking send. Returns false if the buffer
|
||||
// is full (message dropped). Callers should log and/or notify the user.
|
||||
func (mb *MessageBus) TryPublishInbound(msg InboundMessage) bool {
|
||||
select {
|
||||
case mb.inbound <- msg:
|
||||
return true
|
||||
default:
|
||||
logger.InfoCF("bus", "Inbound buffer full, message dropped",
|
||||
map[string]interface{}{
|
||||
"channel": msg.Channel,
|
||||
"sender_id": msg.SenderID,
|
||||
})
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (mb *MessageBus) ConsumeInbound(ctx context.Context) (InboundMessage, bool) {
|
||||
select {
|
||||
case msg := <-mb.inbound:
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ type OutboundMessage struct {
|
|||
Channel string `json:"channel"`
|
||||
ChatID string `json:"chat_id"`
|
||||
Content string `json:"content"`
|
||||
StreamDelta bool `json:"stream_delta,omitempty"` // true = partial token (not a complete message)
|
||||
}
|
||||
|
||||
type MessageHandler func(InboundMessage) error
|
||||
|
|
|
|||
|
|
@ -4,8 +4,10 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
)
|
||||
|
||||
type Channel interface {
|
||||
|
|
@ -20,19 +22,20 @@ type Channel interface {
|
|||
type BaseChannel struct {
|
||||
config interface{}
|
||||
bus *bus.MessageBus
|
||||
running bool
|
||||
running atomic.Bool
|
||||
name string
|
||||
allowList []string
|
||||
}
|
||||
|
||||
func NewBaseChannel(name string, config interface{}, bus *bus.MessageBus, allowList []string) *BaseChannel {
|
||||
return &BaseChannel{
|
||||
bc := &BaseChannel{
|
||||
config: config,
|
||||
bus: bus,
|
||||
name: name,
|
||||
allowList: allowList,
|
||||
running: false,
|
||||
}
|
||||
bc.running.Store(false)
|
||||
return bc
|
||||
}
|
||||
|
||||
func (c *BaseChannel) Name() string {
|
||||
|
|
@ -40,7 +43,7 @@ func (c *BaseChannel) Name() string {
|
|||
}
|
||||
|
||||
func (c *BaseChannel) IsRunning() bool {
|
||||
return c.running
|
||||
return c.running.Load()
|
||||
}
|
||||
|
||||
func (c *BaseChannel) IsAllowed(senderID string) bool {
|
||||
|
|
@ -100,9 +103,15 @@ func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []st
|
|||
Metadata: metadata,
|
||||
}
|
||||
|
||||
c.bus.PublishInbound(msg)
|
||||
if !c.bus.TryPublishInbound(msg) {
|
||||
logger.InfoCF(c.name, "Message dropped (bus full)",
|
||||
map[string]interface{}{
|
||||
"sender_id": msg.SenderID,
|
||||
"channel": msg.Channel,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (c *BaseChannel) setRunning(running bool) {
|
||||
c.running = running
|
||||
c.running.Store(running)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -72,7 +72,7 @@ func (c *MaixCamChannel) acceptConnections(ctx context.Context) {
|
|||
default:
|
||||
conn, err := c.listener.Accept()
|
||||
if err != nil {
|
||||
if c.running {
|
||||
if c.IsRunning() {
|
||||
logger.ErrorCF("maixcam", "Failed to accept connection", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
|
|
|
|||
|
|
@ -175,6 +175,7 @@ type ProviderConfig struct {
|
|||
APIBase string `json:"api_base" env:"PICOCLAW_PROVIDERS_{{.Name}}_API_BASE"`
|
||||
Proxy string `json:"proxy,omitempty" env:"PICOCLAW_PROVIDERS_{{.Name}}_PROXY"`
|
||||
AuthMethod string `json:"auth_method,omitempty" env:"PICOCLAW_PROVIDERS_{{.Name}}_AUTH_METHOD"`
|
||||
Timeout int `json:"timeout,omitempty" env:"PICOCLAW_PROVIDERS_{{.Name}}_TIMEOUT"` // seconds, 0 = default (120s)
|
||||
}
|
||||
|
||||
type GatewayConfig struct {
|
||||
|
|
@ -200,6 +201,7 @@ type WebToolsConfig struct {
|
|||
|
||||
type ToolsConfig struct {
|
||||
Web WebToolsConfig `json:"web"`
|
||||
ProgressiveDisclosure bool `json:"progressive_disclosure" env:"PICOCLAW_TOOLS_PROGRESSIVE_DISCLOSURE"`
|
||||
}
|
||||
|
||||
func DefaultConfig() *Config {
|
||||
|
|
@ -332,9 +334,48 @@ func LoadConfig(path string) (*Config, error) {
|
|||
return nil, err
|
||||
}
|
||||
|
||||
if warnings := cfg.Validate(); len(warnings) > 0 {
|
||||
for _, w := range warnings {
|
||||
fmt.Fprintf(os.Stderr, "config warning: %s\n", w)
|
||||
}
|
||||
}
|
||||
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
// Validate checks configuration for common issues. Returns a list of
|
||||
// warning strings. These are warnings rather than hard errors to maintain
|
||||
// backward compatibility, but they indicate values that will likely cause
|
||||
// problems at runtime.
|
||||
func (c *Config) Validate() []string {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
var warnings []string
|
||||
|
||||
if c.Agents.Defaults.Model == "" {
|
||||
warnings = append(warnings, "agents.defaults.model is empty: no default LLM model configured")
|
||||
}
|
||||
|
||||
if c.Agents.Defaults.MaxTokens <= 0 {
|
||||
warnings = append(warnings, fmt.Sprintf("agents.defaults.max_tokens=%d: should be > 0", c.Agents.Defaults.MaxTokens))
|
||||
}
|
||||
|
||||
if c.Agents.Defaults.MaxToolIterations <= 0 {
|
||||
warnings = append(warnings, fmt.Sprintf("agents.defaults.max_tool_iterations=%d: should be > 0", c.Agents.Defaults.MaxToolIterations))
|
||||
}
|
||||
|
||||
if c.Gateway.Port < 0 || c.Gateway.Port > 65535 {
|
||||
warnings = append(warnings, fmt.Sprintf("gateway.port=%d: must be in range 1-65535", c.Gateway.Port))
|
||||
}
|
||||
|
||||
if c.Heartbeat.Interval < 0 {
|
||||
warnings = append(warnings, fmt.Sprintf("heartbeat.interval=%d: must be >= 0", c.Heartbeat.Interval))
|
||||
}
|
||||
|
||||
return warnings
|
||||
}
|
||||
|
||||
func SaveConfig(path string, cfg *Config) error {
|
||||
cfg.mu.RLock()
|
||||
defer cfg.mu.RUnlock()
|
||||
|
|
|
|||
|
|
@ -146,14 +146,11 @@ func (hs *HeartbeatService) runLoop(stopChan chan struct{}) {
|
|||
func (hs *HeartbeatService) executeHeartbeat() {
|
||||
hs.mu.RLock()
|
||||
enabled := hs.enabled
|
||||
stopped := hs.stopChan == nil
|
||||
handler := hs.handler
|
||||
if !hs.enabled || hs.stopChan == nil {
|
||||
hs.mu.RUnlock()
|
||||
return
|
||||
}
|
||||
hs.mu.RUnlock()
|
||||
|
||||
if !enabled {
|
||||
if !enabled || stopped {
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue