OpenSPG 核心模块深度解析

1. 概述

OpenSPG由三大核心模块组成，构成了完整的知识图谱生命周期管理体系：

┌─────────────────────────────────────────────────────────────┐
│                     OpenSPG 核心架构                         │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────────┐   ┌──────────────┐   ┌──────────────┐   │
│  │              │   │              │   │              │   │
│  │ SPG-Schema   │──▶│ SPG-Builder  │──▶│ SPG-Reasoner │   │
│  │ 语义建模      │   │ 知识构建      │   │ 逻辑推理      │   │
│  │              │   │              │   │              │   │
│  └──────────────┘   └──────────────┘   └──────────────┘   │
│         │                   │                    │         │
│         ▼                   ▼                    ▼         │
│    Schema定义           ETL Pipeline          KGDSL查询    │
│    类型系统             数据转换              推理引擎       │
│                                                              │
└─────────────────────────────────────────────────────────────┘

1.1 模块职责

模块	核心职责	输入	输出
SPG-Schema	定义知识图谱的语义模型	Schema DSL / Java API	ProjectSchema对象
SPG-Builder	从数据源构建知识图谱	原始数据 + Schema	图谱实例数据
SPG-Reasoner	基于规则的知识推理	KGDSL查询 + 图谱数据	推理结果

1.2 技术栈

模块	主要语言	框架/库
SPG-Schema	Java	Spring Boot, MyBatis
SPG-Builder	Java, Python	Pipeline框架, Pemja
SPG-Reasoner	Scala, Java	ANTLR4, Cats, Akka

2. SPG-Schema 语义建模

2.1 模块概述

SPG-Schema是知识图谱的元数据定义层，负责定义实体类型、属性、关系等语义结构。

核心包路径

server/
├── core/schema/model/          # Schema数据模型
│   └── src/main/java/com/antgroup/openspg/core/schema/model/
│       ├── type/               # 类型定义
│       │   ├── BaseSPGType.java
│       │   ├── EntityType.java
│       │   ├── ConceptType.java
│       │   ├── EventType.java
│       │   ├── StandardType.java
│       │   └── ProjectSchema.java
│       ├── predicate/          # 谓词定义
│       │   ├── Property.java
│       │   └── Relation.java
│       ├── semantic/           # 语义定义
│       │   ├── LogicalRule.java
│       │   └── PredicateSemantic.java
│       └── identifier/         # 标识符
│           └── SPGTypeIdentifier.java
│
└── biz/schema/                 # Schema业务逻辑
    └── src/main/java/com/antgroup/openspg/server/biz/schema/
        ├── SchemaManager.java           # Schema管理接口
        └── impl/SchemaManagerImpl.java  # Schema管理实现

2.2 核心类详解

2.2.1 BaseSPGType - SPG类型基类

位置: core/schema/model/src/main/java/com/antgroup/openspg/core/schema/model/type/BaseSPGType.java

java

/**
 * SPG类型的抽象基类
 *
 * SPG类型体系：
 * - EntityType: 实体类型（如User、Company）
 * - ConceptType: 概念类型（如品牌、行政区）
 * - EventType: 事件类型（如疫情事件、恐怖事件）
 * - BasicType: 基础类型（如String、Integer）
 * - StandardType: 标准类型（如PhoneNumber、Email）
 */
public abstract class BaseSPGType extends BaseOntology
    implements WithBasicInfo<SPGTypeIdentifier>,
               WithSPGTypeEnum,
               WithSpgTypeAdvancedConfig,
               WithParentTypeInfo {

    /** 基本信息：名称、中文名、描述等 */
    private final BasicInfo<SPGTypeIdentifier> basicInfo;

    /** 父类型信息 */
    private ParentTypeInfo parentTypeInfo;

    /** 类型类别 */
    private final SPGTypeEnum spgTypeEnum;

    /** 属性列表（包含继承的属性） */
    private List<Property> properties;

    /** 关系列表（包含继承的关系） */
    private List<Relation> relations;

    /** 高级配置：可见性、算子绑定等 */
    private SPGTypeAdvancedConfig advancedConfig;

    // 构造函数和方法...
}

关键方法：

java

// 获取类型标识符
public SPGTypeIdentifier getBaseSpgIdentifier()

// 获取所有属性（包括继承）
public List<Property> getProperties()

// 获取所有关系（包括继承）
public List<Relation> getRelations()

// 根据名称获取属性
public Property getPropertyByName(String propertyName)

// 根据名称获取关系
public Relation getRelationByName(SPGTripleIdentifier relationIdentifier)

// 转换为引用类型
public SPGTypeRef toRef()

2.2.2 EntityType - 实体类型

位置: core/schema/model/src/main/java/com/antgroup/openspg/core/schema/model/type/EntityType.java

java

/**
 * 实体类型定义
 *
 * 实体是知识图谱中的核心节点，代表现实世界中的对象
 * 如：人物、公司、产品等
 */
public class EntityType extends BaseAdvancedType {

    public EntityType(
        BasicInfo<SPGTypeIdentifier> basicInfo,
        ParentTypeInfo parentTypeInfo,
        List<Property> properties,
        List<Relation> relations,
        SPGTypeAdvancedConfig advancedConfig
    ) {
        super(basicInfo, parentTypeInfo, SPGTypeEnum.ENTITY_TYPE,
              properties, relations, advancedConfig);
    }

    @Override
    public SPGTypeEnum getSpgTypeEnum() {
        return SPGTypeEnum.ENTITY_TYPE;
    }
}

使用示例：

java

// 创建Person实体类型
EntityType personType = new EntityType(
    // 基本信息
    new BasicInfo<>(
        new SPGTypeIdentifier("default", "Person"),  // namespace.name
        "人物",                                        // 中文名
        "表示自然人"                                   // 描述
    ),

    // 父类型信息
    ParentTypeInfo.THING,  // 继承自Thing根类型

    // 属性列表
    Arrays.asList(
        new Property(
            new BasicInfo<>(new PredicateIdentifier("name"), "姓名", ""),
            null,
            new SPGTypeRef(BasicType.TEXT),  // 属性类型为Text
            false,  // 非多值属性
            new PropertyAdvancedConfig().setIndexType(IndexTypeEnum.TEXT)
        ),
        new Property(
            new BasicInfo<>(new PredicateIdentifier("age"), "年龄", ""),
            null,
            new SPGTypeRef(BasicType.INTEGER),
            false,
            new PropertyAdvancedConfig()
        )
    ),

    // 关系列表
    Arrays.asList(
        new Relation(
            new BasicInfo<>(new PredicateIdentifier("worksAt"), "工作于", ""),
            personType.toRef(),      // 主体类型
            companyType.toRef(),     // 客体类型
            false,
            new RelationAdvancedConfig()
        )
    ),

    // 高级配置
    new SPGTypeAdvancedConfig()
);

2.2.3 Property - 属性定义

位置: core/schema/model/src/main/java/com/antgroup/openspg/core/schema/model/predicate/Property.java

java

/**
 * 属性定义
 *
 * 属性描述实体的特征，如人物的姓名、年龄等
 */
public class Property extends BasePredicate implements WithPropertyAdvancedConfig {

    /** 属性值类型 */
    private final SPGTypeRef objectTypeRef;

    /** 是否多值属性 */
    private final Boolean multiValue;

    /** 高级配置：索引类型、约束等 */
    private PropertyAdvancedConfig advancedConfig;

    // 子属性列表（用于结构化属性）
    private List<SubProperty> subProperties;

    // 语义关系列表
    private List<PredicateSemantic> semantics;
}

索引类型：

java

public enum IndexTypeEnum {
    EXACT,          // 精确匹配索引
    TEXT,           // 全文检索索引
    VECTOR,         // 向量索引
    TEXT_AND_VECTOR,// 文本+向量混合索引
    RANGE,          // 范围索引（数值、日期）
    NONE            // 无索引
}

