197 lines
7.0 KiB
Rust
197 lines
7.0 KiB
Rust
//! 工作流 DAG 上下文注入 — 当会话关联工作流时,将活跃路径注入 system prompt
|
||
//!
|
||
//! T4 实现。提供两个函数:
|
||
//! - `build_dag_summary`: 从 DagDef 构建可读的 DAG 摘要文本
|
||
//! - `extract_active_path`: 从完整 DAG 中提取当前活跃路径(简化)
|
||
//!
|
||
//! 调用方(前端/workflow 引擎)在启动工作流时将摘要存入
|
||
//! `PerConvState.workflow_dag_summary`,agentic loop 自动注入 system prompt。
|
||
//!
|
||
//! 预留说明(dead_code):本模块的 3 个函数当前无生产调用方——
|
||
//! T4 消费侧已落地(`agentic/mod.rs` 在构建 system prompt 时读
|
||
//! `PerConvState.workflow_dag_summary` 并注入),但生产侧尚无人写入该字段
|
||
//! (`PerConvState::new` 初始化为 `None`,无代码调用 `build_dag_summary`)。
|
||
//! 即完整链路为「工作流启动器 → 调本模块产出摘要 → 写 workflow_dag_summary
|
||
//! → agentic loop 注入 system prompt」,中间「生产者」环节待接通。
|
||
//! 属为后续工作流上下文注入准备的预留,非真死代码,故 allow 抑制告警。
|
||
|
||
#![allow(dead_code)]
|
||
|
||
use df_workflow::dag_def::DagDef;
|
||
|
||
/// 从 DagDef 构建可读的 DAG 摘要文本
|
||
///
|
||
/// 输出格式(5 节点 DAG 为例):
|
||
/// ```
|
||
/// 步骤: 读源码 → 分析依赖 → 改配置 → 验证 → 提交
|
||
/// 当前: 改配置
|
||
/// 完成: 2/5
|
||
/// ```
|
||
pub fn build_dag_summary(dag: &DagDef, current_node_id: Option<&str>) -> String {
|
||
let total = dag.nodes.len();
|
||
if total == 0 {
|
||
return String::new();
|
||
}
|
||
|
||
// 拓扑排序(简化:按 edges 计算入度,分层输出)
|
||
let layers = topological_layers(dag);
|
||
|
||
// 构建步骤链
|
||
let step_chain: Vec<&str> = layers.iter()
|
||
.flat_map(|layer| layer.iter().map(|id| id.as_str()))
|
||
.collect();
|
||
let chain_text = step_chain.join(" → ");
|
||
|
||
// 当前节点
|
||
let current_label = current_node_id
|
||
.and_then(|id| dag.nodes.get(id))
|
||
.map(|n| n.label.as_deref().unwrap_or(&n.node_type))
|
||
.unwrap_or("");
|
||
|
||
let mut result = format!("步骤: {}\n", chain_text);
|
||
if !current_label.is_empty() {
|
||
result.push_str(&format!("当前: {}\n", current_label));
|
||
}
|
||
result.push_str(&format!("完成: 待执行器上报"));
|
||
|
||
result
|
||
}
|
||
|
||
/// 从 DagDef 构建活跃路径(从根节点到当前节点的路径 + 后续 2 层)
|
||
///
|
||
/// 返回简化的步骤链文本,适合注入 system prompt。
|
||
pub fn extract_active_path(dag: &DagDef, current_node_id: Option<&str>) -> Vec<String> {
|
||
let layers = topological_layers(dag);
|
||
|
||
if let Some(current) = current_node_id {
|
||
// 找到当前节点所在的层
|
||
let current_layer_idx = layers.iter().position(|layer| layer.iter().any(|x| x == current));
|
||
|
||
match current_layer_idx {
|
||
Some(idx) => {
|
||
// 从根到当前层的路径
|
||
let path: Vec<String> = layers[..=idx].iter()
|
||
.flat_map(|layer| layer.iter().map(|id| {
|
||
dag.nodes.get(id)
|
||
.map(|n| n.label.clone().unwrap_or_else(|| id.clone()))
|
||
.unwrap_or_else(|| id.clone())
|
||
}))
|
||
.collect();
|
||
|
||
// 后续 2 层
|
||
let next: Vec<String> = layers[idx + 1..]
|
||
.iter()
|
||
.take(2)
|
||
.flat_map(|layer| layer.iter().map(|id| {
|
||
dag.nodes.get(id)
|
||
.map(|n| n.label.clone().unwrap_or_else(|| id.clone()))
|
||
.unwrap_or_else(|| id.clone())
|
||
}))
|
||
.collect();
|
||
|
||
let mut result = path;
|
||
result.extend(next);
|
||
result
|
||
}
|
||
None => {
|
||
// 当前节点不在 DAG 中 → 返回所有节点标签
|
||
layers.iter().flat_map(|layer| layer.iter().map(|id| {
|
||
dag.nodes.get(id)
|
||
.map(|n| n.label.clone().unwrap_or_else(|| id.clone()))
|
||
.unwrap_or_else(|| id.clone())
|
||
})).collect()
|
||
}
|
||
}
|
||
} else {
|
||
// 无当前节点 → 返回所有节点标签
|
||
layers.iter().flat_map(|layer| layer.iter().map(|id| {
|
||
dag.nodes.get(id)
|
||
.map(|n| n.label.clone().unwrap_or_else(|| id.clone()))
|
||
.unwrap_or_else(|| id.clone())
|
||
})).collect()
|
||
}
|
||
}
|
||
|
||
/// 拓扑分层排序:按入度计算每层的节点
|
||
fn topological_layers(dag: &DagDef) -> Vec<Vec<String>> {
|
||
let mut in_degree: std::collections::HashMap<&str, usize> = dag.nodes.keys().map(|k| (k.as_str(), 0)).collect();
|
||
|
||
for edge in &dag.edges {
|
||
if let Some(deg) = in_degree.get_mut(edge.target.as_str()) {
|
||
*deg += 1;
|
||
}
|
||
}
|
||
|
||
let mut layers = Vec::new();
|
||
let mut remaining: std::collections::HashSet<&str> = dag.nodes.keys().map(|k| k.as_str()).collect();
|
||
|
||
while !remaining.is_empty() {
|
||
let current_layer: Vec<String> = remaining.iter()
|
||
.filter(|id| *in_degree.get(*id).unwrap_or(&0) == 0)
|
||
.map(|id| id.to_string())
|
||
.collect();
|
||
|
||
if current_layer.is_empty() {
|
||
// 有环 → 剩余节点全放同一层
|
||
layers.push(remaining.iter().map(|id| id.to_string()).collect());
|
||
break;
|
||
}
|
||
|
||
for id in ¤t_layer {
|
||
remaining.remove(id.as_str());
|
||
// 减少以该节点为起点的边的入度
|
||
for edge in &dag.edges {
|
||
if edge.source == *id {
|
||
if let Some(deg) = in_degree.get_mut(edge.target.as_str()) {
|
||
*deg = deg.saturating_sub(1);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
layers.push(current_layer);
|
||
}
|
||
|
||
layers
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use df_workflow::dag_def::{DagDef, NodeDef, EdgeDef};
|
||
|
||
fn make_dag() -> DagDef {
|
||
let mut dag = DagDef::new();
|
||
dag.add_node("n1".to_string(), "read_file".to_string(), serde_json::json!({}));
|
||
dag.add_node("n2".to_string(), "analyze".to_string(), serde_json::json!({}));
|
||
dag.add_node("n3".to_string(), "modify".to_string(), serde_json::json!({}));
|
||
dag.add_edge("n1".to_string(), "n2".to_string());
|
||
dag.add_edge("n2".to_string(), "n3".to_string());
|
||
dag
|
||
}
|
||
|
||
#[test]
|
||
fn test_topological_layers_linear() {
|
||
let dag = make_dag();
|
||
let layers = topological_layers(&dag);
|
||
assert_eq!(layers.len(), 3);
|
||
assert_eq!(layers[0], vec!["n1"]);
|
||
assert_eq!(layers[1], vec!["n2"]);
|
||
assert_eq!(layers[2], vec!["n3"]);
|
||
}
|
||
|
||
#[test]
|
||
fn test_extract_active_path_midway() {
|
||
let dag = make_dag();
|
||
let path = extract_active_path(&dag, Some("n2"));
|
||
assert_eq!(path.len(), 3); // n1, n2, n3
|
||
assert_eq!(path[1], "n2");
|
||
}
|
||
|
||
#[test]
|
||
fn test_build_dag_summary_empty() {
|
||
let dag = DagDef::new();
|
||
assert_eq!(build_dag_summary(&dag, None), "");
|
||
}
|
||
}
|