创建自定义算子和工作流模板¶

使用 AI Studio SDK 创建自定义算子或工作流模板。创建结果提供后续读取、更新、启停、测试或删除所需的资源引用。

任务流程¶

创建算子时，需要完整的算子定义；创建成功后先读取详情或提交一次测试，再决定是否更新、启用或删除。创建模板时，需要名称和 DSL 内容；创建结果可继续用于读取、更新或删除模板。

准备¶

需要的内容	如何取得	在本页中的作用
已认证的客户端和工作区 ID	在首次接入流程中创建或选择。	确定资源访问和创建范围。
算子定义	由算子开发者提供名称、标识、说明、输入输出结构、运行语言和代码。	创建可测试的自定义算子。
测试输入	由算子输入定义和测试场景提供。	提交算子测试运行。
模板名称和 DSL 内容	由工作流设计提供。	创建工作流模板。

创建和删除会改变工作区资源。只对可安全管理的算子和模板执行示例中的写操作。

创建并测试自定义算子¶

下面的示例创建一个算子，读取其详情，再提交测试运行。测试运行返回状态和输出信息；请求返回不表示业务断言已经满足。

Python

import os

import moi_product_sdk as sdk

client = sdk.new_with_personal_access_token(
    os.environ["PRODUCT_API_BASE_URL"],
    os.environ["PRODUCT_API_KEY"],
)
workspace = client.workspace("<workspace-id>")
operator, created = workspace.create_custom_operator(
    "<operator-name>",
    "<operator-identifier>",
    "<operator-description>",
    {"type": "object"},
    {"type": "object"},
    sdk.with_custom_operator_language("python"),
    sdk.with_custom_operator_code("<operator-code>"),
)
operator_info = operator.info()
test_run = operator.test({"<input-name>": "<input-value>"})

print(created)
print(operator_info)
print(test_run)

package main

import (
	"context"
	"fmt"
	"os"

	sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
)

func main() {
	ctx := context.Background()
	client, err := sdk.NewWithPersonalAccessToken(
		os.Getenv("PRODUCT_API_BASE_URL"),
		os.Getenv("PRODUCT_API_KEY"),
	)
	if err != nil {
		panic(err)
	}
	workspace, err := client.Workspace("<workspace-id>")
	if err != nil {
		panic(err)
	}
	operator, created, err := workspace.CreateCustomOperator(
		ctx,
		"<operator-name>",
		"<operator-identifier>",
		"<operator-description>",
		map[string]any{"type": "object"},
		map[string]any{"type": "object"},
		sdk.WithCustomOperatorLanguage("python"),
		sdk.WithCustomOperatorCode("<operator-code>"),
	)
	if err != nil {
		panic(err)
	}
	operatorInfo, err := operator.Info(ctx)
	if err != nil {
		panic(err)
	}
	testRun, err := operator.Test(ctx, map[string]any{"<input-name>": "<input-value>"})
	if err != nil {
		panic(err)
	}

	fmt.Println(created)
	fmt.Println(operatorInfo)
	fmt.Println(testRun)
}

维护自定义算子¶

使用创建结果或已确认的算子 ID 读取算子后，可以更新、启用、停用或删除。删除前应确认没有仍在使用的工作流引用；删除后不要继续使用原资源引用。

创建工作流模板¶

继续使用本页建立的工作区作用域。创建模板后，使用创建结果更新模板；模板 ID 来自创建结果或已确认的资源结果，不要从模板名称推断。

Python

template, created_template = workspace.create_workflow_template(
    "<template-name>",
    "<workflow-dsl-yaml>",
    sdk.with_workflow_template_key("<template-key>"),
)
updated = template.update("<updated-template-name>", "<updated-workflow-dsl-yaml>")

print(created_template)
print(updated)

template, createdTemplate, err := workspace.CreateWorkflowTemplate(
	ctx,
	"<template-name>",
	"<workflow-dsl-yaml>",
	sdk.WithWorkflowTemplateKey("<template-key>"),
)
if err != nil {
	panic(err)
}
updated, err := template.Update(ctx, "<updated-template-name>", "<updated-workflow-dsl-yaml>")
if err != nil {
	panic(err)
}

fmt.Println(createdTemplate)
fmt.Println(updated)

下一步¶

最后更新于 2026 年 9 月 1 日