2.2.4 Relation - 关系定义

位置: core/schema/model/src/main/java/com/antgroup/openspg/core/schema/model/predicate/Relation.java

java

/**
 * 关系定义
 *
 * 关系描述实体之间的连接，如人物-工作于-公司
 */
public class Relation extends BasePredicate {

    /** 客体类型引用 */
    private final SPGTypeRef objectTypeRef;

    /** 是否多值关系 */
    private final Boolean multiValue;

    /** 关系高级配置 */
    private RelationAdvancedConfig advancedConfig;
}

2.2.5 ProjectSchema - 项目Schema

位置: core/schema/model/src/main/java/com/antgroup/openspg/core/schema/model/type/ProjectSchema.java

java

/**
 * 项目Schema
 *
 * 包含项目中定义的所有SPG类型
 */
public class ProjectSchema extends BaseToString {

    /** SPG类型列表 */
    private final List<BaseSPGType> spgTypes;

    /** SPG类型映射（name -> type） */
    private transient volatile Map<SPGTypeIdentifier, BaseSPGType> spgTypeMap;

    /**
     * 根据类型名称获取SPG类型
     */
    public BaseSPGType getByName(SPGTypeIdentifier name) {
        if (spgTypeMap == null) {
            spgTypeMap = spgTypes.stream()
                .collect(Collectors.toMap(
                    WithBasicInfo::getBaseSpgIdentifier,
                    x -> x
                ));
        }
        return spgTypeMap.get(name);
    }

    /**
     * 根据类型名称字符串获取SPG类型
     */
    public BaseSPGType getByName(String name) {
        return getByName(SPGTypeIdentifier.parse(name));
    }
}

2.3 SchemaManager - Schema管理服务

接口位置: server/biz/schema/src/main/java/com/antgroup/openspg/server/biz/schema/SchemaManager.java

java

/**
 * Schema管理接口
 *
 * 提供Schema的CRUD操作
 */
public interface SchemaManager {

    /**
     * 修改Schema
     *
     * @param request Schema修改请求
     */
    void alterSchema(SchemaAlterRequest request);

    /**
     * 获取项目Schema
     *
     * @param projectId 项目ID
     * @return 项目Schema对象
     */
    ProjectSchema getProjectSchema(Long projectId);

    /**
     * 根据唯一名称获取SPG类型
     *
     * @param uniqueName 唯一名称（格式：namespace.typeName）
     * @return SPG类型对象
     */
    BaseSPGType getSpgType(String uniqueName);

    /**
     * 获取内置属性
     *
     * @param spgTypeEnum SPG类型枚举
     * @return 内置属性列表
     */
    List<Property> getBuiltInProperty(SPGTypeEnum spgTypeEnum);
}

实现位置: server/biz/schema/src/main/java/com/antgroup/openspg/server/biz/schema/impl/SchemaManagerImpl.java

2.4 Schema操作示例

2.4.1 创建Schema

java

// 1. 创建EntityType
EntityType companyType = new EntityType(
    new BasicInfo<>(
        new SPGTypeIdentifier("enterprise", "Company"),
        "公司",
        "企业组织"
    ),
    ParentTypeInfo.THING,
    Arrays.asList(
        createProperty("name", "公司名称", BasicType.TEXT, IndexTypeEnum.TEXT),
        createProperty("foundYear", "成立年份", BasicType.TEXT, IndexTypeEnum.EXACT),
        createProperty("industry", "所属行业", BasicType.TEXT, IndexTypeEnum.TEXT)
    ),
    null,
    new SPGTypeAdvancedConfig()
);

EntityType personType = new EntityType(
    new BasicInfo<>(
        new SPGTypeIdentifier("enterprise", "Person"),
        "人物",
        "自然人"
    ),
    ParentTypeInfo.THING,
    Arrays.asList(
        createProperty("name", "姓名", BasicType.TEXT, IndexTypeEnum.TEXT),
        createProperty("age", "年龄", BasicType.INTEGER, IndexTypeEnum.RANGE)
    ),
    Arrays.asList(
        new Relation(
            new BasicInfo<>(new PredicateIdentifier("worksAt"), "工作于", ""),
            personType.toRef(),
            companyType.toRef(),
            false,
            new RelationAdvancedConfig()
        )
    ),
    new SPGTypeAdvancedConfig()
);

// 2. 设置AlterOperation
companyType.setAlterOperation(AlterOperationEnum.CREATE);
personType.setAlterOperation(AlterOperationEnum.CREATE);

// 3. 创建SchemaDraft
SchemaDraft draft = new SchemaDraft();
draft.setAlterSpgTypes(Arrays.asList(companyType, personType));

// 4. 提交Schema
SchemaAlterRequest request = new SchemaAlterRequest();
request.setProjectId(1L);
request.setSchemaDraft(draft);

schemaManager.alterSchema(request);

2.4.2 查询Schema

java

// 获取项目Schema
ProjectSchema projectSchema = schemaManager.getProjectSchema(1L);

// 获取指定类型
BaseSPGType personType = projectSchema.getByName("enterprise.Person");

// 获取类型的属性
List<Property> properties = personType.getProperties();

// 获取类型的关系
List<Relation> relations = personType.getRelations();

// 获取指定属性
Property nameProperty = personType.getPropertyByName("name");

2.4.3 更新Schema

java

// 1. 获取现有类型
BaseSPGType existingType = schemaManager.getSpgType("enterprise.Person");

// 2. 添加新属性
Property emailProperty = new Property(
    new BasicInfo<>(new PredicateIdentifier("email"), "邮箱", ""),
    null,
    new SPGTypeRef(BasicType.TEXT),
    false,
    new PropertyAdvancedConfig().setIndexType(IndexTypeEnum.TEXT)
);
emailProperty.setAlterOperation(AlterOperationEnum.CREATE);

List<Property> updatedProperties = new ArrayList<>(existingType.getProperties());
updatedProperties.add(emailProperty);

// 3. 创建更新的类型
EntityType updatedType = new EntityType(
    existingType.getBasicInfo(),
    existingType.getParentTypeInfo(),
    updatedProperties,
    existingType.getRelations(),
    existingType.getAdvancedConfig()
);
updatedType.setAlterOperation(AlterOperationEnum.UPDATE);

// 4. 提交更新
SchemaDraft draft = new SchemaDraft();
draft.setAlterSpgTypes(Arrays.asList(updatedType));

SchemaAlterRequest request = new SchemaAlterRequest();
request.setProjectId(1L);
request.setSchemaDraft(draft);

schemaManager.alterSchema(request);

2.5 Schema约束

2.5.1 约束类型

java

public enum ConstraintTypeEnum {
    NOT_NULL,      // 非空约束
    UNIQUE,        // 唯一约束
    REGULAR,       // 正则表达式约束
    ENUM,          // 枚举约束
    RANGE,         // 范围约束
    MULTI_VAL      // 多值约束
}

2.5.2 添加约束示例

java

// 非空约束
Property idProperty = createProperty("id", "ID", BasicType.TEXT, IndexTypeEnum.EXACT);
idProperty.setConstraint(new NotNullConstraint());

// 正则约束（邮箱格式）
Property emailProperty = createProperty("email", "邮箱", BasicType.TEXT, IndexTypeEnum.TEXT);
emailProperty.setConstraint(
    new RegularConstraint("^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$")
);

// 枚举约束
Property genderProperty = createProperty("gender", "性别", BasicType.TEXT, IndexTypeEnum.EXACT);
genderProperty.setConstraint(
    new EnumConstraint(Arrays.asList("男", "女", "未知"))
);

// 范围约束
Property ageProperty = createProperty("age", "年龄", BasicType.INTEGER, IndexTypeEnum.RANGE);
ageProperty.setConstraint(
    new RangeConstraint(0, 150)  // 年龄范围0-150
);

