添加agent并发控制

Author: SugarMGP

config/config.example.yaml

@@ -54,6 +54,7 @@ agent:
   think_interval: 15         # 决策间隔（秒）
   message_buffer_size: 15   # 消息缓冲区大小
   max_step: 12               # ReAct 最大步数
+  max_coroutine: 0           # 最大并发思考进程数（0表示不限制）
   enable_active_retrieval: true # 是否启用主动记忆检索
 
 # 聊天行为配置

internal/agent/concurrency.go

@@ -0,0 +1,92 @@
+package agent
+
+import (
+	"sync"
+
+	"go.uber.org/zap"
+)
+
+// ConcurrencyManager 并发管理器
+type ConcurrencyManager struct {
+	maxConcurrency int
+	currentRunning int
+	queue          []*ThinkTask
+	inQueue        map[int64]bool // 快速去重（群组ID -> 是否在队列中）
+	mu             sync.Mutex
+
+	handler func(groupID int64, isMention bool) // 执行函数
+}
+
+// ThinkTask 思考任务
+type ThinkTask struct {
+	GroupID   int64
+	IsMention bool
+}
+
+// NewConcurrencyManager 创建并发管理器
+func NewConcurrencyManager(max int, h func(groupID int64, isMention bool)) *ConcurrencyManager {
+	return &ConcurrencyManager{
+		maxConcurrency: max,
+		currentRunning: 0,
+		inQueue:        make(map[int64]bool),
+		handler:        h,
+	}
+}
+
+// Submit 提交任务
+func (m *ConcurrencyManager) Submit(groupID int64, isMention bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.inQueue[groupID] {
+		zap.L().Debug("任务已在队列中，跳过", zap.Int64("group_id", groupID))
+		return
+	}
+
+	// 如果设置了最大并发数，且当前运行数已满，则入队
+	if m.maxConcurrency > 0 && m.currentRunning >= m.maxConcurrency {
+		m.queue = append(m.queue, &ThinkTask{
+			GroupID:   groupID,
+			IsMention: isMention,
+		})
+		m.inQueue[groupID] = true
+		zap.L().Debug("并发已满，任务进入队列",
+			zap.Int64("group_id", groupID),
+			zap.Int("current", m.currentRunning),
+			zap.Int("queue_len", len(m.queue)))
+		return
+	}
+
+	m.currentRunning++
+	go m.execute(groupID, isMention)
+}
+
+// execute 执行任务
+func (m *ConcurrencyManager) execute(groupID int64, isMention bool) {
+	defer m.Finish()
+	m.handler(groupID, isMention)
+}
+
+// Finish 任务完成回调
+func (m *ConcurrencyManager) Finish() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.currentRunning--
+	if m.currentRunning < 0 {
+		m.currentRunning = 0
+	}
+
+	// 调度下一个任务
+	if len(m.queue) > 0 {
+		// 取出队首任务
+		task := m.queue[0]
+		m.queue = m.queue[1:]
+		delete(m.inQueue, task.GroupID)
+
+		// 立即启动
+		m.currentRunning++
+		go m.execute(task.GroupID, task.IsMention)
+		zap.L().Debug("从队列调度任务执行", zap.Int64("group_id", task.GroupID))
+	}
+}

internal/agent/react_agent.go

@@ -31,15 +31,16 @@ import (
 
 // Agent 沐沐智能体
 type Agent struct {
-	cfg     *config.Config
-	persona *persona.Persona
-	memory  *memory.Manager
-	model   model.ToolCallingChatModel
-	vision  *llm.VisionClient // 多模态视觉模型
-	bot     *onebot.Client
-	react   *react.Agent
-	tools   []tool.BaseTool
-	mcpMgr  *mcp.Manager // MCP 管理器
+	cfg            *config.Config
+	persona        *persona.Persona
+	memory         *memory.Manager
+	model          model.ToolCallingChatModel
+	vision         *llm.VisionClient // 多模态视觉模型
+	bot            *onebot.Client
+	react          *react.Agent
+	tools          []tool.BaseTool
+	mcpMgr         *mcp.Manager        // MCP 管理器
+	concurrencyMgr *ConcurrencyManager // 并发管理器
 
 	jargonMgr *jargon.Manager   // 黑话管理器
 	learner   *learning.Learner // 后台学习系统
@@ -79,6 +80,9 @@ func New(
 		stopCh:            make(chan struct{}),
 	}
 
+	// 初始化并发管理器
+	a.concurrencyMgr = NewConcurrencyManager(cfg.Agent.MaxCoroutine, a.think)
+
 	// 初始化黑话管理器
 	a.jargonMgr = jargon.New(mem)
 
@@ -248,7 +252,7 @@ func (a *Agent) onMessage(msg *onebot.GroupMessage) {
 
 	// 如果被 @ 了，立即触发一次思考（跳过等待）
 	if isMentioned {
-		go a.think(msg.GroupID, true)
+		go a.concurrencyMgr.Submit(msg.GroupID, true)
 	}
 }
 
@@ -437,7 +441,7 @@ func (a *Agent) thinkCycle() {
 		if rand.Float64() > speakProb {
 			continue
 		}
-		a.think(gc.GroupID, false)
+		a.concurrencyMgr.Submit(gc.GroupID, false)
 	}
 }
 
@@ -483,7 +487,7 @@ func (a *Agent) getSpeakProbability(groupID int64) float64 {
 	return baseProb
 }
 
-// think 进行思考和决策
+// think 提交思考任务
 func (a *Agent) think(groupID int64, isMention bool) {
 	if a.bot.IsSelfMuted(groupID) {
 		return

internal/config/config.go

@@ -67,6 +67,7 @@ type AgentConfig struct {
 	ThinkInterval         int  `yaml:"think_interval"`          // 决策间隔（秒）
 	MessageBufferSize     int  `yaml:"message_buffer_size"`     // 消息缓冲区大小
 	MaxStep               int  `yaml:"max_step"`                // ReAct 最大步数
+	MaxCoroutine          int  `yaml:"max_coroutine"`           // 最大并发思考进程数（0表示不限制）
 	EnableActiveRetrieval bool `yaml:"enable_active_retrieval"` // 是否启用主动记忆检索（阈值固定0.7）
 }