Update README.md

Add observability screenshot and bullet point. Also update strap line and descriptions at the top to try and describe the benefit of this software

README.md

@@ -2,7 +2,7 @@
   <img src="./webui/react-ui/public/logo_1.png" alt="LocalAGI Logo" width="220"/>
 </p>
 
-<h3 align="center"><em>Your AI. Your Hardware. Your Rules.</em></h3>
+<h3 align="center"><em>Your AI. Your Hardware. Your Rules</em></h3>
 
 <div align="center">
   
@@ -13,9 +13,9 @@
 
 </div>
 
-We empower you building AI Agents that you can run locally, without coding.
+Create customizable AI assistants, automations, chat bots and agents that run 100% locally. No need for agentic Python libraries or cloud service keys, just bring your GPU (or even just CPU) and a web browser.
 
-**LocalAGI** is a powerful, self-hostable AI Agent platform designed for maximum privacy and flexibility. A complete drop-in replacement for OpenAI's Responses APIs with advanced agentic capabilities. No clouds. No data leaks. Just pure local AI that works on consumer-grade hardware (CPU and GPU).
+**LocalAGI** is a powerful, self-hostable AI Agent platform that allows you to design AI automations without writing code. A complete drop-in replacement for OpenAI's Responses APIs with advanced agentic capabilities. No clouds. No data leaks. Just pure local AI that works on consumer-grade hardware (CPU and GPU).
 
 ## 🛡️ Take Back Your Privacy
 