3. SPG-Builder 知识构建

3.1 模块概述

SPG-Builder负责从各种数据源（CSV、JSON、数据库等）抽取、转换、加载知识到图数据库。

核心包路径

builder/
├── core/                          # 核心执行引擎
│   └── src/main/java/com/antgroup/openspg/builder/core/
│       ├── physical/              # 物理执行节点
│       │   ├── BasePhysicalNode.java
│       │   ├── process/           # 处理器
│       │   │   ├── BaseProcessor.java
│       │   │   ├── SPGTypeMappingProcessor.java
│       │   │   ├── PythonProcessor.java
│       │   │   └── LLMNlExtractProcessor.java
│       │   └── PhysicalPlan.java
│       └── runtime/
│           └── BuilderContext.java
│
├── runner/local/                  # 本地执行器
│   └── src/main/java/com/antgroup/openspg/builder/runner/local/
│       ├── physical/
│       │   ├── source/            # 数据源
│       │   │   ├── BaseSourceReader.java
│       │   │   └── impl/
│       │   │       ├── CsvFileSourceReader.java
│       │   │       └── StringSourceReader.java
│       │   └── sink/              # 数据写入
│       │       ├── BaseSinkWriter.java
│       │       └── impl/
│       │           ├── GraphStoreSinkWriter.java
│       │           └── Neo4jSinkWriter.java
│       └── LocalBuilderRunner.java
│
└── model/                         # 数据模型
    └── src/main/java/com/antgroup/openspg/builder/model/
        ├── record/                # 记录模型
        │   ├── BaseRecord.java
        │   ├── BuilderRecord.java
        │   ├── BaseSPGRecord.java
        │   ├── EntityRecord.java
        │   ├── RelationRecord.java
        │   ├── ChunkRecord.java
        │   └── SubGraphRecord.java
        └── pipeline/              # Pipeline配置
            └── config/

3.2 核心类详解

3.2.1 BuilderContext - 构建上下文

位置: builder/core/src/main/java/com/antgroup/openspg/builder/core/runtime/BuilderContext.java

java

/**
 * Builder运行时上下文
 *
 * 包含执行过程中需要的所有配置和状态信息
 */
public class BuilderContext {

    /** 项目ID */
    private Long projectId;

    /** 任务ID */
    private Long taskId;

    /** 批量大小 */
    private Integer batchSize = 100;

    /** 并行度 */
    private Integer parallelism = 1;

    /** LLM并发执行数 */
    private Integer modelExecuteNum = 4;

    /** 图存储URL */
    private String graphStoreUrl;

    /** Python解释器路径 */
    private String pythonExec;

    /** Python库路径 */
    private String pythonPaths;

    /** Schema URL */
    private String schemaUrl;

    /** 项目配置（JSON格式） */
    private String project;

    /** 操作类型（INSERT/UPDATE/DELETE/UPSERT） */
    private RecordAlterOperationEnum operation;

    /** Schema目录 */
    private ProjectSchema catalog;

    // getter/setter...
}

3.2.2 BasePhysicalNode - 物理节点基类

位置: builder/core/src/main/java/com/antgroup/openspg/builder/core/physical/BasePhysicalNode.java

java

/**
 * 物理执行节点基类
 *
 * 三种物理节点：
 * 1. Source Node - 数据源节点
 * 2. Process Node - 处理节点
 * 3. Sink Node - 写入节点
 */
public abstract class BasePhysicalNode implements Comparable<BasePhysicalNode> {

    /** 节点ID */
    protected final String id;

    /** 节点名称 */
    protected final String name;

    /** 运行时上下文 */
    protected BuilderContext context;

    /** 是否已初始化 */
    private volatile boolean isInitialized = false;

    public BasePhysicalNode(String id, String name) {
        this.id = id;
        this.name = name;
    }

    /**
     * 初始化节点
     */
    public void init(BuilderContext context) throws BuilderException {
        this.context = context;
        if (!isInitialized) {
            synchronized (this) {
                if (!isInitialized) {
                    doInit(context);
                    isInitialized = true;
                }
            }
        }
    }

    /**
     * 子类实现具体初始化逻辑
     */
    public void doInit(BuilderContext context) throws BuilderException {}

    /**
     * 关闭资源
     */
    public abstract void close() throws Exception;
}

3.2.3 数据记录模型

BaseRecord - 记录基类

java

/**
 * 所有记录的基类
 */
public abstract class BaseRecord extends BaseValObj {}

BuilderRecord - 原始KV记录

java

/**
 * 构建器记录
 *
 * 用于表示从数据源读取的原始KV数据
 */
public class BuilderRecord extends BaseRecord {

    /** 记录ID */
    private final String recordId;

    /** SPG类型标识 */
    private SPGTypeIdentifier identifier;

    /** 属性键值对 */
    private final Map<String, String> props;

    public BuilderRecord(String recordId,
                        SPGTypeIdentifier identifier,
                        Map<String, String> props) {
        this.recordId = recordId;
        this.identifier = identifier;
        this.props = props;
    }

    /**
     * 获取属性值
     */
    public String getPropValue(String propKey) {
        return props.get(propKey);
    }

    /**
     * 创建新记录（使用新属性）
     */
    public BuilderRecord withNewProps(Map<String, String> newProps) {
        return new BuilderRecord(recordId, identifier, newProps);
    }
}

BaseSPGRecord - SPG记录基类

java

/**
 * SPG记录基类
 *
 * 表示映射到SPG Schema后的结构化记录
 */
public abstract class BaseSPGRecord extends BaseRecord {

    /** 实体/关系ID */
    private String id;

    /** 实体/关系名称 */
    private String name;

    /** SPG类型标识 */
    private SPGTypeIdentifier spgTypeIdentifier;

    /** 属性列表 */
    private List<BasePropertyRecord> properties = new ArrayList<>();

    /**
     * 获取记录类型
     */
    public abstract SPGRecordTypeEnum getRecordType();

    /**
     * 添加属性
     */
    public void addProperty(BasePropertyRecord property) {
        properties.add(property);
    }
}

EntityRecord - 实体记录

java

/**
 * 实体记录
 */
public class EntityRecord extends BaseSPGRecord {

    public EntityRecord(String id,
                       String name,
                       SPGTypeIdentifier spgTypeIdentifier) {
        super(id, name, spgTypeIdentifier);
    }

    @Override
    public SPGRecordTypeEnum getRecordType() {
        return SPGRecordTypeEnum.ENTITY;
    }
}

RelationRecord - 关系记录

java

/**
 * 关系记录
 */
public class RelationRecord extends BaseSPGRecord {

    /** 源实体ID */
    private String srcId;

    /** 目标实体ID */
    private String dstId;

    /** 关系名称 */
    private String relationName;

    public RelationRecord(String id,
                         String srcId,
                         String dstId,
                         String relationName,
                         SPGTypeIdentifier srcType,
                         SPGTypeIdentifier dstType) {
        super(id, relationName, srcType);
        this.srcId = srcId;
        this.dstId = dstId;
        this.relationName = relationName;
    }

    @Override
    public SPGRecordTypeEnum getRecordType() {
        return SPGRecordTypeEnum.RELATION;
    }
}

3.3 数据处理Pipeline

3.3.1 Pipeline架构

┌──────────┐   ┌──────────┐   ┌──────────┐   ┌──────────┐
│  Source  │──▶│ Processor│──▶│ Processor│──▶│   Sink   │
│  Reader  │   │    1     │   │    2     │   │  Writer  │
└──────────┘   └──────────┘   └──────────┘   └──────────┘
     │              │              │              │
     ▼              ▼              ▼              ▼
 读取数据        类型映射       Python处理       写入图库
BuilderRecord  BaseSPGRecord  Enhanced SPG   Neo4j/TuGraph

3.3.2 Source Reader - 数据源读取

CsvFileSourceReader - CSV文件读取器

