重构: F-09 batch5 llm_concurrency会话级改造(决策c)
- state.rs: per_conv HashMap<conv_id,Semaphore>(每对话permits=2) + per_conv_permits热改 + acquire_per_conv(conv_id) + release_conv - agentic.rs: loop入口acquire_global+acquire_per_conv(conv_id)整loop持有(会话级并发上限3) + 删每轮acquire/drop - compress/title/knowledge_inject: acquire_per_conv(conv_id)+签名conv_id - project.rs: 扫描合成key + ai_conversation_delete release_conv global: LLM调用并发→并发会话数上限;F-260616-12 retry permit自洽 主代兜底: cargo check --workspace 0 + test 96 + grep印证
This commit is contained in:
@@ -81,20 +81,28 @@ impl Default for KnowledgeConfig {
|
||||
// LLM 调用并发控制（双层 Semaphore + 可选 per-provider 层）
|
||||
// ============================================================
|
||||
|
||||
/// LLM 调用并发控制 — 全局 + 单对话双层 Semaphore + 可选 per-provider 层
|
||||
/// LLM 调用并发控制 — 会话级 global + 单对话内 per_conv 双层 Semaphore + 可选 per-provider 层
|
||||
///
|
||||
/// 限流对象：所有真实 LLM 调用（主循环 stream_llm / 标题生成 / 知识提炼）。
|
||||
/// 不限流本地工具执行（tools.execute）——本地操作无外部成本、不受 RPM 约束。
|
||||
///
|
||||
/// ## F-260616-09 B 批5 / 决策 c-1：global 改「并发会话数上限」
|
||||
/// `global` permits 默认 3（`new(3, 2)`），语义从「全局 LLM 调用并发上限」改为
|
||||
/// 「并发会话数上限」：`run_agentic_loop` 入口 `acquire_global()` 拿 1 permit，持有整个 loop
|
||||
/// 生命周期（含工具执行/审批等待/重试）。同一时刻最多 3 个对话并发跑 loop，第 4 个排队。
|
||||
/// 收益：token 暴增护栏（决策 c 原意）。
|
||||
///
|
||||
/// ## per_conv 改 HashMap<conv_id, Semaphore>
|
||||
/// 原应用级单信号量（AiSession 单例 + generating 互斥下退化）改为
|
||||
/// `HashMap<String, Arc<Semaphore>>`：每对话一份（permits=2，主循环 + 标题 + 提炼各自限流）。
|
||||
/// `acquire_per_conv(conv_id)`：lock HashMap → 无则建（permits=2）→ clone Arc → 释放 lock → acquire_owned。
|
||||
/// conv 退出清理（`release_conv(conv_id)`）：loop 结束 + 无 pending 审批时 remove 条目（防 HashMap 无限增长）。
|
||||
///
|
||||
/// 运行时调整：tokio Semaphore 的 permits 数构造时固定、不可增减，
|
||||
/// 故用 `Arc<Mutex<Arc<Semaphore>>>` 双层包装——替换内层 Arc 即重建 Semaphore。
|
||||
/// 故 `global` 用 `Arc<Mutex<Arc<Semaphore>>>` 双层包装——替换内层 Arc 即重建 Semaphore。
|
||||
/// 已持有旧 permit 的任务不受影响（permit 绑定旧 Semaphore，软收敛），
|
||||
/// 新请求 lock 后克隆到最新 Arc、自动走新限制。旧 Semaphore 随最后 permit 释放而 drop。
|
||||
///
|
||||
/// 注意：per_conv 当前是应用级单一信号量(非 per-conv map)。因 AiSession 为单例 +
|
||||
/// generating 互斥,同一时刻仅一个对话的 loop 在跑,per_conv 退化为"单对话内并发"
|
||||
/// (主循环 stream_llm + 标题生成 + 知识提炼)。未来若支持多对话并发,
|
||||
/// 需改为 HashMap<conv_id, Semaphore>。
|
||||
/// per_conv HashMap 内每条 Arc<Semaphore> 不需运行时改 permits（对话内并发上限 2 固定），故无重建需求。
|
||||
///
|
||||
/// ## F-260614-04: per-provider 层(可选)
|
||||
/// `per_provider` 为 HashMap<provider_id, Arc<Semaphore>>。调用方(agentic loop)经
|
||||
@@ -105,8 +113,16 @@ impl Default for KnowledgeConfig {
|
||||
/// (set_provider_caps 传 min(sum, global_cap)),非运行时强约束。
|
||||
#[derive(Clone)]
|
||||
pub struct LlmConcurrency {
|
||||
/// 会话级并发上限(F--09 B 批5/决策 c-1)：permits=默认 3，run_agentic_loop 入口拿 1 持整 loop。
|
||||
global: Arc<Mutex<Arc<Semaphore>>>,
|
||||
per_conv: Arc<Mutex<Arc<Semaphore>>>,
|
||||
/// 单对话内并发上限(F-09 B 批5)：HashMap<conv_id, Semaphore>，每对话 permits=2。
|
||||
/// acquire 时按 conv_id 取/建；release_conv 在 loop 结束 + 无 pending 时 remove。
|
||||
per_conv: Arc<Mutex<HashMap<String, Arc<Semaphore>>>>,
|
||||
/// 每对话内并发上限(permits 默认 2，构造时传入 new(3, 2) 的第二参)。
|
||||
/// F-09 B 批5:AtomicUsize 支持运行时热改(ai_set_concurrency_config 的 per_conv_limit)。
|
||||
/// 热改后**已建对话**的旧 Semaphore 不变(permits 构造时固定)，**新建对话**用新值；
|
||||
/// 为使热改立即全量生效，set_per_conv 同时清空 HashMap 强制重建(软收敛：旧 permit 随 Drop 释放)。
|
||||
per_conv_permits: Arc<AtomicUsize>,
|
||||
/// F-260614-04: per-provider 信号量表。空 = 无 per-provider 限流(单 provider 路径零变化)。
|
||||
/// Arc<Mutex<HashMap>>:运行时增删 provider 配置时替换/插入,acquire 时 clone Arc。
|
||||
per_provider: Arc<Mutex<HashMap<String, Arc<Semaphore>>>>,
|
||||
@@ -116,31 +132,53 @@ impl LlmConcurrency {
|
||||
pub fn new(global: usize, per_conv: usize) -> Self {
|
||||
Self {
|
||||
global: Arc::new(Mutex::new(Arc::new(Semaphore::new(global)))),
|
||||
per_conv: Arc::new(Mutex::new(Arc::new(Semaphore::new(per_conv)))),
|
||||
per_conv: Arc::new(Mutex::new(HashMap::new())),
|
||||
per_conv_permits: Arc::new(AtomicUsize::new(per_conv)),
|
||||
per_provider: Arc::new(Mutex::new(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
/// 取全局并发 permit（重建后新请求自动走最新 Semaphore）
|
||||
/// 取会话级并发 permit(F-09 B 批5/决策 c-1)：run_agentic_loop 入口拿 1 持整个 loop 生命周期。
|
||||
/// 重建后(set_global)新请求自动走最新 Semaphore。
|
||||
pub async fn acquire_global(&self) -> tokio::sync::OwnedSemaphorePermit {
|
||||
let sema = self.global.lock().await.clone();
|
||||
sema.acquire_owned().await.expect("llm global semaphore closed")
|
||||
}
|
||||
|
||||
/// 取单对话并发 permit
|
||||
pub async fn acquire_per_conv(&self) -> tokio::sync::OwnedSemaphorePermit {
|
||||
let sema = self.per_conv.lock().await.clone();
|
||||
/// 取单对话内并发 permit(F-09 B 批5)：按 conv_id 取/建 Semaphore，permits=当前 per_conv_permits(默认 2)。
|
||||
/// lock HashMap → 无则建 → clone Arc → 释放 lock → acquire_owned。锁持有短(不含 await acquire)。
|
||||
pub async fn acquire_per_conv(&self, conv_id: &str) -> tokio::sync::OwnedSemaphorePermit {
|
||||
let sema = {
|
||||
let mut map = self.per_conv.lock().await;
|
||||
let permits = self.per_conv_permits.load(std::sync::atomic::Ordering::SeqCst);
|
||||
map.entry(conv_id.to_string())
|
||||
.or_insert_with(|| Arc::new(Semaphore::new(permits)))
|
||||
.clone()
|
||||
};
|
||||
sema.acquire_owned().await.expect("llm per_conv semaphore closed")
|
||||
}
|
||||
|
||||
/// F-09 B 批5：conv 退出清理。loop 结束 + 无 pending 审批时 remove 该 conv 的 Semaphore 条目。
|
||||
/// **时机由调用方判断**(agentic loop 退出点)：仅在确信无后续 acquire 时调用，否则误删会致
|
||||
/// 该 conv 下次 acquire 重建 Semaphore(限流计数清零，非致命，但语义偏离)。
|
||||
/// remove 后已持 permit 不受影响(permit 绑旧 Arc，随 Drop 释放)，仅阻止新条目累积。
|
||||
pub async fn release_conv(&self, conv_id: &str) {
|
||||
self.per_conv.lock().await.remove(conv_id);
|
||||
}
|
||||
|
||||
/// 重建全局 Semaphore（软收敛：旧 permit 不回收，待其释放后新限制完全生效）
|
||||
pub async fn set_global(&self, permits: usize) {
|
||||
*self.global.lock().await = Arc::new(Semaphore::new(permits));
|
||||
}
|
||||
|
||||
/// 重建单对话 Semaphore
|
||||
/// 重建单对话 Semaphore(F-09 B 批5 后 per_conv 为 HashMap)。
|
||||
/// 行为：更新 per_conv_permits(AtomicUsize) + 清空 HashMap(软收敛：旧 permit 随 Drop 释放，
|
||||
/// 新对话 acquire 用新 permits 值重建 Semaphore)。已建对话若仍在跑，旧 Semaphore 不变；
|
||||
/// 下次该 conv 新 acquire 时因 HashMap 已清空会重建为新 permits。
|
||||
pub async fn set_per_conv(&self, permits: usize) {
|
||||
*self.per_conv.lock().await = Arc::new(Semaphore::new(permits));
|
||||
self.per_conv_permits
|
||||
.store(permits, std::sync::atomic::Ordering::SeqCst);
|
||||
self.per_conv.lock().await.clear();
|
||||
}
|
||||
|
||||
/// F-260614-04: 取 per-provider 并发 permit(可选)。
|
||||
@@ -151,7 +189,7 @@ impl LlmConcurrency {
|
||||
/// 调用方(agentic loop)用法:
|
||||
/// ```ignore
|
||||
/// let _global_permit = llm_concurrency.acquire_global().await;
|
||||
/// let _per_conv_permit = llm_concurrency.acquire_per_conv().await;
|
||||
/// let _per_conv_permit = llm_concurrency.acquire_per_conv(&conv_id).await;
|
||||
/// let _provider_permit = llm_concurrency.acquire_for_provider(&provider_id).await;
|
||||
/// ```
|
||||
/// 三 permit 均绑 guard Drop 自动释放。None 时无 permit 需释放(行为同 F-01 前)。
|
||||
|
||||
Reference in New Issue
Block a user