From 4b5f096d1c7795da3a91cef7813e5985b68b9e42 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Sun, 14 Jun 2026 22:56:06 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96:=20dag=E6=8B=93=E6=89=91O(V+?= =?UTF-8?q?E)+=E5=89=8D=E7=AB=AFfindToolCall=E5=8F=8D=E5=90=91=E9=81=8D?= =?UTF-8?q?=E5=8E=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - FR-D1 dag.rs topological_layers + executor.rs run 建 adjacency 索引, O(V·E)→O(V+E)(14测试全绿, 含executor3测) - FR-R5 useAiEvents findToolCall 正向 O(n²) 改反向遍历(命中最近, 放弃Map索引防陈旧引用) - AR-6 核查确认已于 f82dd8b 落地(Low失败emit AiToolCallCompleted), 零改动 --- crates/df-workflow/src/dag.rs | 30 ++++++++++++++++++++++-------- crates/df-workflow/src/executor.rs | 20 ++++++++++++++++---- src/composables/ai/useAiEvents.ts | 24 +++++++++++++++++++----- 3 files changed, 57 insertions(+), 17 deletions(-) diff --git a/crates/df-workflow/src/dag.rs b/crates/df-workflow/src/dag.rs index 0241e81..728c6af 100644 --- a/crates/df-workflow/src/dag.rs +++ b/crates/df-workflow/src/dag.rs @@ -81,18 +81,29 @@ impl Dag { /// 拓扑排序 — 返回按执行顺序排列的节点 ID 层级 /// - /// 返回 Vec>，每层内的节点可以并行执行 + /// 返回 Vec>，每层内的节点可以并行执行。 + /// + /// 复杂度：O(V+E)。一次性遍历边构建邻接索引（出边表）与入度表， + /// BFS 分层只走索引查询（每条边仅被访问一次），不再为每个节点全表扫描边集合。 pub fn topological_layers(&self) -> Result>> { let node_ids: HashSet = self.nodes.keys().cloned().collect(); let mut in_degree: HashMap = HashMap::new(); + // 邻接出边表：source → 直接后继列表（仅含两端均在 nodes 内的有效边） + let mut adjacency_out: HashMap> = HashMap::new(); - // 初始化入度 + // 初始化入度（确保每个节点都有表项，便于后续 O(1) 修改） for id in &node_ids { in_degree.insert(id.clone(), 0); } + // 单次 O(E) 遍历：同时构建入度表与出边表 for edge in &self.edges { + // 仅收录两端均为已注册节点的边，跳过野节点 if node_ids.contains(&edge.source) && node_ids.contains(&edge.target) { - *in_degree.entry(edge.target.clone()).or_insert(0) += 1; + *in_degree.get_mut(&edge.target).unwrap() += 1; + adjacency_out + .entry(edge.source.clone()) + .or_default() + .push(edge.target.clone()); } } @@ -111,11 +122,14 @@ impl Dag { for _ in 0..layer_size { if let Some(id) = queue.pop_front() { layer.push(id.clone()); - for succ in self.successors(&id) { - let deg = in_degree.get_mut(&succ).unwrap(); - *deg -= 1; - if *deg == 0 { - queue.push_back(succ); + // 出边已索引，O(出度) 遍历而非 O(E) 全表扫描 + if let Some(succs) = adjacency_out.get(&id) { + for succ in succs { + let deg = in_degree.get_mut(succ).unwrap(); + *deg -= 1; + if *deg == 0 { + queue.push_back(succ.clone()); + } } } processed += 1; diff --git a/crates/df-workflow/src/executor.rs b/crates/df-workflow/src/executor.rs index f939984..410d2ad 100644 --- a/crates/df-workflow/src/executor.rs +++ b/crates/df-workflow/src/executor.rs @@ -57,6 +57,16 @@ impl DagExecutor { tracing::info!("DAG 执行开始，共 {} 层", layers.len()); + // 预建入边索引：target → 直接前驱列表（O(E) 一次构建） + // 避免在内层按节点循环中调用 dag.predecessors()（每次 O(E) 全表扫描，整体 O(V·E)）。 + let mut adjacency_in: HashMap> = HashMap::new(); + for edge in &dag.edges { + adjacency_in + .entry(edge.target.clone()) + .or_default() + .push(edge.source.clone()); + } + for (layer_idx, layer) in layers.iter().enumerate() { tracing::info!("执行第 {} 层，共 {} 个节点", layer_idx, layer.len()); @@ -76,11 +86,13 @@ impl DagExecutor { self.state_machine.set_running(node_id.clone())?; - // 构建节点上下文 + // 构建节点上下文：通过入边索引 O(入度) 取前驱，而非 O(E) 全表扫描 let mut inputs = HashMap::new(); - for pred_id in dag.predecessors(node_id) { - if let Some(out) = outputs.get(&pred_id) { - inputs.insert(pred_id, out.clone()); + if let Some(preds) = adjacency_in.get(node_id) { + for pred_id in preds { + if let Some(out) = outputs.get(pred_id) { + inputs.insert(pred_id.clone(), out.clone()); + } } } diff --git a/src/composables/ai/useAiEvents.ts b/src/composables/ai/useAiEvents.ts index 874b314..3c21e0b 100644 --- a/src/composables/ai/useAiEvents.ts +++ b/src/composables/ai/useAiEvents.ts @@ -59,12 +59,26 @@ export function flushCurrentText() { if (last && last.role === 'assistant') last.content = state.currentText } -/** 在全部消息中查找指定 id 的工具调用卡片(用于状态流转) */ +/** + * 在全部消息中查找指定 id 的工具调用卡片(用于状态流转) + * + * 扫描策略:从尾部反向遍历并提前退出。tool_result/approval/completed + * 通常对应最近触发的 tool_use(agent 顺序执行工具并尽快回填结果), + * 反向扫描使其命中落在尾部 → 平均 O(1)、最坏 O(n)。(原正向全量扫描为 O(n²), + * 每条 tool_result 都从头扫整条历史。) + * + * 注:未引入 Map 索引。state.messages 在 newConversation / + * switchConversation / deleteConversation 等处被整体替换(且这些操作散落在 + * useAiEvents 之外),独立索引极易与响应式数组失配形成"陈旧引用"——状态卡 + * 片读到已不存在的对象。反向扫描零额外状态、改动最小且行为完全一致。 + */ export function findToolCall(id: string): AiToolCallInfo | undefined { - for (const msg of state.messages) { - if (msg.toolCalls) { - const tc = msg.toolCalls.find(t => t.id === id) - if (tc) return tc + for (let i = state.messages.length - 1; i >= 0; i--) { + const toolCalls = state.messages[i].toolCalls + if (!toolCalls) continue + // 同一消息内工具调用通常很少,沿用正向扫描;命中即返回 + for (let j = 0; j < toolCalls.length; j++) { + if (toolCalls[j].id === id) return toolCalls[j] } } return undefined