位置: builder/runner/local/src/main/java/com/antgroup/openspg/builder/runner/local/physical/source/impl/CsvFileSourceReader.java

java

/**
 * CSV文件数据源读取器
 */
public class CsvFileSourceReader extends BaseSourceReader<CsvSourceNodeConfig> {

    private Queue<BaseRecord> queue;
    private CSVReader csvReader;
    private Iterator<String[]> lines;
    private AtomicLong lineNumber;

    public CsvFileSourceReader(String id, String name, CsvSourceNodeConfig config) {
        super(id, name, config);
    }

    @Override
    public void doInit(BuilderContext context) throws BuilderException {
        // 初始化队列
        queue = new ArrayBlockingQueue<>(
            context.getBatchSize() * context.getParallelism()
        );

        // 打开CSV文件
        csvReader = new CSVReaderBuilder(new FileReader(config.getUrl()))
            .withSkipLines(config.getStartRow() - 1)
            .build();

        lineNumber = new AtomicLong(config.getStartRow() - 1);
        lines = csvReader.iterator();
    }

    @Override
    public List<BaseRecord> read() {
        putQueue();  // 填充队列
        List<BaseRecord> results = new ArrayList<>(context.getBatchSize());

        // 批量读取
        for (int i = 0; i < context.getBatchSize(); i++) {
            BaseRecord record = queue.poll();
            if (record == null) {
                putQueue();
                record = queue.poll();
            }
            if (record == null) {
                break;
            }
            results.add(record);
        }
        return results;
    }

    /**
     * 解析CSV行为BuilderRecord
     */
    private BuilderRecord parse(String[] fields) {
        Map<String, String> props = new HashMap<>();
        for (int i = 0; i < config.getColumns().size(); i++) {
            String column = config.getColumns().get(i);
            if (i >= fields.length) {
                props.put(column, null);
            } else {
                props.put(column, fields[i]);
            }
        }
        return new BuilderRecord("line" + lineNumber.get(), null, props);
    }

    @Override
    public void close() throws Exception {
        if (csvReader != null) {
            csvReader.close();
        }
    }
}

使用示例：

java

// 创建CSV源配置
CsvSourceNodeConfig config = new CsvSourceNodeConfig(
    "/data/employees.csv",  // 文件路径
    1,                       // 起始行
    Arrays.asList("id", "name", "age", "department")  // 列名
);

// 创建读取器
CsvFileSourceReader reader = new CsvFileSourceReader(
    "csv_source_1",
    "Employee CSV Source",
    config
);

// 初始化
BuilderContext context = new BuilderContext();
context.setBatchSize(100);
reader.init(context);

// 读取数据
List<BaseRecord> records = reader.read();

// 关闭
reader.close();

3.3.3 Processor - 数据处理

SPGTypeMappingProcessor - 类型映射处理器

位置: builder/core/src/main/java/com/antgroup/openspg/builder/core/physical/process/SPGTypeMappingProcessor.java

java

/**
 * SPG类型映射处理器
 *
 * 将BuilderRecord（原始KV数据）映射为BaseSPGRecord（SPG类型实例）
 */
public class SPGTypeMappingProcessor extends BaseProcessor<SPGTypeMappingNodeConfigs> {

    private final List<SPGTypeMappingHelper> mappingHelpers;

    public SPGTypeMappingProcessor(String id,
                                   String name,
                                   SPGTypeMappingNodeConfigs config) {
        super(id, name, config);
        this.mappingHelpers = config.getMappingNodeConfigs().stream()
            .map(SPGTypeMappingHelper::new)
            .collect(Collectors.toList());
    }

    @Override
    public List<BaseRecord> process(List<BaseRecord> inputs) {
        List<BaseSPGRecord> resultSpgRecords = new ArrayList<>();

        // 将输入记录按SPG类型分组
        Map<SPGTypeIdentifier, List<BuilderRecord>> groupedRecords =
            groupByIdentifier(inputs);

        // 为每个映射助手处理记录
        for (SPGTypeMappingHelper helper : mappingHelpers) {
            List<BuilderRecord> records =
                groupedRecords.get(helper.getIdentifier());

            if (records != null) {
                for (BuilderRecord record : records) {
                    // 映射为SPG记录
                    List<BaseSPGRecord> spgRecords = helper.toSPGRecords(record, true);
                    resultSpgRecords.addAll(spgRecords);
                }
            }
        }

        return (List) resultSpgRecords;
    }
}

配置示例：

java

// 创建Person类型映射配置
SPGTypeMappingNodeConfig personMappingConfig = new SPGTypeMappingNodeConfig();
personMappingConfig.setSpgTypeName("Person");
personMappingConfig.setIdField("id");
personMappingConfig.setNameField("name");

// 属性映射
Map<String, String> propertyMappings = new HashMap<>();
propertyMappings.put("name", "name");        // SPG属性 -> CSV列
propertyMappings.put("age", "age");
propertyMappings.put("email", "email");
personMappingConfig.setPropertyMappings(propertyMappings);

// 创建映射配置集合
SPGTypeMappingNodeConfigs configs = new SPGTypeMappingNodeConfigs();
configs.setMappingNodeConfigs(Arrays.asList(personMappingConfig));

// 创建处理器
SPGTypeMappingProcessor processor = new SPGTypeMappingProcessor(
    "mapping_1",
    "Person Mapping",
    configs
);

// 使用
processor.init(context);
List<BaseRecord> spgRecords = processor.process(builderRecords);

3.3.4 Sink Writer - 数据写入

GraphStoreSinkWriter - 图存储写入器

位置: builder/runner/local/src/main/java/com/antgroup/openspg/builder/runner/local/physical/sink/impl/GraphStoreSinkWriter.java

java

/**
 * 图存储写入器
 *
 * 将SPGRecord写入图数据库
 */
public class GraphStoreSinkWriter extends BaseSinkWriter<GraphStoreSinkNodeConfig> {

    private GraphStoreClient graphStoreClient;
    private CheckProcessor checkProcessor;

    public GraphStoreSinkWriter(String id,
                               String name,
                               GraphStoreSinkNodeConfig config) {
        super(id, name, config);
    }

    @Override
    public void doInit(BuilderContext context) throws BuilderException {
        // 获取图存储客户端
        graphStoreClient = GraphStoreClientDriverManager.getClient(
            context.getGraphStoreUrl()
        );

        // 初始化校验处理器
        checkProcessor = new CheckProcessor();
        checkProcessor.init(context);
    }

    @Override
    public void write(List<BaseRecord> records) {
        if (!config.getIsWriter()) {
            return;  // 不写入模式
        }

        // 如果是UPSERT操作，先进行校验
        if (RecordAlterOperationEnum.UPSERT == context.getOperation()) {
            records = checkProcessor.process(records);
        }

        // 批量写入
        batchWriteToGraphStore(records);
    }

    /**
     * 批量写入图存储
     */
    private void batchWriteToGraphStore(List<BaseRecord> records) {
        // 转换为SPGRecordAlterItem
        List<SPGRecordAlterItem> items = records.stream()
            .map(record -> new SPGRecordAlterItem(
                context.getOperation(),
                (BaseSPGRecord) record
            ))
            .collect(Collectors.toList());

        // 调用图存储客户端写入
        graphStoreClient.manipulateRecord(
            new SPGRecordManipulateCmd(items)
        );
    }

    @Override
    public void close() throws Exception {
        // 清理资源
    }
}

3.4 完整构建示例

java

/**
 * 完整的知识构建Pipeline示例
 */
public class CompleteBuilderExample {