@@ -37,6 +37,7 @@ LocalAGI ensures your data stays exactly where you want it—on your hardware. N
 - 🖼 **Multimodal Support**: Ready for vision, text, and more.
 - 🔧 **Extensible Custom Actions**: Easily script dynamic agent behaviors in Go (interpreted, no compilation!).
 - 🛠 **Fully Customizable Models**: Use your own models or integrate seamlessly with [LocalAI](https://github.com/mudler/LocalAI).
+- 📊 **Observability**: Monitor agent status and view detailed observable updates in real-time.
 
 ## 🛠️ Quickstart
 
@@ -194,6 +195,8 @@ LocalAGI is part of the powerful Local family of privacy-focused AI tools:
 ![Web UI Dashboard](https://github.com/user-attachments/assets/a40194f9-af3a-461f-8b39-5f4612fbf221)
 ![Web UI Agent Settings](https://github.com/user-attachments/assets/fb3c3e2a-cd53-4ca8-97aa-c5da51ff1f83)
 ![Web UI Create Group](https://github.com/user-attachments/assets/102189a2-0fba-4a1e-b0cb-f99268ef8062)
+![Web UI Agent Observability](https://github.com/user-attachments/assets/f7359048-9d28-4cf1-9151-1f5556ce9235)
+
 
 ### Connectors Ready-to-Go
 

core/agent/agent.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"sync"
@@ -989,7 +990,6 @@ func (a *Agent) periodicallyRun(timer *time.Timer) {
 }
 
 func (a *Agent) Run() error {
-
 	a.startNewConversationsConsumer()
 	xlog.Debug("Agent is now running", "agent", a.Character.Name)
 	// The agent run does two things:
@@ -1004,36 +1004,68 @@ 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
+
+	parallelJobs := a.options.parallelJobs
+	if a.options.parallelJobs == 0 {
+		parallelJobs = 1
+	}
+
+	for i := 0; i < 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/observer.go

@@ -36,7 +36,8 @@ func NewSSEObserver(agent string, manager sse.Manager) *SSEObserver {
 }
 
 func (s *SSEObserver) NewObservable() *types.Observable {
-	id := atomic.AddInt32(&s.maxID, 1)
+	id  := atomic.AddInt32(&s.maxID, 1)
+	
 	return &types.Observable{
 		ID:    id - 1,
 		Agent: s.agent,

core/agent/options.go

@@ -54,7 +54,8 @@ type options struct {
 
 	newConversationsSubscribers []func(openai.ChatCompletionMessage)
 
-	observer Observer
+	observer     Observer
+	parallelJobs int
 }
 
 func (o *options) SeparatedMultimodalModel() bool {
@@ -63,6 +64,7 @@ func (o *options) SeparatedMultimodalModel() bool {
 
 func defaultOptions() *options {
 	return &options{
+		parallelJobs: 1,
 		periodicRuns: 15 * time.Minute,
 		LLMAPI: llmOptions{
 			APIURL: "http://localhost:8080",
@@ -138,6 +140,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/agent/state_test.go

@@ -25,6 +25,7 @@ var _ = Describe("Agent test", func() {
 			agent, err = New(
 				WithLLMAPIURL(apiURL),
 				WithModel(testModel),
+				WithTimeout("10m"),
 				WithRandomIdentity(),
 			)
 			Expect(err).ToNot(HaveOccurred())

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

@@ -166,7 +166,56 @@ func (a *AgentPool) CreateAgent(name string, agentConfig *AgentConfig) error {
 		}
 	}(a.pool[name])
 
-	return a.startAgentWithConfig(name, agentConfig)
+	return a.startAgentWithConfig(name, agentConfig, nil)
+}
+
+func (a *AgentPool) RecreateAgent(name string, agentConfig *AgentConfig) error {
+	a.Lock()
+	defer a.Unlock()
+
+	oldAgent := a.agents[name]
+	var o *types.Observable
+	obs := oldAgent.Observer()
+	if obs != nil {
+		o = obs.NewObservable()
+		o.Name = "Restarting Agent"
+		o.Icon = "sync"
+		o.Creation = &types.Creation{}
+		obs.Update(*o)
+	}
+
+	stateFile, characterFile := a.stateFiles(name)
+
+	os.Remove(stateFile)
+	os.Remove(characterFile)
+
+	oldAgent.Stop()
+
+	a.pool[name] = *agentConfig
+	delete(a.agents, name)
+
+	if err := a.save(); err != nil {
+		if obs != nil {
+			o.Completion = &types.Completion{Error: err.Error()}
+			obs.Update(*o)
+		}
+		return err
+	}
+
+	if err := a.startAgentWithConfig(name, agentConfig, obs); err != nil {
+		if obs != nil {
+			o.Completion = &types.Completion{Error: err.Error()}
+			obs.Update(*o)
+		}
+		return err
+	}
+
+	if obs != nil {
+		o.Completion = &types.Completion{}
+		obs.Update(*o)
+	}
+
+	return nil
 }
 
 func createAgentAvatar(APIURL, APIKey, model, imageModel, avatarDir string, agent AgentConfig) error {
@@ -268,8 +317,13 @@ func (a *AgentPool) GetStatusHistory(name string) *Status {
 	return a.agentStatus[name]
 }
 
-func (a *AgentPool) startAgentWithConfig(name string, config *AgentConfig) error {
+func (a *AgentPool) startAgentWithConfig(name string, config *AgentConfig, obs Observer) error {
-	manager := sse.NewManager(5)
+	var manager sse.Manager
+	if m, ok := a.managers[name]; ok {
+		manager = m
+	} else {
+		manager = sse.NewManager(5)
+	}
 	ctx := context.Background()
 	model := a.defaultModel
 	multimodalModel := a.defaultMultimodalModel
@@ -331,6 +385,10 @@ func (a *AgentPool) startAgentWithConfig(name string, config *AgentConfig) error
 	// 	dynamicPrompts = append(dynamicPrompts, p.ToMap())
 	// }
 
+	if obs == nil {
+		obs = NewSSEObserver(name, manager)
+	}
+
 	opts := []Option{
 		WithModel(model),
 		WithLLMAPIURL(a.apiURL),
@@ -407,7 +465,7 @@ func (a *AgentPool) startAgentWithConfig(name string, config *AgentConfig) error
 				c.AgentResultCallback()(state)
 			}
 		}),
-		WithObserver(NewSSEObserver(name, manager)),
+		WithObserver(obs),
 	}
 
 	if config.HUD {
@@ -466,6 +524,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...)
@@ -510,7 +572,7 @@ func (a *AgentPool) StartAll() error {
 		if a.agents[name] != nil { // Agent already started
 			continue
 		}
-		if err := a.startAgentWithConfig(name, &config); err != nil {
+		if err := a.startAgentWithConfig(name, &config, nil); err != nil {
 			xlog.Error("Failed to start agent", "name", name, "error", err)
 		}
 	}
@@ -548,7 +610,7 @@ func (a *AgentPool) Start(name string) error {
 		return nil
 	}
 	if config, ok := a.pool[name]; ok {
-		return a.startAgentWithConfig(name, &config)
+		return a.startAgentWithConfig(name, &config, nil)
 	}
 
 	return fmt.Errorf("agent %s not found", name)

services/connectors/irc.go

@@ -77,8 +77,9 @@ func (i *IRC) Start(a *agent.Agent) {
 	}
 	i.conn.UseTLS = false
 	i.conn.AddCallback("001", func(e *irc.Event) {
-		xlog.Info("Connected to IRC server", "server", i.server)
+		xlog.Info("Connected to IRC server", "server", i.server, "arguments", e.Arguments)
 		i.conn.Join(i.channel)
+		i.nickname = e.Arguments[0]
 		xlog.Info("Joined channel", "channel", i.channel)
 	})
 
@@ -207,6 +208,13 @@ func (i *IRC) Start(a *agent.Agent) {
 
 	// Start the IRC client in a goroutine
 	go i.conn.Loop()
+	go func() {
+		select  {
+		case <-a.Context().Done():
+			i.conn.Quit()
+			return
+		}
+	}()
 }
 
 // IRCConfigMeta returns the metadata for IRC connector configuration fields

webui/app.go

@@ -176,17 +176,7 @@ func (a *App) UpdateAgentConfig(pool *state.AgentPool) func(c *fiber.Ctx) error
 			return errorJSONMessage(c, err.Error())
 		}
 
-		// Remove the agent first
+		if err := pool.RecreateAgent(agentName, &newConfig); err != nil {
-		if err := pool.Remove(agentName); err != nil {
-			return errorJSONMessage(c, "Error removing agent: "+err.Error())
-		}
-
-		// Create agent with new config
-		if err := pool.CreateAgent(agentName, &newConfig); err != nil {
-			// Try to restore the old configuration if update fails
-			if restoreErr := pool.CreateAgent(agentName, oldConfig); restoreErr != nil {
-				return errorJSONMessage(c, fmt.Sprintf("Failed to update agent and restore failed: %v, %v", err, restoreErr))
-			}
 			return errorJSONMessage(c, "Error updating agent: "+err.Error())
 		}
 

webui/react-ui/src/pages/AgentStatus.jsx

@@ -99,8 +99,17 @@ function AgentStatus() {
           creation: data.creation,
           progress: data.progress,
           completion: data.completion,
-          // children are always built client-side
         };
+        // Events can be received out of order
+        if (data.creation)
+          updated.creation = data.creation;
+        if (data.completion)
+          updated.completion = data.completion;
+        if (data.parent_id && !prevMap[data.parent_id])
+          prevMap[data.parent_id] = {
+            id: data.parent_id,
+            name: "unknown",
+          };
         const newMap = { ...prevMap, [data.id]: updated };
         setObservableTree(buildObservableTree(newMap));
         return newMap;