feat: 群聊并发 — 主会话繁忙时创建协会议话

- Orchestrator 新增 per-session 处理锁 + co-session 跟踪
- 主会话处理中时,新消息自动创建协会议话(并发 LLM 调用)
- 协会议话上限 3 个,超出排队等待 500ms
- 主会话结束后自动释放锁

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2026-06-22 21:07:16 +08:00
parent 489657ec08
commit 6faef5b6e5
@@ -43,6 +43,12 @@ type Orchestrator struct {
ocrProvider llm.LLMProvider // OCR 模型 (文字提取,与视觉模型并行调用) ocrProvider llm.LLMProvider // OCR 模型 (文字提取,与视觉模型并行调用)
videoProvider llm.LLMProvider // 视频模型 (短视频理解) videoProvider llm.LLMProvider // 视频模型 (短视频理解)
asrProvider llm.ASRProvider // ASR 语音识别 (语音消息转录) asrProvider llm.ASRProvider // ASR 语音识别 (语音消息转录)
// 群聊并发:主会话繁忙时创建协会议话
sessionProcMu sync.Mutex
sessionProc map[string]bool // sessionID → currently processing
activeCoSessions map[string]int // sessionID → active co-session count
maxCoSessions int // max concurrent co-sessions per main session (default 3)
} }
// SetResponseCache sets the response cache (optional, for Phase 0.2). // SetResponseCache sets the response cache (optional, for Phase 0.2).
@@ -124,6 +130,9 @@ func NewOrchestrator(
synthesizer: NewSynthesizer(chatAdapter, nil), synthesizer: NewSynthesizer(chatAdapter, nil),
memoryRetriever: memoryRetriever, memoryRetriever: memoryRetriever,
memoryExtractor: memoryExtractor, memoryExtractor: memoryExtractor,
sessionProc: make(map[string]bool),
activeCoSessions: make(map[string]int),
maxCoSessions: 3,
} }
} }
@@ -172,7 +181,42 @@ func (o *Orchestrator) ProcessInput(
} }
}() }()
// 0. 发布合成开始事件 // 0. 群聊并发:检测主会话是否繁忙,决定是主会话还是协会议话
isCoSession := false
o.sessionProcMu.Lock()
if o.sessionProc[params.SessionID] {
if o.activeCoSessions[params.SessionID] >= o.maxCoSessions {
o.sessionProcMu.Unlock()
logger.Printf("[orchestrator] 协会议话已达上限,排队等待")
time.Sleep(500 * time.Millisecond)
o.sessionProcMu.Lock()
}
o.activeCoSessions[params.SessionID]++
isCoSession = true
o.sessionProcMu.Unlock()
logger.Printf("[orchestrator] 主会话繁忙,创建协会议话 (session=%s, active=%d)", params.SessionID, o.activeCoSessions[params.SessionID])
} else {
o.sessionProc[params.SessionID] = true
o.sessionProcMu.Unlock()
defer func() {
o.sessionProcMu.Lock()
delete(o.sessionProc, params.SessionID)
o.sessionProcMu.Unlock()
}()
}
if isCoSession {
defer func() {
o.sessionProcMu.Lock()
o.activeCoSessions[params.SessionID]--
if o.activeCoSessions[params.SessionID] <= 0 {
delete(o.activeCoSessions, params.SessionID)
}
o.sessionProcMu.Unlock()
}()
}
// 0.5 发布合成开始事件
o.getBus().Publish(bus.BusEvent{ o.getBus().Publish(bus.BusEvent{
Type: bus.EventSynthesisStarted, Type: bus.EventSynthesisStarted,
SessionID: params.SessionID, SessionID: params.SessionID,