    public static void main(String[] args) throws Exception {

        // ========== 1. 创建构建上下文 ==========
        BuilderContext context = new BuilderContext();
        context.setProjectId(1L);
        context.setBatchSize(100);
        context.setParallelism(1);
        context.setGraphStoreUrl("neo4j://localhost:7687?user=neo4j&password=password");
        context.setOperation(RecordAlterOperationEnum.UPSERT);

        // ========== 2. 创建CSV数据源 ==========
        CsvSourceNodeConfig csvConfig = new CsvSourceNodeConfig(
            "/data/employees.csv",
            1,
            Arrays.asList("id", "name", "age", "department")
        );

        CsvFileSourceReader sourceReader = new CsvFileSourceReader(
            "source_1", "CSV Source", csvConfig
        );

        // ========== 3. 创建类型映射处理器 ==========
        SPGTypeMappingNodeConfig mappingConfig = new SPGTypeMappingNodeConfig();
        mappingConfig.setSpgTypeName("Person");
        mappingConfig.setIdField("id");
        mappingConfig.setNameField("name");
        mappingConfig.setPropertyMappings(Map.of(
            "name", "name",
            "age", "age",
            "department", "department"
        ));

        SPGTypeMappingNodeConfigs mappingConfigs = new SPGTypeMappingNodeConfigs();
        mappingConfigs.setMappingNodeConfigs(Arrays.asList(mappingConfig));

        SPGTypeMappingProcessor mappingProcessor = new SPGTypeMappingProcessor(
            "mapping_1", "Type Mapping", mappingConfigs
        );

        // ========== 4. 创建图存储写入器 ==========
        GraphStoreSinkNodeConfig sinkConfig = new GraphStoreSinkNodeConfig();
        sinkConfig.setIsWriter(true);

        GraphStoreSinkWriter sinkWriter = new GraphStoreSinkWriter(
            "sink_1", "Graph Store Writer", sinkConfig
        );

        // ========== 5. 初始化所有节点 ==========
        sourceReader.init(context);
        mappingProcessor.init(context);
        sinkWriter.init(context);

        // ========== 6. 执行Pipeline ==========
        int totalRecords = 0;
        List<BaseRecord> batch;

        while (!(batch = sourceReader.read()).isEmpty()) {
            System.out.println("读取批次，大小: " + batch.size());

            // 类型映射
            List<BaseRecord> mappedRecords = mappingProcessor.process(batch);
            System.out.println("映射后记录数: " + mappedRecords.size());

            // 写入图存储
            sinkWriter.write(mappedRecords);

            totalRecords += batch.size();
            System.out.println("已处理记录数: " + totalRecords);
        }

        // ========== 7. 清理资源 ==========
        sourceReader.close();
        mappingProcessor.close();
        sinkWriter.close();

        System.out.println("知识构建完成！总共处理 " + totalRecords + " 条记录。");
    }
}

4. SPG-Reasoner 逻辑推理

4.1 模块概述

SPG-Reasoner是基于KGDSL（知识图谱领域专用语言）的推理引擎，支持复杂的图查询和逻辑推理。

核心包路径

reasoner/
├── kgdsl-parser/                  # KGDSL解析器
│   └── src/main/antlr4/
│       └── KGDSL.g4              # ANTLR4语法定义
│
├── lube-logical/                  # 逻辑层
│   └── src/main/scala/com/antgroup/openspg/reasoner/lube/logical/
│       ├── operators/             # 逻辑算子
│       │   ├── LogicalOperator.scala
│       │   ├── PatternScan.scala
│       │   ├── LinkedExpand.scala
│       │   ├── Filter.scala
│       │   ├── Project.scala
│       │   └── Aggregate.scala
│       └── planning/              # 逻辑计划
│           └── LogicalPlanner.scala
│
├── lube-physical/                 # 物理层
│   └── src/main/scala/com/antgroup/openspg/reasoner/lube/physical/
│       ├── operators/             # 物理算子
│       │   ├── Start.scala
│       │   ├── Unfold.scala
│       │   └── Fold.scala
│       └── planning/
│           └── PhysicalPlanner.scala
│
├── runner/local-runner/           # 本地执行器
│   └── src/main/scala/com/antgroup/openspg/reasoner/runner/local/
│       └── LocalReasonerRunner.scala
│
└── catalog/openspg-catalog/       # Schema目录
    └── src/main/scala/com/antgroup/openspg/reasoner/catalog/impl/
        └── OpenSPGCatalog.scala

4.2 核心类详解

4.2.1 ReasonerManager - 推理管理服务

接口位置: server/biz/service/src/main/java/com/antgroup/openspg/server/biz/service/ReasonerManager.java

java

/**
 * 推理管理接口
 */
public interface ReasonerManager {

    /**
     * 执行Thinker推理
     *
     * @param request Thinker任务请求
     * @return Thinker任务响应
     */
    ThinkerTaskResponse thinker(ThinkerTaskRequest request);

    /**
     * 执行推理任务
     *
     * @param request 推理任务请求
     * @return 推理任务响应
     */
    ReasonerTaskResponse reason(ReasonerTaskRequest request);

    /**
     * 获取推理Schema
     *
     * @param projectId 项目ID
     * @return 项目Schema
     */
    ProjectSchema getReasonSchema(Long projectId);
}

实现位置: server/biz/service/src/main/java/com/antgroup/openspg/server/biz/service/impl/ReasonerManagerImpl.java

java

@Service
public class ReasonerManagerImpl implements ReasonerManager {

    @Autowired
    private ProjectManager projectManager;

    @Autowired
    private ReasonerService reasonerService;

    @Autowired
    private CatalogService catalogService;

    @Override
    public ReasonerTaskResponse reason(ReasonerTaskRequest request) {
        // 获取图存储URL
        String graphStoreUrl = getGraphStoreUrl(request.getProjectId());

        // 创建推理任务
        ReasonerTask reasonerTask = new ReasonerTask();
        reasonerTask.setTaskId(
            request.getTaskId() != null
                ? request.getTaskId()
                : UUID.randomUUID().toString()
        );
        reasonerTask.setDsl(request.getDsl());
        reasonerTask.setParams(request.getParams());
        reasonerTask.setProjectId(request.getProjectId());
        reasonerTask.setGraphStoreUrl(graphStoreUrl);

        log.info("执行KGDSL: " + request.getDsl());

        // 执行推理任务
        ReasonerTask result = reasonerService.runTask(reasonerTask);

        // 构建响应
        ReasonerTaskResponse response = new ReasonerTaskResponse();
        response.setProjectId(request.getProjectId());
        response.setTask(result);

        return response;
    }

    @Override
    public ProjectSchema getReasonSchema(Long projectId) {
        return catalogService.getSchemaInfo(
            projectId,
            getGraphStoreUrl(projectId)
        );
    }

    protected String getGraphStoreUrl(Long projectId) {
        return projectManager.getGraphStoreUrl(projectId);
    }
}

4.2.2 逻辑算子

SPG-Reasoner使用一系列逻辑算子构建查询计划：

PatternScan - 模式扫描

位置: reasoner/lube-logical/src/main/scala/com/antgroup/openspg/reasoner/lube/logical/operators/PatternScan.scala

scala

/**
 * 模式扫描算子
 *
 * 扫描匹配指定模式的节点和边
 */
case class PatternScan(
    pattern: Pattern,              // 图模式
    alias: Option[String] = None,  // 别名
    in: LogicalOperator = Start    // 输入算子
) extends StackingLogicalOperator {

    override def refFields: List[Var] = {
        // 返回模式中引用的所有变量
        pattern.topology.nodes.map(_.alias).toList
    }

    override def fields: List[Var] = {
        // 返回模式产生的所有字段
        pattern.topology.alias.toList
    }
}

LinkedExpand - 链接扩展

scala

/**
 * 链接扩展算子
 *
 * 从起始节点沿着边扩展到邻居节点
 */
case class LinkedExpand(
    startAlias: Var,               // 起始节点别名
    edgePattern: Connection,       // 边模式
    endAlias: Option[Var],         // 结束节点别名
    direction: Direction,          // 扩展方向
    in: LogicalOperator            // 输入算子
) extends StackingLogicalOperator {

    override def refFields: List[Var] = List(startAlias)

    override def fields: List[Var] = {
        endAlias.toList ++ List(edgePattern.alias)
    }
}

