feat: allow the agent to perform things concurrently (#74)

* feat: allow the agent to perform things concurrently

Signed-off-by: mudler <mudler@localai.io>

* Apply suggestions from code review

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* collect errors

Signed-off-by: mudler <mudler@localai.io>

---------

Signed-off-by: mudler <mudler@localai.io>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

core/agent/agent.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"sync"
@@ -1004,36 +1005,63 @@ func (a *Agent) Run() error {
 
 	// Expose a REST API to interact with the agent to ask it things
 
-	//todoTimer := time.NewTicker(a.options.periodicRuns)
 	timer := time.NewTimer(a.options.periodicRuns)
+
+	// we fire the periodicalRunner only once.
+	go a.periodicalRunRunner(timer)
+	var errs []error
+	var muErr sync.Mutex
+	var wg sync.WaitGroup
+
+	for i := 0; i <= a.options.parallelJobs; i++ {
+		xlog.Debug("Starting agent worker", "worker", i)
+		wg.Add(1)
+		go func() {
+			e := a.run(timer)
+			muErr.Lock()
+			errs = append(errs, e)
+			muErr.Unlock()
+			wg.Done()
+		}()
+	}
+
+	wg.Wait()
+
+	return errors.Join(errs...)
+}
+
+func (a *Agent) run(timer *time.Timer) error {
 	for {
 		xlog.Debug("Agent is now waiting for a new job", "agent", a.Character.Name)
 		select {
 		case job := <-a.jobQueue:
-			a.loop(timer, job)
+			if !timer.Stop() {
+				<-timer.C
+			}
+			xlog.Debug("Agent is consuming a job", "agent", a.Character.Name, "job", job)
+			a.consumeJob(job, UserRole)
+			timer.Reset(a.options.periodicRuns)
 		case <-a.context.Done():
 			// Agent has been canceled, return error
 			xlog.Warn("Agent has been canceled", "agent", a.Character.Name)
 			return ErrContextCanceled
+		}
+	}
+}
+
+func (a *Agent) periodicalRunRunner(timer *time.Timer) {
+	for {
+		select {
+		case <-a.context.Done():
+			// Agent has been canceled, return error
+			xlog.Warn("periodicalRunner has been canceled", "agent", a.Character.Name)
+			return
 		case <-timer.C:
 			a.periodicallyRun(timer)
 		}
 	}
 }
 
-func (a *Agent) loop(timer *time.Timer, job *types.Job) {
-	// Remember always to reset the timer - if we don't the agent will stop..
-	defer timer.Reset(a.options.periodicRuns)
-	// Consume the job and generate a response
-	// TODO: Give a short-term memory to the agent
-	// stop and drain the timer
-	if !timer.Stop() {
-		<-timer.C
-	}
-	xlog.Debug("Agent is consuming a job", "agent", a.Character.Name, "job", job)
-	a.consumeJob(job, UserRole)
-}
-
 func (a *Agent) Observer() Observer {
 	return a.observer
 }

core/agent/options.go

@@ -55,6 +55,7 @@ type options struct {
 	newConversationsSubscribers []func(openai.ChatCompletionMessage)
 
 	observer     Observer
+	parallelJobs int
 }
 
 func (o *options) SeparatedMultimodalModel() bool {
@@ -138,6 +139,13 @@ func EnableKnowledgeBaseWithResults(results int) Option {
 	}
 }
 
+func WithParallelJobs(jobs int) Option {
+	return func(o *options) error {
+		o.parallelJobs = jobs
+		return nil
+	}
+}
+
 func WithNewConversationSubscriber(sub func(openai.ChatCompletionMessage)) Option {
 	return func(o *options) error {
 		o.newConversationsSubscribers = append(o.newConversationsSubscribers, sub)

core/state/config.go

@@ -61,6 +61,7 @@ type AgentConfig struct {
 	SystemPrompt          string `json:"system_prompt" form:"system_prompt"`
 	LongTermMemory        bool   `json:"long_term_memory" form:"long_term_memory"`
 	SummaryLongTermMemory bool   `json:"summary_long_term_memory" form:"summary_long_term_memory"`
+	ParallelJobs          int    `json:"parallel_jobs" form:"parallel_jobs"`
 }
 
 type AgentConfigMeta struct {
@@ -260,6 +261,16 @@ func NewAgentConfigMeta(
 				Step:         1,
 				Tags:         config.Tags{Section: "AdvancedSettings"},
 			},
+			{
+				Name:         "parallel_jobs",
+				Label:        "Parallel Jobs",
+				Type:         "number",
+				DefaultValue: 5,
+				Min:          1,
+				Step:         1,
+				HelpText:     "Number of concurrent tasks that can run in parallel",
+				Tags:         config.Tags{Section: "AdvancedSettings"},
+			},
 		},
 		MCPServers: []config.Field{
 			{

core/state/pool.go

@@ -466,6 +466,10 @@ func (a *AgentPool) startAgentWithConfig(name string, config *AgentConfig) error
 		opts = append(opts, WithLoopDetectionSteps(config.LoopDetectionSteps))
 	}
 
+	if config.ParallelJobs > 0 {
+		opts = append(opts, WithParallelJobs(config.ParallelJobs))
+	}
+
 	xlog.Info("Starting agent", "name", name, "config", config)
 
 	agent, err := New(opts...)