1. 为什么 Agent 的“计划”总在真实任务里翻车如果你正在做 Agent 平台大概率遇到过这种场景让 Planner 输出一份步骤清单Executor 照着跑跑到第三步发现前置条件根本没满足或者两个步骤互相等待形成死锁。更麻烦的是上游需求改了一个参数整份计划要重新生成Token 哗哗地烧。我试过把计划从自然语言换成 JSON-LD 描述再用 SPARQL 把依赖闭包抽出来编译成一张 DAG。做完这件事之后调度器第一次有了“地图”而不是“猜谜”。这篇就聚焦这条编译链路JSON-LD 怎么定义节点和边SPARQL 怎么查依赖Rust 侧怎么把结果拓扑排序成 DAG以及怎么用查询验证编译结果的一致性。适合谁看正在搭 Agent 编排层、知识图谱驱动任务系统、或者想用 Rust 写调度器的同学。你需要一点 RDF/SPARQL 基础但不需要是语义网专家。下面所有配置和命令都可以直接复制跑。核心检索词先摆出来JSON-LD 编译 DAG、Agent 调度器、SPARQL 图遍历、Rust Agent OS、任务 DAG 生成。这几个词贯穿全文你按需跳读。2. 前置准备TaoToken 接入与 Rust 工程骨架2.1 为什么这里需要 TaoToken编译链路本身是纯本地的图计算但 Agent 平台里 Planner 审查 DAG、CA 校验节点这些环节要调模型。TaoToken 提供统一的模型接入层省去你在多个供应商之间切来切去。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 注意 API 地址不带 UTM 参数。你需要先去控制台拿一个 Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 然后在 API Keys 页面创建https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。拿到之后写进环境变量后面 Rust 侧用 reqwest 调用。export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意Key 不要硬编码进仓库用环境变量或 .env 文件.env 记得进 .gitignore。2.2 Cargo.toml 依赖骨架DAG 编译需要三块能力RDF/JSON-LD 解析、SPARQL 查询、图算法。Rust 生态里 oxigraph 同时覆盖前两块petgraph 负责拓扑排序。[package] name agent-harness-dag version 0.1.0 edition 2021 [dependencies] oxigraph 0.4 petgraph 0.6 serde { version 1, features [derive] } serde_json 1 tokio { version 1, features [full] } reqwest { version 0.12, features [json] } anyhow 1 thiserror 1 tracing 0.1 tracing-subscriber 0.3oxigraph 既是三元组存储又能跑 SPARQL省得你再接一个外部图数据库。petgraph 的toposort直接给你拓扑序检测环也方便。2.3 目录结构agent-harness-dag/ ├── Cargo.toml ├── src/ │ ├── main.rs │ ├── compiler.rs # JSON-LD - DAG 编译入口 │ ├── dag.rs # DAG 节点/边定义 │ └── verify.rs # SPARQL 一致性校验 └── data/ └── skills.jsonld # 技能图谱3. 可复制配置DAG 节点定义与编译入口3.1 DAG 节点与边的数据结构先定义 Rust 侧的 DAG 表示。节点带 IRI、类型、执行模式边带执行语义顺序、并行、条件。// src/dag.rs use serde::{Deserialize, Serialize}; use std::collections::HashMap; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)] pub struct DagNode { pub iri: String, pub node_type: String, pub skill: String, pub assigned_to: OptionString, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub enum EdgeKind { Then, // exec:then 顺序 Parallel, // exec:parallel 并行 Conditional, // exec:conditional 条件 } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DagEdge { pub from: String, pub to: String, pub kind: EdgeKind, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TaskDag { pub root: String, pub nodes: HashMapString, DagNode, pub edges: VecDagEdge, }assigned_to是给调度器用的哪个 Agent 负责这个节点。编译阶段可以先留空由 SA 后续填充。3.2 JSON-LD 技能图谱样例技能之间的依赖用skill:requires表达任务用task:how指向技能。这份数据放进data/skills.jsonld。{ context: { skill: https://example.org/skill#, task: https://example.org/task#, exec: https://example.org/exec#, requires: { id: skill:requires, type: id }, how: { id: task:how, type: id } }, graph: [ { id: skill:rust-jwt-auth, type: skill:Skill, skill:requires: skill:rust-basics }, { id: skill:rust-basics, type: skill:Skill, skill:requires: skill:cargo-setup }, { id: skill:cargo-setup, type: skill:Skill }, { id: skill:token-security, type: skill:Skill, skill:requires: skill:hash-algorithms }, { id: skill:hash-algorithms, type: skill:Skill }, { id: task:jwt-refactor, type: task:ImplementationTask, task:how: [skill:rust-jwt-auth, skill:token-security] } ] }3.3 编译入口SPARQL 抽依赖闭包编译的核心是一条 SPARQL CONSTRUCT把skill:requires的传递闭包抽出来再交给 Rust 建图。// src/compiler.rs use oxigraph::store::Store; use oxigraph::sparql::QueryResults; use crate::dag::{DagEdge, DagNode, EdgeKind, TaskDag}; use std::collections::HashMap; const DEP_CLOSURE_QUERY: str r# PREFIX skill: https://example.org/skill# PREFIX task: https://example.org/task# CONSTRUCT { ?skill skill:requires ?dep . ?dep skill:requires ?transitive . } WHERE { BIND (skill:rust-jwt-auth AS ?skill) ?skill skill:requires* ?dep . OPTIONAL { ?dep skill:requires ?transitive . } } #; pub fn compile(store: Store, root_task: str) - anyhow::ResultTaskDag { let mut nodes: HashMapString, DagNode HashMap::new(); let mut edges: VecDagEdge Vec::new(); let results store.query(DEP_CLOSURE_QUERY)?; if let QueryResults::Graph(g) results { for triple in g { let t triple?; let from t.subject.to_string(); let to t.object.to_string(); nodes.entry(from.clone()).or_insert(DagNode { iri: from.clone(), node_type: skill:Skill.into(), skill: from.clone(), assigned_to: None, }); nodes.entry(to.clone()).or_insert(DagNode { iri: to.clone(), node_type: skill:Skill.into(), skill: to.clone(), assigned_to: None, }); edges.push(DagEdge { from, to, kind: EdgeKind::Then }); } } Ok(TaskDag { root: root_task.to_string(), nodes, edges }) }skill:requires*是零到多跳的传递闭包OPTIONAL保证叶子节点也能被带出来。编译出来的边默认是Then后续可以根据exec:parallel标注改写。3.4 拓扑排序与环检测拿到边之后用 petgraph 建图并拓扑排序。有环直接报错因为 DAG 不允许环。// src/compiler.rs 续 use petgraph::graph::DiGraph; use petgraph::algo::toposort; pub fn toposort_dag(dag: TaskDag) - anyhow::ResultVecString { let mut g: DiGraphString, () DiGraph::new(); let mut idx HashMap::new(); for iri in dag.nodes.keys() { let i g.add_node(iri.clone()); idx.insert(iri.clone(), i); } for e in dag.edges { let a idx[e.from]; let b idx[e.to]; g.add_edge(a, b, ()); } match toposort(g, None) { Ok(order) Ok(order.into_iter().map(|i| g[i].clone()).collect()), Err(_) anyhow::bail!(检测到环依赖DAG 编译失败), } }拓扑序就是调度器的执行顺序基线。并行分支在拓扑序里是相邻的SA 可以据此决定并发度。4. 验证请求用 SPARQL 校验编译结果一致性4.1 加载数据并跑编译main.rs里把 JSON-LD 读进 Store跑编译打印拓扑序。// src/main.rs mod compiler; mod dag; mod verify; use oxigraph::store::Store; use oxigraph::io::RdfFormat; use std::fs::File; use std::io::BufReader; #[tokio::main] async fn main() - anyhow::Result() { tracing_subscriber::fmt::init(); let store Store::new()?; let file File::open(data/skills.jsonld)?; store.load_from_reader(RdfFormat::JsonLd { profile: Default::default() }, BufReader::new(file))?; let dag compiler::compile(store, task:jwt-refactor)?; let order compiler::toposort_dag(dag)?; println!(拓扑序: {:?}, order); Ok(()) }跑起来cargo run预期输出类似拓扑序: [skill:cargo-setup, skill:rust-basics, skill:rust-jwt-auth, skill:hash-algorithms, skill:token-security]顺序说明cargo-setup在最前token-security在最后符合依赖方向。4.2 一致性校验查询编译完要验证两件事所有叶子节点是否可达是否存在孤立节点。用 SPARQL 查。// src/verify.rs use oxigraph::store::Store; use oxigraph::sparql::QueryResults; const LEAF_CHECK: str r# PREFIX skill: https://example.org/skill# SELECT ?leaf WHERE { ?leaf a skill:Skill . FILTER NOT EXISTS { ?leaf skill:requires ?any . } } #; pub fn check_leaves(store: Store) - anyhow::ResultVecString { let mut leaves Vec::new(); if let QueryResults::Solutions(solutions) store.query(LEAF_CHECK)? { for sol in solutions { let sol sol?; if let Some(term) sol.get(leaf) { leaves.push(term.to_string()); } } } Ok(leaves) }预期输出叶子节点是skill:cargo-setup和skill:hash-algorithms。如果编译结果里这两个节点没有入边说明闭包抽取正确。4.3 用模型对话验证 DAG 语义编译出的 DAG 可以序列化成 JSON丢给模型做语义审查。比如让模型判断“这张 DAG 是否覆盖了 JWT 重构所需的全部技能”。调用走 TaoToken 的模型对话入口https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。let client reqwest::Client::new(); let resp client .post(format!({}/v1/chat/completions, std::env::var(TAOTOKEN_BASE_URL)?)) .bearer_auth(std::env::var(TAOTOKEN_API_KEY)?) .json(serde_json::json!({ model: claude-sonnet-4-5, messages: [{role: user, content: format!(审查这张 DAG 的完整性: {}, serde_json::to_string(dag)?)}] })) .send() .await?;这一步不是必须的但能把“图结构正确”和“语义正确”分开验证。图结构用 SPARQL 保证语义用模型兜底。5. 本篇常见错排查5.1 SPARQL 闭包查不到传递依赖最常见的原因是context里requires的type没写id导致 JSON-LD 把对象当成字面量而不是 IRI。检查你的 contextrequires: { id: skill:requires, type: id }少了type: idskill:requires的值会被解析成字符串SPARQL 里就匹配不上。5.2 拓扑排序报环如果toposort返回 Err说明依赖图里有环。用下面这条查询定位环上的节点PREFIX skill: https://example.org/skill# SELECT ?a ?b WHERE { ?a skill:requires ?b . ?b skill:requires ?a . }双向依赖在技能图谱里通常是数据录入错误改掉其中一条边即可。5.3 oxigraph 加载 JSON-LD 报 profile 错误oxigraph 0.4 的RdfFormat::JsonLd需要指定 profile。如果你用的是旧版本改成RdfFormat::JsonLd { profile: Default::default() }。报错信息通常是unexpected token实际是格式枚举不匹配。5.4 编译出的 DAG 节点数比预期少检查OPTIONAL子句。如果只写?skill skill:requires* ?dep叶子节点没有 requires 的不会被?dep绑定导致丢失。加上OPTIONAL { ?dep skill:requires ?transitive . }或者改用UNION把叶子补回来。5.5 模型审查返回超时DAG 序列化后可能很长超过模型上下文。先做摘要只传节点 IRI 列表和边数量别传完整 JSON。或者用 Coding Plan 的长上下文模型https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。6. 把编译链路接进你的 Agent 平台到这里JSON-LD 到 DAG 的编译链路已经能跑通了。回顾一下关键点用skill:requires*抽传递闭包用 petgraph 做拓扑排序和环检测用 SPARQL 校验叶子节点和孤立节点。这套流程把“计划”从自然语言变成了可计算的数据结构调度器拿到的不再是一段文字而是一张带类型和边的图。下一步你可以做的把EdgeKind::Parallel接进调度器让 SA 按并行边决定并发把 DAG 序列化后存进黑板让 CA 按节点校验上游需求变更时只重查受影响的子图局部重建 DAG。接入相关的文档在这里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。如果你用 Claude Code 做开发Anthropic 兼容入口在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-code-anthropicutm_campaignrewrite 可以直接把编译器的单测生成接进去。最后留一个我踩过的坑别在编译阶段就把assigned_to填死。调度是运行时的事编译只负责结构正确。把分配逻辑留给 SADAG 保持纯粹后面改起来才不痛苦。