Filter - 过滤

scala

/**
 * 过滤算子
 *
 * 根据条件表达式过滤记录
 */
case class Filter(
    expr: Expr,                    // 过滤条件表达式
    in: LogicalOperator            // 输入算子
) extends StackingLogicalOperator {

    override def refFields: List[Var] = expr.refFields.toList

    override def fields: List[Var] = in.fields
}

Project - 投影

scala

/**
 * 投影算子
 *
 * 选择指定的字段
 */
case class Project(
    fields: List[Var],             // 投影字段列表
    in: LogicalOperator            // 输入算子
) extends StackingLogicalOperator {

    override def refFields: List[Var] = fields

    override def fields: List[Var] = fields
}

Aggregate - 聚合

scala

/**
 * 聚合算子
 *
 * 执行分组和聚合操作
 */
case class Aggregate(
    group: List[Var],              // 分组字段
    aggregations: List[AggFunc],   // 聚合函数
    in: LogicalOperator            // 输入算子
) extends StackingLogicalOperator {

    override def refFields: List[Var] = {
        group ++ aggregations.flatMap(_.refFields)
    }

    override def fields: List[Var] = {
        group ++ aggregations.map(_.alias)
    }
}

4.3 KGDSL语言

4.3.1 KGDSL基本语法

KGDSL (Knowledge Graph Domain Specific Language) 是OpenSPG的查询语言。

基本结构：

Define (pattern) [Constraint {...}] [Rule: ...]

示例1：查询人物的朋友

kgdsl

// 定义模式：(人物)-[朋友关系]->(朋友)
Define (s:Person)-[p:friendOf]->(o:Person)

// 规则：返回所有人物及其朋友
Rule: result = (s, o)

示例2：带条件的查询

kgdsl

// 查询在特定公司工作的员工
Define (s:Person)-[p:worksAt]->(o:Company)

// 约束条件
Constraint {
    o.name == "阿里巴巴"
    s.age >= 25
}

// 规则
Rule: employees = s

示例3：多跳查询

kgdsl

// 查询通过共同朋友认识的潜在好友
Define (p1:Person)-[:friendOf]->(mutual:Person)-[:friendOf]->(p2:Person)

Constraint {
    p1.id != p2.id
}

Rule: potentialFriends = (p1, p2, mutual)

示例4：聚合查询

kgdsl

// 统计每个公司的员工数
Define (s:Person)-[:worksAt]->(o:Company)

Rule:
    companyStats = group(o.name).agg(count(s) as employeeCount)

4.3.2 KGDSL执行流程

KGDSL查询
    ↓
┌───────────────┐
│ KGDSL Parser  │ ← ANTLR4解析
└───────┬───────┘
        ↓
┌───────────────┐
│ Logical Plan  │ ← 逻辑计划生成
└───────┬───────┘   (PatternScan, Filter, Project等)
        ↓
┌───────────────┐
│ Optimization  │ ← 逻辑优化
└───────┬───────┘   (谓词下推、连接重排序等)
        ↓
┌───────────────┐
│ Physical Plan │ ← 物理计划生成
└───────┬───────┘   (Unfold, Join, Aggregate等)
        ↓
┌───────────────┐
│   Execution   │ ← 物理执行
└───────┬───────┘
        ↓
    查询结果

4.4 推理使用示例

4.4.1 Java API调用

java

/**
 * 使用Java API执行推理查询
 */
public class ReasonerExample {

    @Autowired
    private ReasonerManager reasonerManager;

    public void executeReasoning() {

        // ========== 1. 创建推理请求 ==========
        ReasonerTaskRequest request = new ReasonerTaskRequest();
        request.setProjectId(1L);
        request.setTaskId(UUID.randomUUID().toString());

        // ========== 2. 编写KGDSL查询 ==========
        String kgdsl = """
            // 查询在阿里巴巴工作的员工
            Define (s:Person)-[p:worksAt]->(o:Company)

            Constraint {
                o.name == "阿里巴巴"
            }

            Rule: employees = s
        """;
        request.setDsl(kgdsl);

        // 参数（可选）
        Map<String, Object> params = new HashMap<>();
        params.put("minAge", 25);
        request.setParams(params);

        // ========== 3. 执行推理 ==========
        ReasonerTaskResponse response = reasonerManager.reason(request);

        // ========== 4. 处理结果 ==========
        ReasonerTask task = response.getTask();
        System.out.println("任务ID: " + task.getTaskId());
        System.out.println("状态: " + task.getStatus());

        // 获取结果
        List<Map<String, Object>> results = task.getResults();
        for (Map<String, Object> result : results) {
            System.out.println("员工: " + result);
        }
    }
}

4.4.2 REST API调用

bash

# 执行推理查询
curl -X POST http://localhost:8887/public/v1/reason/execute \
  -H "Content-Type: application/json" \
  -d '{
    "projectId": 1,
    "dsl": "Define (s:Person)-[p:worksAt]->(o:Company)\nConstraint { o.name == \"阿里巴巴\" }\nRule: employees = s",
    "params": {}
  }'

响应：

json

{
  "success": true,
  "data": {
    "projectId": 1,
    "task": {
      "taskId": "abc-123-def",
      "status": "SUCCESS",
      "results": [
        {
          "id": "P001",
          "name": "张三",
          "age": 30
        },
        {
          "id": "P002",
          "name": "李四",
          "age": 28
        }
      ]
    }
  }
}

4.4.3 复杂推理示例

社交网络分析：找出影响力人物

kgdsl

// 定义：一个人的影响力 = 他的朋友数量 + 朋友的朋友数量 * 0.5

// 第一层朋友
Define (p:Person)-[:friendOf]->(f1:Person)

// 第二层朋友
Define (f1)-[:friendOf]->(f2:Person)

Constraint {
    p.id != f2.id
}

Rule:
    influence = group(p.id, p.name).agg(
        count(distinct f1) as direct_friends,
        count(distinct f2) as indirect_friends,
        count(distinct f1) + count(distinct f2) * 0.5 as influence_score
    )
    order by influence_score desc
    limit 10

企业关系网络：找出关联企业

kgdsl

// 通过共同股东或高管找出关联企业

// 路径1：共同股东
Define (c1:Company)<-[:investIn]-(investor:Person)-[:investIn]->(c2:Company)

// 路径2：高管任职
Define (c1:Company)<-[:worksAt]-(exec:Person)-[:worksAt]->(c2:Company)

Constraint {
    c1.id != c2.id
    exec.position in ["CEO", "董事长", "总经理"]
}

Rule:
    relatedCompanies = group(c1.name, c2.name).agg(
        count(*) as connection_count,
        collect(investor.name) as common_investors,
        collect(exec.name) as common_executives
    )
    where connection_count >= 2

5. 三大模块集成

5.1 完整工作流程

┌─────────────────────────────────────────────────────────────┐
│                   完整知识图谱构建流程                         │
└─────────────────────────────────────────────────────────────┘

第一步：Schema定义 (SPG-Schema)
    ↓
┌─────────────────┐
│ SchemaManager   │ 定义实体类型、关系
│ .alterSchema()  │ - EntityType: Person, Company
└────────┬────────┘ - Relation: worksAt
         │
         ▼
第二步：数据构建 (SPG-Builder)
    ↓
┌─────────────────┐
│ Source Reader   │ 读取原始数据
│ .read()         │ - CSV/JSON/DB
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Type Mapping    │ 映射到Schema
│ .process()      │ BuilderRecord → EntityRecord
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Sink Writer     │ 写入图数据库
│ .write()        │ Neo4j/TuGraph
└────────┬────────┘
         │
         ▼
第三步：知识推理 (SPG-Reasoner)
    ↓
