diff --git a/crates/df-nodes/src/ai_node.rs b/crates/df-nodes/src/ai_node.rs index efb6d56..b822de6 100644 --- a/crates/df-nodes/src/ai_node.rs +++ b/crates/df-nodes/src/ai_node.rs @@ -7,9 +7,12 @@ //! 适合嵌入自动化链路(如 想法 → AI分析 → 脚本落地 → 人工审批)。 use std::collections::HashMap; +use std::sync::Arc; use async_trait::async_trait; use df_ai::provider::{ChatMessage, CompletionRequest, LlmProvider}; +use df_storage::crud::TaskRepo; +use df_storage::db::Database; use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; /// AI 节点解析后的参数(execute 与参数解析解耦,便于单测覆盖取值/默认/校验逻辑) @@ -109,7 +112,20 @@ fn parse_params( } /// AI 节点 -pub struct AiNode; +/// +/// 持有 Arc 在 NodeRegistry 注册时注入(state.rs build_registry 工厂闭包 move 捕获 +/// db.clone(),对齐 TaskAdvanceNode 先例)。execute 从 NodeContext.config 读 provider 配置 + +/// 可选 task_id:有 task_id 时把产出落 task.output_json(AiNode 自审闭环,决策 a)。 +pub struct AiNode { + db: Arc, +} + +impl AiNode { + /// 构造节点(注册表工厂调用,注入数据库句柄) + pub fn new(db: Arc) -> Self { + Self { db } + } +} #[async_trait] impl Node for AiNode { @@ -152,7 +168,7 @@ impl Node for AiNode { response.usage.completion_tokens ); - Ok(NodeOutput::from_value(serde_json::json!({ + let output = serde_json::json!({ "text": response.text, "model": response.model, "usage": { @@ -160,7 +176,25 @@ impl Node for AiNode { "completion_tokens": response.usage.completion_tokens, "total_tokens": response.usage.total_tokens, }, - }))) + }); + + // 决策 a(AiNode 自审闭环):有 task_id 时落产出到 task.output_json。 + // 走通用 update_field(output_json 已在 tasks 白名单,crud.rs:342-346), + // 非 status 状态机收口字段,合法可写。落库失败不阻断节点返回(产出已在内存,工作流可继续)。 + if let Some(task_id) = ctx.config.get("task_id").and_then(|v| v.as_str()) { + if let Ok(json_str) = serde_json::to_string(&output) { + let repo = TaskRepo::new(&self.db); + if let Err(e) = repo.update_field(task_id, "output_json", &json_str).await { + tracing::warn!( + task_id = task_id, + error = %e, + "AiNode 落 output_json 失败(产出已在内存,节点继续)" + ); + } + } + } + + Ok(NodeOutput::from_value(output)) } fn schema(&self) -> NodeSchema { @@ -195,6 +229,257 @@ impl Node for AiNode { } } +// ============================================================ +// ai_self_review 节点 — AI 自审闭环(决策 a 步骤③) +// ============================================================ +// +// 复用 AiNode 的 provider 调用逻辑(parse_params + build_provider + complete),差异: +// 1. prompt 固定四维度审查模板(需求符合度/完整性/正确性/边界),不从 config/inputs 取 prompt +// 2. system_prompt 强约束「只输出 JSON」+ temperature=0 +// 3. LLM 输出按结构化 JSON 解析,失败兜底 verdict=unknown(防 LLM 不按要求输出) +// 4. 自审输入:task.description(需求,经 ctx.config["task_id"] 读 task) + task.output_json.text(产出) +// 5. 自审结果写回 task.output_json 加 review 子字段(不覆盖 text,保留对照) +// 6. review 摘要塞 NodeOutput.data,供下游 human_review 经 DAG inputs["ai_self_review"] 读 +// +// verdict 处理(首版保守):自审是辅助,最终决策权在人。verdict=fail 不主动 Err, +// 节点返 Ok 让 human_review 继续,人在审批卡看到 fail 结论再定(human reject 走 Err→工作流 +// failed→②-4 回调退回 in_review)。 + +/// 自审四维度 system prompt:严格审查员角色 + 只输出 JSON 强约束。 +const REVIEW_SYSTEM_PROMPT: &str = "\ +你是严格的代码/产出审查员。审查任务产出是否符合需求,按四维度给出结构化结论。\ +只输出 JSON,不要任何额外文字、不要 markdown 代码块包裹。"; + +/// 解析 LLM 自审输出为结构化 review JSON。 +/// +/// 成功路径:serde_json::from_str 得到 Object 且含 verdict 字段 → 原样返回。 +/// 兜底路径:解析失败 / 非 Object / 缺 verdict → 返回 verdict=unknown + summary=原文, +/// 防 LLM 不按要求输出导致下游崩溃。dimensions 留空对象(前端容缺展示)。 +fn parse_review_json(raw: &str) -> serde_json::Value { + // 先尝试整段解析;LLM 偶尔会包 markdown 代码块,剥离 ```json ... ``` 后重试一次。 + let trimmed = raw.trim(); + let cleaned = trimmed + .strip_prefix("```json") + .or_else(|| trimmed.strip_prefix("```")) + .map(|s| s.trim_end_matches("```").trim()) + .unwrap_or(trimmed); + + if let Ok(v) = serde_json::from_str::(cleaned) { + if v.is_object() && v.get("verdict").and_then(|x| x.as_str()).is_some() { + return v; + } + } + // 兜底:保留原文供人查阅,verdict=unknown 不阻断流程(自审辅助,人定)。 + serde_json::json!({ + "verdict": "unknown", + "dimensions": {}, + "summary": format!("(自审输出解析失败,原文: {})", truncate_for_summary(cleaned)), + "suggestions": [], + }) +} + +/// summary 截断(防原文过长撑爆 output_json / 审批卡)。 +fn truncate_for_summary(s: &str) -> String { + const MAX: usize = 300; + if s.chars().count() <= MAX { + return s.to_string(); + } + let truncated: String = s.chars().take(MAX).collect(); + format!("{truncated}…") +} + +/// AI 自审节点(决策 a 步骤③) +pub struct AiSelfReviewNode { + db: Arc, +} + +impl AiSelfReviewNode { + pub fn new(db: Arc) -> Self { + Self { db } + } + + /// 拼装自审 user prompt:任务需求 + 待审产出 + 四维度审查要求 + 输出格式。 + /// description / output_text 缺失时给占位(不报错,信任调用方注入合法 task_id)。 + fn build_review_prompt(description: &str, output_text: &str) -> String { + format!( + "\ +## 任务需求 +{description} + +## 待审产出 +{output_text} + +## 审查维度 +1. 需求符合度:产出是否覆盖需求描述的所有要点 +2. 产出完整性:是否有遗漏、未完成的部分 +3. 正确性:逻辑/事实/语法是否正确 +4. 边界处理:异常输入、空值、错误路径是否考虑 + +## 输出格式(严格 JSON) +{{\"verdict\":\"pass\"|\"fail\",\"dimensions\":{{\"requirement_fit\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"completeness\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"correctness\":{{\"score\":0-10,\"issues\":[\"...\"]}},\"boundary\":{{\"score\":0-10,\"issues\":[\"...\"]}}}},\"summary\":\"一句话总结\",\"suggestions\":[\"改进建议\"]}} +verdict=fail 当且仅当任一维度 score<6 或有阻断性 issue。" + ) + } +} + +#[async_trait] +impl Node for AiSelfReviewNode { + async fn execute(&self, ctx: NodeContext) -> NodeResult { + tracing::info!("AiSelfReviewNode 执行: node_id={}", ctx.node_id); + + let p = parse_params(&ctx.config, &ctx.inputs)?; + + // ── 读任务(需求 + 产出) ── + // task_id 必填(自审对象是具体任务的产出);缺失报错而非静默跳过。 + let task_id = ctx + .config + .get("task_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("AiSelfReviewNode 缺少必填参数: task_id"))?; + + let repo = TaskRepo::new(&self.db); + let task = repo + .get_by_id(task_id) + .await + .map_err(|e| anyhow::anyhow!("自审读任务失败: {}", e))? + .ok_or_else(|| anyhow::anyhow!("自审任务 {} 不存在", task_id))?; + + // output_json 可能无(ai_execute 未跑或失败) → 待审产出占位提示,不报错。 + let output_text = task + .output_json + .as_deref() + .and_then(|s| serde_json::from_str::(s).ok()) + .and_then(|v| v.get("text").and_then(|t| t.as_str()).map(String::from)) + .unwrap_or_else(|| "(无产出,仅依据需求审查)".to_string()); + + let prompt = Self::build_review_prompt(&task.description, &output_text); + + let provider: Box = + df_ai::build_provider(&p.protocol, &p.base_url, &p.api_key, &p.default_model); + + let request = CompletionRequest { + model: p.model.clone(), + messages: vec![ + ChatMessage::system(REVIEW_SYSTEM_PROMPT), + ChatMessage::user(prompt), + ], + temperature: Some(0.0), // 强约束确定性输出 + max_tokens: p.max_tokens, + stream: false, + tools: None, + tool_choice: None, + }; + + tracing::info!( + "AiSelfReviewNode 调用 LLM: model={}, base_url={}", + p.default_model, + p.base_url + ); + let response = provider.complete(request).await?; + + // ── 解析自审 JSON(失败兜底 verdict=unknown) ── + let review = parse_review_json(&response.text); + let verdict = review + .get("verdict") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + // summary / suggestions 在 move review 前先取走(供 NodeOutput.data 平铺) + let summary = review + .get("summary") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let suggestions = review + .get("suggestions") + .cloned() + .unwrap_or(serde_json::json!([])); + + tracing::info!( + "AiSelfReviewNode 完成: verdict={}, model={}", + verdict, + response.model + ); + + // ── 写回 task.output_json 加 review 子字段(不覆盖 text) ── + // 读现有 output_json(若有),合并 review 子字段后整体写回。 + let mut output_obj: serde_json::Map = task + .output_json + .as_deref() + .and_then(|s| serde_json::from_str::(s).ok()) + .and_then(|v| v.as_object().cloned()) + .unwrap_or_default(); + // review 子字段:自审结论 + reviewed_at + reviewer_model(便于前端展示与人对照) + let mut review_with_meta = match review { + serde_json::Value::Object(m) => m, + other => { + let mut m = serde_json::Map::new(); + m.insert("verdict".into(), other); + m + } + }; + review_with_meta.insert( + "reviewed_at".into(), + serde_json::Value::Number(df_core::now_millis().into()), + ); + review_with_meta.insert("reviewer_model".into(), serde_json::json!(response.model)); + output_obj.insert("review".into(), serde_json::Value::Object(review_with_meta)); + + let merged = serde_json::Value::Object(output_obj); + if let Ok(json_str) = serde_json::to_string(&merged) { + if let Err(e) = repo.update_field(task_id, "output_json", &json_str).await { + tracing::warn!( + task_id = task_id, + error = %e, + "AiSelfReviewNode 落 review 子字段失败(摘要已在 NodeOutput,节点继续)" + ); + } + } + + // ── NodeOutput.data 塞 review 摘要,供下游 human_review 经 inputs["ai_self_review"] 读 ── + // human_review(HumanNode)从 config/inputs 拼 description;此处把 verdict/summary/suggestions + // 平铺进 data,HumanNode 零改动即可透传(前端/模板层可据此拼审批卡描述)。 + Ok(NodeOutput::from_value(serde_json::json!({ + "verdict": verdict, + "summary": summary, + "suggestions": suggestions, + "review": merged.get("review").cloned().unwrap_or(serde_json::Value::Null), + "model": response.model, + }))) + } + + fn schema(&self) -> NodeSchema { + NodeSchema { + params: serde_json::json!({ + "type": "object", + "properties": { + "task_id": { "type": "string", "description": "自审目标任务 ID(必填)" }, + "protocol": { "type": "string", "description": "协议类型:openai_compat(默认)或 anthropic" }, + "base_url": { "type": "string" }, + "api_key": { "type": "string" }, + "model": { "type": "string" }, + "max_tokens": { "type": "integer" } + }, + "required": ["task_id", "base_url", "api_key"] + }), + output: serde_json::json!({ + "type": "object", + "properties": { + "verdict": { "type": "string", "enum": ["pass", "fail", "unknown"] }, + "summary": { "type": "string" }, + "suggestions": { "type": "array", "items": { "type": "string" } }, + "review": { "type": "object" }, + "model": { "type": "string" } + } + }), + } + } + + fn node_type(&self) -> &str { + "ai_self_review" + } +} + #[cfg(test)] mod tests { use super::*; @@ -327,4 +612,136 @@ mod tests { response.model, response.text, response.usage ); } + + // ============================================================ + // 决策 a 自审闭环单测 — output_json 写回 + review JSON 解析兜底 + // ============================================================ + + /// 内存 DB 构造 helper:插父项目 + 指定 task,返回 (db, task_id)。 + async fn setup_task_db(task_output_json: Option<&str>) -> (Database, String) { + use df_storage::crud::ProjectRepo; + use df_storage::models::{ProjectRecord, TaskRecord}; + let db = Database::open_in_memory().await.expect("open_in_memory"); + ProjectRepo::new(&db) + .insert(ProjectRecord { + id: "p1".to_string(), + name: "proj".to_string(), + description: "".to_string(), + status: "planning".to_string(), + idea_id: None, + path: None, + stack: None, + created_at: "0".to_string(), + updated_at: "0".to_string(), + }) + .await + .expect("insert project"); + TaskRepo::new(&db) + .insert(TaskRecord { + id: "t1".to_string(), + project_id: "p1".to_string(), + title: "t1".to_string(), + description: "实现登录接口".to_string(), + status: "testing".to_string(), + priority: 2, + branch_name: None, + assignee: None, + workflow_def_id: None, + base_branch: None, + review_rounds: 0, + output_json: task_output_json.map(String::from), + created_at: "0".to_string(), + updated_at: "0".to_string(), + }) + .await + .expect("insert task"); + (db, "t1".to_string()) + } + + /// 决策 a 步骤②:AiNode.execute 落产出验证 — 有 task_id 时 output_json 写入。 + /// 用 mock provider 避免真调 LLM:自定义 Node 替换不便,改为直接验证 update_field 路径 + /// (execute 内部调 provider.complete 需真 LLM,此处降级验证 parse_review_json 兜底 + + /// 直接 repo.update_field 路径,与 execute 落库逻辑同一条 crud.update_field)。 + #[tokio::test] + async fn update_field_writes_output_json() { + // 验证 output_json 在 tasks 白名单(update_field 不报「不允许的字段名」)。 + // 这正是 AiNode.execute / AiSelfReviewNode.execute 落产出的底层路径。 + let (db, task_id) = setup_task_db(None).await; + let repo = TaskRepo::new(&db); + let payload = r#"{"text":"产出内容","model":"glm-4","usage":{}}"#; + let ok = repo + .update_field(&task_id, "output_json", payload) + .await + .expect("update_field output_json"); + assert!(ok, "update_field 应返回 affected>0=true"); + let after = repo.get_by_id(&task_id).await.unwrap().unwrap(); + assert_eq!( + after.output_json.as_deref(), + Some(payload), + "output_json 应被写入" + ); + // 原始 text 字段(产出对照)保留 + let parsed: serde_json::Value = serde_json::from_str(after.output_json.as_deref().unwrap()).unwrap(); + assert_eq!(parsed["text"], json!("产出内容")); + } + + /// 步骤③:parse_review_json 合规 JSON 直通(保留 verdict/dimensions/summary/suggestions)。 + #[test] + fn parse_review_json_valid_passes_through() { + let raw = r#"{"verdict":"pass","dimensions":{"requirement_fit":{"score":9,"issues":[]}},"summary":"符合需求","suggestions":["可加单测"]}"#; + let v = parse_review_json(raw); + assert_eq!(v["verdict"], json!("pass")); + assert_eq!(v["summary"], json!("符合需求")); + assert_eq!(v["suggestions"], json!(["可加单测"])); + assert_eq!(v["dimensions"]["requirement_fit"]["score"], json!(9)); + } + + /// 步骤③:parse_review_json 非 JSON / 缺 verdict → 兜底 verdict=unknown + summary 含原文。 + #[test] + fn parse_review_json_invalid_falls_back_unknown() { + // 非 JSON + let v = parse_review_json("这不是 JSON,LLM 没按要求输出"); + assert_eq!(v["verdict"], json!("unknown")); + assert!(v["summary"].as_str().unwrap().contains("不是 JSON")); + assert!(v["summary"].as_str().unwrap().contains("解析失败")); + + // JSON 但缺 verdict + let v = parse_review_json(r#"{"foo":"bar"}"#); + assert_eq!(v["verdict"], json!("unknown")); + + // JSON 非 object(数组) + let v = parse_review_json("[1,2,3]"); + assert_eq!(v["verdict"], json!("unknown")); + } + + /// 步骤③:LLM 偶尔包 ```json 代码块,parse_review_json 应剥离后解析成功。 + #[test] + fn parse_review_json_strips_markdown_code_fence() { + let raw = "```json\n{\"verdict\":\"fail\",\"summary\":\"缺边界处理\"}\n```"; + let v = parse_review_json(raw); + assert_eq!(v["verdict"], json!("fail")); + assert_eq!(v["summary"], json!("缺边界处理")); + } + + /// 步骤③:truncate_for_summary 长文截断。 + #[test] + fn truncate_for_summary_long_text() { + let short = "短文"; + assert_eq!(truncate_for_summary(short), "短文"); + let long: String = "字".repeat(500); + let t = truncate_for_summary(&long); + assert!(t.ends_with("…")); + assert!(t.chars().count() <= 302, "截断后含省略号应 ≈300 字"); + } + + /// 步骤③:AiSelfReviewNode.build_review_prompt 含需求 + 产出 + 四维度。 + #[test] + fn build_review_prompt_contains_inputs() { + let p = AiSelfReviewNode::build_review_prompt("登录接口", "已实现 /login"); + assert!(p.contains("登录接口"), "prompt 应含需求"); + assert!(p.contains("已实现 /login"), "prompt 应含产出"); + assert!(p.contains("需求符合度"), "prompt 应含四维度"); + assert!(p.contains("边界处理")); + assert!(p.contains("verdict"), "prompt 应含输出格式约束"); + } } diff --git a/crates/df-nodes/src/task_workflow_templates.rs b/crates/df-nodes/src/task_workflow_templates.rs index dd0ab3c..5927633 100644 --- a/crates/df-nodes/src/task_workflow_templates.rs +++ b/crates/df-nodes/src/task_workflow_templates.rs @@ -56,7 +56,9 @@ fn in_progress_template() -> DagDef { /// 通过则完成回调(②-3)推进 status 到 testing。 fn testing_template() -> DagDef { let mut dag = DagDef::new(); - dag.add_node("ai_self_review", "ai", serde_json::json!({})); + // 决策 a 步骤③:ai_self_review 节点类型对齐 state.rs 注册的独立自审节点 + // (四维度 prompt + JSON 解析兜底 + 写回 output_json 加 review 子字段)。 + dag.add_node("ai_self_review", "ai_self_review", serde_json::json!({})); dag.add_node( "human_review", "human", @@ -122,9 +124,9 @@ mod tests { let dag = template_for("testing").expect("testing 模板存在"); assert_eq!(dag.nodes.len(), 2, "testing: ai + human 两节点"); assert_eq!(dag.edges.len(), 1, "testing: 单串行边"); - // 节点类型 + // 节点类型(决策 a 步骤③:ai_self_review 独立节点类型) let ai = dag.nodes.get("ai_self_review").expect("ai_self_review 存在"); - assert_eq!(ai.node_type, "ai"); + assert_eq!(ai.node_type, "ai_self_review"); let human = dag.nodes.get("human_review").expect("human_review 存在"); assert_eq!(human.node_type, "human"); // 边方向:ai → human diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index 5095207..8b4892d 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -254,8 +254,20 @@ fn build_registry(db: Arc) -> NodeRegistry { registry.register("human", |_config| { Box::new(df_nodes::human_node::HumanNode) }); - registry.register("ai", |_config| { - Box::new(df_nodes::ai_node::AiNode) + // AiNode(df_nodes::ai_node, impl Node trait): + // 决策 a(AiNode 自审闭环)步骤②:AiNode 持 Arc,execute 完成后若有 task_id + // 则把产出落 task.output_json。工厂闭包 move 捕获 db 句柄(Arc clone 廉价)。 + let ai_db = db.clone(); + registry.register("ai", move |_config| { + Box::new(df_nodes::ai_node::AiNode::new(ai_db.clone())) + }); + // AiSelfReviewNode(df_nodes::ai_node, impl Node trait): + // 决策 a 步骤③:四维度自审(prompt 固定+JSON 解析兜底),写回 output_json 加 review 子字段, + // review 摘要塞 NodeOutput.data 供下游 human_review 经 DAG inputs 透传。 + // testing 模板 ai_self_review 节点类型对齐此注册 key。 + let review_db = db.clone(); + registry.register("ai_self_review", move |_config| { + Box::new(df_nodes::ai_node::AiSelfReviewNode::new(review_db.clone())) }); // TaskAdvanceNode(df_nodes::task_advance_node, impl Node trait): // 推进链阶段 2 工作流联动入口 — DAG 内触发 advance_task。