墨台开发文档

七、完整实战Demo

Demo A: 社交网络 - 寻找共同好友

GraphStructure { (me:Person)-[r1:knows]->(friend:Person), (you:Person)-[r2:knows]->(friend) } Rule { R1: me.id == 'user_001' R2: you.id == 'user_002' R3: me.id != you.id } Action { get(friend.id as mutual_friend_id, friend.name as mutual_friend_name) }

Demo B: 金融风控 - 可疑交易检测

GraphStructure { (s:Account)-[t:transfer]->(target:Account) } Rule { R1('大额交易'): t.amount > 100000 R2('夜间交易'): t.hour >= 0 && t.hour <= 6 R3('频繁交易'): group(s, target).count(t) > 10

  risk_score = rule_value(R1 && R2, 100,
               rule_value(R1, 60,
               rule_value(R3, 40, 0)))

  R4('高风险'): risk_score >= 60

} Action { get(s.id as from_account, target.id as to_account, t.amount, risk_score as risk_level) }

Demo C: 电影推荐 - 导演编剧协作分析

GraphStructure { (film:Film)-[e1:directedBy]->(director:Director), (film)-[e2:writtenBy]->(writer:Writer), (director)-[e3:collaboratedWith]->(writer) } Rule { R1('80后导演'): director.birthYear > 1980 R2('同性别组合'): director.gender == writer.gender

  collab_count = group(director, writer).count(film)
  R3('多次合作'): collab_count >= 3

} Action { get(director.name as director_name, writer.name as writer_name, collab_count as collaboration_times) }

Demo D: 企业图谱 - 股权穿透

GraphStructure { A [Company, start='true'] B [Company] A -> B [holdShare] repeat(1, 5) as path } Rule { // 保留最短路径 R: group(A, B).keep_shortest_path(path)

  // 计算穿透后的持股比例
  final_ratio = {path}.reduce((pre, cur) => pre * cur.ratio, 1.0)

  R1('实际控制'): final_ratio > 0.5

} Action { get(A.id as parent_company, B.id as subsidiary, final_ratio as control_ratio) }

Demo E: 地理位置查询 - 附近地铁站