┌─────────────────┐
│ ReasonerManager │ 执行KGDSL查询
│ .reason()       │ - 图遍历
└────────┬────────┘ - 逻辑推理
         │         - 聚合统计
         ▼
    推理结果

5.2 集成代码示例

java

/**
 * 三大模块集成示例
 */
public class IntegratedExample {

    @Autowired
    private SchemaManager schemaManager;

    @Autowired
    private ReasonerManager reasonerManager;

    public void completeWorkflow() throws Exception {

        Long projectId = 1L;

        // ==================== 第一步：定义Schema ====================

        System.out.println("步骤1：定义Schema");

        // 定义Company类型
        EntityType companyType = new EntityType(
            new BasicInfo<>(
                new SPGTypeIdentifier("enterprise", "Company"),
                "公司", "企业组织"
            ),
            ParentTypeInfo.THING,
            Arrays.asList(
                createProperty("name", "公司名称", BasicType.TEXT, IndexTypeEnum.TEXT),
                createProperty("foundYear", "成立年份", BasicType.TEXT, IndexTypeEnum.EXACT)
            ),
            null,
            new SPGTypeAdvancedConfig()
        );

        // 定义Person类型
        EntityType personType = new EntityType(
            new BasicInfo<>(
                new SPGTypeIdentifier("enterprise", "Person"),
                "人物", "自然人"
            ),
            ParentTypeInfo.THING,
            Arrays.asList(
                createProperty("name", "姓名", BasicType.TEXT, IndexTypeEnum.TEXT),
                createProperty("age", "年龄", BasicType.INTEGER, IndexTypeEnum.RANGE)
            ),
            Arrays.asList(
                new Relation(
                    new BasicInfo<>(new PredicateIdentifier("worksAt"), "工作于", ""),
                    personType.toRef(),
                    companyType.toRef(),
                    false,
                    new RelationAdvancedConfig()
                )
            ),
            new SPGTypeAdvancedConfig()
        );

        companyType.setAlterOperation(AlterOperationEnum.CREATE);
        personType.setAlterOperation(AlterOperationEnum.CREATE);

        // 提交Schema
        SchemaDraft draft = new SchemaDraft();
        draft.setAlterSpgTypes(Arrays.asList(companyType, personType));

        SchemaAlterRequest schemaRequest = new SchemaAlterRequest();
        schemaRequest.setProjectId(projectId);
        schemaRequest.setSchemaDraft(draft);

        schemaManager.alterSchema(schemaRequest);
        System.out.println("Schema定义完成");

        // ==================== 第二步：构建知识 ====================

        System.out.println("\n步骤2：构建知识图谱");

        // 创建构建上下文
        BuilderContext context = new BuilderContext();
        context.setProjectId(projectId);
        context.setBatchSize(100);
        context.setGraphStoreUrl("neo4j://localhost:7687?user=neo4j&password=password");
        context.setOperation(RecordAlterOperationEnum.UPSERT);

        // 创建Pipeline
        CsvSourceNodeConfig csvConfig = new CsvSourceNodeConfig(
            "/data/employees.csv", 1,
            Arrays.asList("id", "name", "age", "company")
        );
        CsvFileSourceReader sourceReader = new CsvFileSourceReader(
            "source_1", "CSV Source", csvConfig
        );

        SPGTypeMappingNodeConfig mappingConfig = new SPGTypeMappingNodeConfig();
        mappingConfig.setSpgTypeName("Person");
        mappingConfig.setIdField("id");
        mappingConfig.setNameField("name");
        mappingConfig.setPropertyMappings(Map.of(
            "name", "name",
            "age", "age"
        ));

        SPGTypeMappingNodeConfigs mappingConfigs = new SPGTypeMappingNodeConfigs();
        mappingConfigs.setMappingNodeConfigs(Arrays.asList(mappingConfig));

        SPGTypeMappingProcessor mappingProcessor = new SPGTypeMappingProcessor(
            "mapping_1", "Type Mapping", mappingConfigs
        );

        GraphStoreSinkNodeConfig sinkConfig = new GraphStoreSinkNodeConfig();
        sinkConfig.setIsWriter(true);

        GraphStoreSinkWriter sinkWriter = new GraphStoreSinkWriter(
            "sink_1", "Graph Store Writer", sinkConfig
        );

        // 初始化
        sourceReader.init(context);
        mappingProcessor.init(context);
        sinkWriter.init(context);

        // 执行构建
        int totalRecords = 0;
        List<BaseRecord> batch;

        while (!(batch = sourceReader.read()).isEmpty()) {
            List<BaseRecord> mappedRecords = mappingProcessor.process(batch);
            sinkWriter.write(mappedRecords);
            totalRecords += batch.size();
        }

        // 清理
        sourceReader.close();
        mappingProcessor.close();
        sinkWriter.close();

        System.out.println("知识构建完成，共处理 " + totalRecords + " 条记录");

        // ==================== 第三步：知识推理 ====================

        System.out.println("\n步骤3：知识推理");

        // 查询1：统计各公司员工数
        String kgdsl1 = """
            Define (s:Person)-[p:worksAt]->(o:Company)

            Rule:
                companyStats = group(o.name).agg(
                    count(s) as employeeCount,
                    avg(s.age) as avgAge
                )
                order by employeeCount desc
        """;

        ReasonerTaskRequest reasonRequest1 = new ReasonerTaskRequest();
        reasonRequest1.setProjectId(projectId);
        reasonRequest1.setDsl(kgdsl1);

        ReasonerTaskResponse reasonResponse1 = reasonerManager.reason(reasonRequest1);
        System.out.println("公司员工统计：");
        printResults(reasonResponse1.getTask().getResults());

        // 查询2：找出年龄在25-35岁的员工
        String kgdsl2 = """
            Define (s:Person)-[p:worksAt]->(o:Company)

            Constraint {
                s.age >= 25
                s.age <= 35
            }

            Rule: youngEmployees = (s, o)
        """;

        ReasonerTaskRequest reasonRequest2 = new ReasonerTaskRequest();
        reasonRequest2.setProjectId(projectId);
        reasonRequest2.setDsl(kgdsl2);

        ReasonerTaskResponse reasonResponse2 = reasonerManager.reason(reasonRequest2);
        System.out.println("\n年轻员工列表：");
        printResults(reasonResponse2.getTask().getResults());

        System.out.println("\n完整工作流程执行完毕！");
    }

    private Property createProperty(String name, String nameZh,
                                    BasicType type, IndexTypeEnum indexType) {
        return new Property(
            new BasicInfo<>(new PredicateIdentifier(name), nameZh, ""),
            null,
            new SPGTypeRef(type),
            false,
            new PropertyAdvancedConfig().setIndexType(indexType)
        );
    }

    private void printResults(List<Map<String, Object>> results) {
        for (Map<String, Object> result : results) {
            System.out.println("  " + result);
        }
    }
}

6. 完整应用示例

6.1 企业知识图谱应用

这个示例展示了如何构建一个完整的企业知识图谱系统。

java

/**
 * 企业知识图谱应用
 */
public class EnterpriseKnowledgeGraphApp {

    @Autowired
    private SchemaManager schemaManager;

    @Autowired
    private ReasonerManager reasonerManager;

    /**
     * 初始化企业知识图谱
     */
    public void initialize() throws Exception {
        // 1. 定义Schema
        defineSchema();

        // 2. 导入数据
        importData();

        // 3. 执行分析
        analyzeData();
    }

    /**
     * 定义企业知识图谱Schema
     */
    private void defineSchema() {
        // 定义实体类型
        EntityType companyType = defineCompanyType();
        EntityType personType = definePersonType();
        EntityType productType = defineProductType();

        // 提交Schema
        SchemaDraft draft = new SchemaDraft();
        draft.setAlterSpgTypes(Arrays.asList(
            companyType,
            personType,
            productType
        ));

        SchemaAlterRequest request = new SchemaAlterRequest();
        request.setProjectId(1L);
        request.setSchemaDraft(draft);

        schemaManager.alterSchema(request);
    }

