Method processMessages

pkg/agent/processor.go:21–91

processMessages 处理消息队列

(ctx context.Context)

// processMessages 处理消息队列
func (a *Agent) processMessages(ctx context.Context) {
	procLog.Info(ctx, "processMessages started", map[string]any{"agent_id": a.id})

	a.mu.Lock()
	if a.state != types.AgentStateReady {
		procLog.Warn(ctx, "agent not ready, skipping", map[string]any{"agent_id": a.id, "state": a.state})
		a.mu.Unlock()
		return // 已经在处理中
	}
	a.state = types.AgentStateWorking
	a.iterationCount = 0          // 重置迭代计数
	a.initialThinkingSent = false // 重置初始思考事件标志，允许新用户消息触发新的"任务规划"
	initialMsgCount := len(a.messages)
	procLog.Info(ctx, "agent state changed to working", map[string]any{"agent_id": a.id, "message_count": initialMsgCount})
	a.mu.Unlock()

	defer func() {
		a.mu.Lock()
		a.state = types.AgentStateReady
		// 检查是否有新的用户消息需要处理
		// 只有当最后一条消息是用户消息时才需要重新处理
		// （避免 assistant 响应触发无限循环）
		hasNewUserMessage := false
		if len(a.messages) > initialMsgCount {
			lastMsg := a.messages[len(a.messages)-1]
			hasNewUserMessage = lastMsg.Role == types.MessageRoleUser
		}
		a.mu.Unlock()

		// 如果有新的用户消息，重新触发处理
		// 注意：使用新的 context，而不是可能已取消的旧 context
		// 这样即使用户点击了"停止"，新消息仍然可以被处理
		if hasNewUserMessage {
			newCtx := context.Background()
			go a.processMessages(newCtx)
		}
	}()

	// 发送状态变更事件
	a.eventBus.EmitMonitor(&types.MonitorStateChangedEvent{
		State: types.AgentStateWorking,
	})

	// 设置断点
	a.setBreakpoint(types.BreakpointPreModel)

	procLog.Info(ctx, "calling runModelStep", map[string]any{"agent_id": a.id})

	// 调用模型
	if err := a.runModelStep(ctx); err != nil {
		procLog.Error(ctx, "runModelStep failed", map[string]any{"agent_id": a.id, "error": err.Error()})
		a.eventBus.EmitMonitor(&types.MonitorErrorEvent{
			Severity: "error",
			Phase:    "model",
			Message:  err.Error(),
		})
	}