GraphStructure { // 使用linked edge进行空间查询 (park:Park)-[e:nearby(park.boundary, subway.center, 1000)]->(subway:Subway) } Rule { R1: park.city == '北京' } Action { get(park.name as park_name, subway.name as subway_station, subway.line as subway_line) }

八、语法速查表

语法元素	示例
节点定义	(a:Person) 或 A [Person]
边定义	(a)-[r:knows]->(b)
反向边	(a)<-[r:knows]-(b)
双向边	(a)<->[r:knows]-(b)
变长路径	repeat(1, 10)
节点约束	(a:Person WHERE age > 18)
边约束	-[r:knows WHERE year > 2020]->
逻辑规则	R1('说明'): 表达式
赋值规则	变量 = 表达式
属性赋值	s.newProp = 表达式
分组聚合	group(a, b).sum(c)
条件值	rule_value(条件, 真值, 假值)
参数引用	$paramName

这些示例覆盖了KGDSL的主要语法特性，可以直接用于OpenSPG的知识图谱查询和推理规则定义。

定位对比

维度	Schema DSL	KGDSL
核心职责	定义知识图谱的静态结构	执行动态查询和推理
定义内容	实体类型、属性、关系、约束、索引	图模式匹配、规则表达式、推理逻辑
执行时机	系统初始化/Schema变更时	运行时查询执行
生命周期	持久化存储（永久有效）	临时执行（即时处理）
修改频率	低（业务结构变化时）	高（频繁查询）

KGDSL 风险规则执行完整流程详解

我来通过一个金融风控场景的具体例子，完整讲解KGDSL从定义到执行的全过程。

场景：可疑转账风险检测

GraphStructure { (user:Account)-[t:transfer]->(target:Account) } Rule { R1('大额交易'): t.amount > 100000 R2('夜间交易'): t.hour >= 0 && t.hour <= 6 R3('高风险'): R1 && R2 } Action { get(user.id, target.id, t.amount) }

一、完整执行流程图

┌─────────────────────────────────────────────────────────────────────────────┐ │ KGDSL 风险规则执行流程 │ └─────────────────────────────────────────────────────────────────────────────┘

KGDSL文本
    │
    ▼

┌──────────────────────────────────────────────────────────────────────────────┐ │ 阶段1: 词法语法解析 (ANTLR4) │ │ ─────────────────────────────────────────────────────────────────────────── │ │ 入口: OpenSPGDslParser.parseMultipleStatement() │ │ 文件: reasoner/kgdsl-parser/.../OpenSPGDslParser.scala │ │ │ │ [KGDSL文本] ──ANTLR Lexer──> [Token流] ──ANTLR Parser──> [ParseTree] │ │ │ │ 输出: Block (未解析的逻辑计划) │ │ ├── SourceBlock(graph) │ │ ├── MatchBlock(patterns) ← GraphStructure解析结果 │ │ ├── FilterBlock(rules) ← Rule解析结果 │ │ └── TableResultBlock ← Action解析结果 │ └──────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────────────────┐ │ 阶段2: 逻辑计划生成 (LogicalPlanner) │ │ ─────────────────────────────────────────────────────────────────────────── │ │ 入口: LogicalPlanner.plan(dag) │ │ 文件: reasoner/lube-logical/.../LogicalPlanner.scala │ │ │ │ Block树 ──resolve()──> SolvedModel ──planBlock()──> LogicalOperator树 │ │ │ │ 输出: LogicalOperator DAG │ │ Start(Account) │ │ └─> PatternScan(transfer pattern) │ │ └─> Filter(R1 && R2) │ │ └─> Select(user.id, target.id, t.amount) │ └──────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────────────────┐ │ 阶段3: 逻辑优化 (LogicalOptimizer) │ │ ─────────────────────────────────────────────────────────────────────────── │ │ 入口: LogicalOptimizer.optimize() │ │ 文件: reasoner/lube-logical/.../LogicalOptimizer.scala │ │ │ │ 优化规则: │ │ ├── FilterPushDown - 谓词下推到PatternScan │ │ ├── ProjectMerge - 合并相邻投影 │ │ ├── FilterMerge - 合并相邻过滤 │ │ └── EdgeToProperty - 边转属性优化 │ │ │ │ 优化后: 规则条件被下推到图扫描阶段 │ └──────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────────────────┐ │ 阶段4: 物理计划生成 (PhysicalPlanner) │ │ ─────────────────────────────────────────────────────────────────────────── │ │ 入口: PhysicalPlanner.plan[T](logicalOp) │ │ 文件: reasoner/lube-physical/.../PhysicalPlanner.scala │ │ │ │ LogicalOperator ──模式匹配──> PhysicalOperator │ │ │ │ 转换规则: │ │ LogicalOperators.Start → PhysicalOperators.Start │ │ LogicalOperators.PatternScan → PhysicalOperators.PatternScan │ │ LogicalOperators.Filter → PhysicalOperators.Filter │ │ LogicalOperators.Select → PhysicalOperators.Select │ │ │ │ 输出: PhysicalOperator[LocalRDG] DAG │ └──────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────────────────┐ │ 阶段5: 火山模型执行 (LocalRDG) │ │ ─────────────────────────────────────────────────────────────────────────── │ │ 入口: physicalPlan.rdg (惰性求值) │ │ 文件: reasoner/runner/local-runner/.../LocalRDG.java │ │ │ │ 执行流程 (自顶向下拉取，自底向上返回): │ │ │ │ Select.rdg │ │ ↑ 拉取 │ │ Filter.rdg.filter(rule) ← 执行规则过滤 │ │ ↑ 拉取 │ │ PatternScan.rdg.patternScan() ← 执行图模式匹配 │ │ ↑ 拉取 │ │ Start.rdg ← 创建初始RDG，获取起始节点 │ │ ↑ │ │ GraphState (Neo4j/TuGraph) ← 底层图存储 │ └──────────────────────────────────────────────────────────────────────────────┘ │ ▼ Result (匹配的风险交易记录)

二、各阶段详细代码路径

阶段1: 词法语法解析

入口代码 (KGReasonerSession.scala:369):

def plan2UnresolvedLogicalPlan(query: String, params: Map[String, Object]): List[Block] = { // 调用Parser解析KGDSL val blocks = planToBlock(query, params) registerSchema(blocks) blocks }

private def planToBlock(query: String, param: Map[String, Object]): List[Block] = { // 核心：调用ANTLR解析器 val blocks = parser.parseMultipleStatement(query, param) // 提取参数 variableParameters = parser.getAllParameters() blocks }

Parser实现 (OpenSPGDslParser.scala): class OpenSPGDslParser extends ParserInterface {

def parseMultipleStatement(text: String, param: Map[String, Object]): List[Block] = { // 1. ANTLR词法分析 val lexer = new KGDSLLexer(CharStreams.fromString(text)) val tokens = new CommonTokenStream(lexer)

 // 2. ANTLR语法分析
 val parser = new KGDSLParser(tokens)
 val tree = parser.script()  // 获取语法树

 // 3. 遍历语法树，生成Block
 parseKgDsl(tree.kg_dsl(), param)

} }

解析结果示例: Block结构: TableResultBlock(binds=[user.id, target.id, t.amount]) └── FilterBlock(rule=R3: R1 && R2) └── MatchBlock(patterns={ path1: GraphPath( graphPattern=GraphPattern( root=user, nodes={user: PatternElement(Account), target: PatternElement(Account)}, edges={user: [PatternConnection(alias=t, relTypes={transfer}, target=target)]} ) ) }) └── SourceBlock(graph=KG)

阶段2: 逻辑计划生成

入口代码 (LogicalPlanner.scala:60):

def plan(input: Block)(implicit context: LogicalPlannerContext): List[LogicalOperator] = { // 1. 解析类型信息 val source = resolve(input)

// 2. 递归规划Block树 val planWithoutResult = planBlock(input.dependencies.head, input, None, source)

// 3. 根据结果类型生成最终算子 val plan = input match { case t: TableResultBlock => Select(planWithoutResult, t.binds.fields, t.asList, t.distinct) case d: DDLBlock => DDL(planWithoutResult, d.ddlOp) } List(plan) }

Block到LogicalOperator的转换 (LogicalPlanner.scala:228):

private def planNonLeaf(block: Block, ...): LogicalOperator = { block match { case MatchBlock(_, matches) => // 图模式匹配 → PatternScan/ExpandInto planMatchPattern(solvedPattern(matches.head._2, solvedModel), plan)

 case FilterBlock(_, rule) =>
   // 规则过滤 → Filter算子
   planFilter(rule, plan)

 case ProjectBlock(_, projects) =>
   // 属性投影 → Project算子
   planProject(projects, root, plan)

 case AggregationBlock(_, aggregations, group) =>
   // 聚合 → Aggregate算子
   planAggregate(aggregations, group, plan)

} }

生成的逻辑计划: LogicalOperator DAG: Select[user.id, target.id, t.amount] └── Filter(rule: R1 && R2) └── PatternScan(pattern: user-[t:transfer]->target) └── Start(graph: KG, types: {Account})

阶段3: 物理计划生成

入口代码 (PhysicalPlanner.scala:48):

def plan[T <: RDG[T]: TypeTag](input: LogicalOperator)(implicit context: PhysicalPlannerContext[T]): PhysicalOperator[T] = {

input match { case start: LogicalOperators.Start => Start(start.graph.graphName, start.alias, start.fields, start.types)

 case LogicalOperators.PatternScan(in, pattern) =>
   PatternScan(plan[T](in), pattern, input.fields)

 case LogicalOperators.Filter(in, expr) =>
   Filter(plan[T](in), expr)

 case LogicalOperators.Select(in, orderedFields, as, distinct) =>
   Select(plan[T](in), orderedFields, as, distinct)

} }

生成的物理计划: PhysicalOperator[LocalRDG] DAG: Select[LocalRDG] └── Filter[LocalRDG](rule: R1 && R2) └── PatternScan[LocalRDG](pattern)└── Start[LocalRDG](Account)

阶段4: 火山模型执行 (关键!)

执行入口 (KGReasonerSession.scala:110):

def getResult(physicalPlan: PhysicalOperator[T]): Result = { physicalPlan match { case select: Select[T] => select.row // 触发惰性求值，自顶向下拉取 case _ => physicalPlan.rdg } }

火山模型核心 - 惰性求值链:

每个PhysicalOperator都有一个rdg属性，调用时触发计算：

// PhysicalOperators.PatternScan class PatternScan[T <: RDG[T]] { lazy val rdg: T = { children.head.rdg.patternScan(pattern) // 调用子节点的rdg，然后执行patternScan } }

// PhysicalOperators.Filter class Filter[T <: RDG[T]] { lazy val rdg: T = { children.head.rdg.filter(rule) // 调用子节点的rdg，然后执行filter } }

阶段5: LocalRDG执行规则匹配

PatternScan - 图模式扫描 (LocalRDG.java:289):

public LocalRDG patternScan(Pattern pattern) { // 并行执行模式匹配 for (KgGraph<IVertexId> kgGraphId : this.kgGraphList) { IVertexId id = kgGraphId.getVertex(this.startVertexAlias).get(0).getId();

     Callable<KgGraph<IVertexId>> patternScanCallable = () -> {
         // 核心：调用PatternMatcher执行图模式匹配
         return patternMatcher.patternMatch(
             id,                    // 起始节点ID
             pattern,               // 图模式 (user)-[t:transfer]->(target)
             rootVertexRuleList,    // 节点规则
             edgeRuleMap,           // 边规则
             maxPathLimit           // 路径限制
         );
     };

     futureList.add(threadPoolExecutor.submit(patternScanCallable));
 }

 // 收集匹配结果
 for (Future<KgGraph<IVertexId>> future : futureList) {
     KgGraph<IVertexId> kgGraph = future.get(this.executorTimeoutMs, TimeUnit.MILLISECONDS);
     if (kgGraph != null) {
         newKgGraphList.add(kgGraph);  // 匹配成功的子图
     }
 }

 this.kgGraphList = newKgGraphList;
 return this;

}

Filter - 规则过滤 (风险判断核心!) (LocalRDG.java:705):

public LocalRDG filter(Rule rule) { // 获取规则使用的顶点和边 Tuple2<Set<String>, Set<Connection>> tuple2 = RunnerUtil.getRuleUseVertexAndEdgeSet(rule, this.kgGraphSchema);

 // 获取规则表达式字符串 (如 "t.amount > 100000 && t.hour >= 0 && t.hour <= 6")
 List<String> exprStringSet = WareHouseUtils.getRuleList(rule);

 List<KgGraph<IVertexId>> newKgGraphList = new ArrayList<>();

 // 对每个匹配的子图执行规则过滤
 for (KgGraph<IVertexId> kgGraph : this.kgGraphList) {
     // 核心：执行规则表达式求值
     List<KgGraph<IVertexId>> resultList = RunnerUtil.filterKgGraph(
         kgGraph,           // 当前子图
         vertexSet,         // 涉及的顶点
         edgeSet,           // 涉及的边
         this.kgGraphSchema,
         exprStringSet,     // 规则表达式
         this.maxPathLimit,
         this.taskId
     );

     // 只保留满足规则的子图
     newKgGraphList.addAll(resultList);
 }

 // 更新结果集
 this.kgGraphList = newKgGraphList;

 // 记录执行日志 (哪些ID通过/失败)
 this.executionRecorder.stageResultWithDetail(
     "filter(" + rule.getName() + ")",
     this.kgGraphList.size(),
     generateRuntimeLog(failedStartIdSet, afterStartId),
     Lists.newArrayList(rule)
 );

 return this;

}

三、规则表达式求值过程

规则求值核心 (RunnerUtil.filterKgGraph):

public static List<KgGraph<IVertexId>> filterKgGraph( KgGraph<IVertexId> kgGraph, Set<String> vertexSet, Set<String> edgeSet, PartialGraphPattern schema, List<String> exprStringSet, Long maxPathLimit, String taskId) {

 // 1. 将KgGraph转换为上下文Map
 Map<String, Object> context = kgGraph2Context(schema, kgGraph);
 // context = {
 //   "user": {"id": "acc_001", "balance": 50000, ...},
 //   "target": {"id": "acc_002", ...},
 //   "t": {"amount": 150000, "hour": 3, ...}
 // }

 // 2. 使用QLExpress执行规则表达式
 for (String expr : exprStringSet) {
     // 例如: expr = "t.amount > 100000 && t.hour >= 0 && t.hour <= 6"
     Object result = expressRunner.execute(expr, context);

     if (result instanceof Boolean && !(Boolean) result) {
         return Collections.emptyList();  // 规则不满足，过滤掉
     }
 }

 // 3. 所有规则都满足，返回匹配的子图
 return Collections.singletonList(kgGraph);

}

四、完整执行流程示例

假设图中有以下数据：

节点: Account(id=acc_001) Account(id=acc_002) Account(id=acc_003)

边: acc_001 -[transfer{amount:150000, hour:3}]-> acc_002 ✓ 高风险 acc_001 -[transfer{amount:50000, hour:14}]-> acc_003 ✗ 金额不足 acc_002 -[transfer{amount:200000, hour:10}]-> acc_003 ✗ 非夜间

执行过程:

Start: 获取所有Account节点 → [acc_001, acc_002, acc_003]
PatternScan: 对每个节点查找 transfer 边 acc_001 → 匹配到2条边 → 2个KgGraph acc_002 → 匹配到1条边 → 1个KgGraph acc_003 → 无出边 → 0个KgGraph 结果: 3个KgGraph
Filter(R1: amount > 100000): KgGraph1(acc_001→acc_002, amount=150000) → ✓ 通过 KgGraph2(acc_001→acc_003, amount=50000) → ✗ 过滤 KgGraph3(acc_002→acc_003, amount=200000) → ✓ 通过结果: 2个KgGraph
Filter(R2: hour >= 0 && hour <= 6): KgGraph1(acc_001→acc_002, hour=3) → ✓ 通过 KgGraph3(acc_002→acc_003, hour=10) → ✗ 过滤结果: 1个KgGraph
Filter(R3: R1 && R2): KgGraph1 → ✓ (已经满足R1和R2) 结果: 1个KgGraph
Select: 提取 user.id, target.id, t.amount 输出: [("acc_001", "acc_002", 150000)]

五、关键代码文件路径

阶段	文件	功能
ANTLR语法	reasoner/kgdsl-parser/.../KGDSL.g4	KGDSL语法定义
DSL解析器	reasoner/kgdsl-parser/.../OpenSPGDslParser.scala	解析KGDSL生成Block
会话入口	reasoner/runner/runner-common/.../KGReasonerSession.scala:101	执行入口 getResult()
逻辑规划	reasoner/lube-logical/.../LogicalPlanner.scala:60	Block → LogicalOperator
逻辑优化	reasoner/lube-logical/.../LogicalOptimizer.scala	谓词下推等优化
物理规划	reasoner/lube-physical/.../PhysicalPlanner.scala:48	LogicalOp → PhysicalOp
本地执行	reasoner/runner/local-runner/.../LocalRDG.java:289	patternScan执行
规则过滤	reasoner/runner/local-runner/.../LocalRDG.java:705	filter规则执行
规则求值	reasoner/runner/runner-common/.../RunnerUtil.java	表达式求值
模式匹配	reasoner/udf/src/.../PatternMatcher.java	图模式匹配引擎

六、总结

KGDSL风险规则的执行流程：

解析阶段: KGDSL → Block (未解析逻辑计划)
规划阶段: Block → LogicalOperator → PhysicalOperator
执行阶段: 火山模型惰性求值，自顶向下拉取
匹配阶段: PatternMatcher在图存储中执行模式匹配
过滤阶段: 对匹配的子图执行规则表达式求值
输出阶段: 返回满足所有规则的风险记录

核心优势：

声明式规则: 业务人员可以直接编写规则，无需写代码
规则下推: 优化器将规则尽可能下推到图扫描阶段
并行执行: ThreadPoolExecutor并行处理多个起始节点
火山模型: 惰性求值，按需计算，内存友好