    /**
     * 导入企业数据
     */
    private void importData() throws Exception {
        BuilderContext context = createBuilderContext();

        // 导入公司数据
        importCompanies(context);

        // 导入员工数据
        importEmployees(context);

        // 导入产品数据
        importProducts(context);
    }

    /**
     * 分析企业数据
     */
    private void analyzeData() {
        // 分析1：企业规模分析
        analyzeCompanyScale();

        // 分析2：人才流动分析
        analyzeTalentFlow();

        // 分析3：产品竞争分析
        analyzeProductCompetition();
    }

    // 具体实现方法...
}

附录

A. 核心类路径速查表

模块	类名	完整路径
SPG-Schema
	`BaseSPGType`	`core/schema/model/.../type/BaseSPGType.java`
	`EntityType`	`core/schema/model/.../type/EntityType.java`
	`Property`	`core/schema/model/.../predicate/Property.java`
	`Relation`	`core/schema/model/.../predicate/Relation.java`
	`ProjectSchema`	`core/schema/model/.../type/ProjectSchema.java`
	`SchemaManager`	`server/biz/schema/.../SchemaManager.java`
SPG-Builder
	`BuilderContext`	`builder/core/.../runtime/BuilderContext.java`
	`BasePhysicalNode`	`builder/core/.../physical/BasePhysicalNode.java`
	`BaseSourceReader`	`builder/runner/local/.../source/BaseSourceReader.java`
	`BaseProcessor`	`builder/core/.../process/BaseProcessor.java`
	`BaseSinkWriter`	`builder/runner/local/.../sink/BaseSinkWriter.java`
	`BuilderRecord`	`builder/model/.../record/BuilderRecord.java`
	`EntityRecord`	`builder/model/.../record/EntityRecord.java`
SPG-Reasoner
	`ReasonerManager`	`server/biz/service/.../ReasonerManager.java`
	`LogicalOperator`	`reasoner/lube-logical/.../operators/LogicalOperator.scala`
	`PatternScan`	`reasoner/lube-logical/.../operators/PatternScan.scala`
	`LinkedExpand`	`reasoner/lube-logical/.../operators/LinkedExpand.scala`

B. 常用枚举类型

java

// SPG类型枚举
public enum SPGTypeEnum {
    BASIC_TYPE,      // 基础类型
    STANDARD_TYPE,   // 标准类型
    ENTITY_TYPE,     // 实体类型
    CONCEPT_TYPE,    // 概念类型
    EVENT_TYPE       // 事件类型
}

// 索引类型枚举
public enum IndexTypeEnum {
    EXACT,          // 精确匹配
    TEXT,           // 全文检索
    VECTOR,         // 向量索引
    TEXT_AND_VECTOR,// 混合索引
    RANGE,          // 范围索引
    NONE            // 无索引
}

// 操作类型枚举
public enum AlterOperationEnum {
    CREATE,         // 创建
    UPDATE,         // 更新
    DELETE          // 删除
}

// 记录操作枚举
public enum RecordAlterOperationEnum {
    INSERT,         // 插入
    UPDATE,         // 更新
    DELETE,         // 删除
    UPSERT          // 插入或更新
}

C. 参考资源

OpenSPG官方文档: https://openspg.yuque.com/ndx6g9/docs
GitHub仓库: https://github.com/OpenSPG/openspg
SPG白皮书: https://openspg.github.io/v2/blog/design_philosophy/white_paper/openspg
KAG论文: https://arxiv.org/abs/2409.13731

版本信息

文档版本: 1.0
OpenSPG版本: 0.8
最后更新: 2025-12-13

许可证

本文档基于 Apache License 2.0 开源协议。

OpenSPG 核心模块深度解析 ​

目录 ​

1. 概述 ​

1.1 模块职责 ​

1.2 技术栈 ​

2. SPG-Schema 语义建模 ​

2.1 模块概述 ​

核心包路径 ​

2.2 核心类详解 ​

2.2.1 BaseSPGType - SPG类型基类 ​

2.2.2 EntityType - 实体类型 ​

2.2.3 Property - 属性定义 ​

2.2.4 Relation - 关系定义 ​

2.2.5 ProjectSchema - 项目Schema ​

2.3 SchemaManager - Schema管理服务 ​

2.4 Schema操作示例 ​

2.4.1 创建Schema ​

2.4.2 查询Schema ​

2.4.3 更新Schema ​

2.5 Schema约束 ​

2.5.1 约束类型 ​

2.5.2 添加约束示例 ​

3. SPG-Builder 知识构建 ​

3.1 模块概述 ​

核心包路径 ​

3.2 核心类详解 ​

3.2.1 BuilderContext - 构建上下文 ​

3.2.2 BasePhysicalNode - 物理节点基类 ​

3.2.3 数据记录模型 ​

3.3 数据处理Pipeline ​

3.3.1 Pipeline架构 ​

3.3.2 Source Reader - 数据源读取 ​

3.3.3 Processor - 数据处理 ​

3.3.4 Sink Writer - 数据写入 ​

3.4 完整构建示例 ​

4. SPG-Reasoner 逻辑推理 ​

4.1 模块概述 ​

核心包路径 ​

4.2 核心类详解 ​

4.2.1 ReasonerManager - 推理管理服务 ​

4.2.2 逻辑算子 ​

4.3 KGDSL语言 ​

4.3.1 KGDSL基本语法 ​

4.3.2 KGDSL执行流程 ​

4.4 推理使用示例 ​

4.4.1 Java API调用 ​

4.4.2 REST API调用 ​

4.4.3 复杂推理示例 ​

5. 三大模块集成 ​

5.1 完整工作流程 ​

5.2 集成代码示例 ​

6. 完整应用示例 ​

6.1 企业知识图谱应用 ​

附录 ​

A. 核心类路径速查表 ​

B. 常用枚举类型 ​

C. 参考资源 ​

版本信息 ​

许可证 ​

OpenSPG 核心模块深度解析

目录

1. 概述

1.1 模块职责

1.2 技术栈

2. SPG-Schema 语义建模

2.1 模块概述

核心包路径

2.2 核心类详解

2.2.1 BaseSPGType - SPG类型基类

2.2.2 EntityType - 实体类型

2.2.3 Property - 属性定义

2.2.4 Relation - 关系定义

2.2.5 ProjectSchema - 项目Schema

2.3 SchemaManager - Schema管理服务

2.4 Schema操作示例

2.4.1 创建Schema

2.4.2 查询Schema

2.4.3 更新Schema

2.5 Schema约束

2.5.1 约束类型

2.5.2 添加约束示例

3. SPG-Builder 知识构建

3.1 模块概述

核心包路径

3.2 核心类详解

3.2.1 BuilderContext - 构建上下文

3.2.2 BasePhysicalNode - 物理节点基类

3.2.3 数据记录模型

3.3 数据处理Pipeline

3.3.1 Pipeline架构

3.3.2 Source Reader - 数据源读取

3.3.3 Processor - 数据处理

3.3.4 Sink Writer - 数据写入

3.4 完整构建示例

4. SPG-Reasoner 逻辑推理

4.1 模块概述

核心包路径

4.2 核心类详解

4.2.1 ReasonerManager - 推理管理服务

4.2.2 逻辑算子

4.3 KGDSL语言

4.3.1 KGDSL基本语法

4.3.2 KGDSL执行流程

4.4 推理使用示例

4.4.1 Java API调用

4.4.2 REST API调用

4.4.3 复杂推理示例

5. 三大模块集成

5.1 完整工作流程

5.2 集成代码示例

6. 完整应用示例

6.1 企业知识图谱应用

附录

A. 核心类路径速查表

B. 常用枚举类型

C. 参考资源

版本信息

许可证