工作流相关

更新时间：2026-09-04

创建工作流模版

使用以下代码可以创建工作流模版。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量（可选，推荐使用）
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息和服务端点
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'  # 百舸AIHC服务地址
AK = os.getenv('AK') or 'your-access-key-id'                # 您的Access Key ID
SK = os.getenv('SK') or 'your-secret-access-key'            # 您的Secret Access Key

# 创建BCE客户端配置
config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)

# 创建 aihc client
aihc_client = AihcClient(config)

# 创建工作流模版
try:
    __logger.info('--------------------------------CreatePipeline start--------------------------------')
    name = "python-sdk-test-pipeline"  # 工作流模版名称
    description = "示例 Pipeline"      # 工作流模版描述
    pipeline_type = "normal"           # 工作流模版类型
    resource_pool_id = "aihc-xxx"      # 替换为实际的资源池ID
    concurrency_policy = "Allow"       # 并发策略：Allow / Forbid / Replace
    manifest = (                       # 工作流模版 YAML 定义，详见接口文档
        "version: v1\n"
        "kind: PipelineTemplate\n"
        "taskTemplates:\n"
        "  - name: custom-task-template\n"
        "    type: CustomTask\n"
    )
    response = aihc_client.workflow.CreatePipeline(
        name=name,
        manifest=manifest,
        description=description,
        pipelineType=pipeline_type,
        resourcePoolId=resource_pool_id,
        concurrencyPolicy=concurrency_policy,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为创建工作流模版

查询工作流模版列表

使用以下代码可以查询工作流模版列表。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 查询工作流模版列表
try:
    __logger.info('--------------------------------DescribePipelines start--------------------------------')
    page_number = 1   # 页码，从 1 开始
    page_size = 10    # 每页条数
    response = aihc_client.workflow.DescribePipelines(
        pageNumber=page_number,
        pageSize=page_size,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为查询工作流模版列表

查询工作流模版详情

使用以下代码可以查询工作流模版详情。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 查询工作流模版详情
try:
    __logger.info('--------------------------------DescribePipeline start--------------------------------')
    pipeline_id = "pipe-xxx"  # 替换为实际的工作流模版ID
    response = aihc_client.workflow.DescribePipeline(
        pipelineId=pipeline_id,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为查询工作流模版详情

更新工作流模版

使用以下代码可以更新工作流模版。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 更新工作流模版
try:
    __logger.info('--------------------------------UpdatePipeline start--------------------------------')
    pipeline_id = "pipe-xxx"         # 替换为实际的工作流模版ID
    new_name = "updated-pipeline"    # 更新后的模版名称
    new_description = "更新后的描述"  # 更新后的描述
    response = aihc_client.workflow.UpdatePipeline(
        pipelineId=pipeline_id,
        name=new_name,
        description=new_description,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为更新工作流模版

删除工作流模版

使用以下代码可以删除工作流模版。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 删除工作流模版
try:
    __logger.info('--------------------------------DeletePipeline start--------------------------------')
    pipeline_id = "pipe-xxx"  # 替换为实际的工作流模版ID
    response = aihc_client.workflow.DeletePipeline(
        pipelineId=pipeline_id,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为删除工作流模版

运行工作流

使用以下代码可以运行工作流（创建工作流运行实例）。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 运行工作流
try:
    __logger.info('--------------------------------CreatePipelineRun start--------------------------------')
    pipeline_id = "pipe-xxx"  # 替换为实际的工作流模版ID
    response = aihc_client.workflow.CreatePipelineRun(
        pipelineId=pipeline_id,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为运行工作流

查询工作流运行列表

使用以下代码可以查询工作流运行列表。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 查询工作流运行列表
try:
    __logger.info('--------------------------------DescribePipelineRuns start--------------------------------')
    pipeline_id = "pipe-xxx"  # 替换为实际的工作流模版ID
    page_number = 1           # 页码，从 1 开始
    page_size = 10            # 每页条数
    response = aihc_client.workflow.DescribePipelineRuns(
        pipelineId=pipeline_id,
        pageNumber=page_number,
        pageSize=page_size,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为查询工作流运行列表

查询工作流运行详情

使用以下代码可以查询工作流运行详情。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 查询工作流运行详情
try:
    __logger.info('--------------------------------DescribePipelineRun start--------------------------------')
    run_id = "run-xxx"  # 替换为实际的工作流运行ID
    response = aihc_client.workflow.DescribePipelineRun(
        runId=run_id,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为查询工作流运行详情

查询工作流运行节点列表

使用以下代码可以查询工作流运行节点列表。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 查询工作流运行节点列表
try:
    __logger.info('--------------------------------DescribePipelineRunNodes start--------------------------------')
    run_id = "run-xxx"  # 替换为实际的工作流运行ID
    response = aihc_client.workflow.DescribePipelineRunNodes(
        runId=run_id,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为查询工作流运行节点列表

重试工作流运行

使用以下代码可以重试工作流运行。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 重试工作流运行
try:
    __logger.info('--------------------------------RetryPipelineRun start--------------------------------')
    run_id = "run-xxx"  # 替换为实际的工作流运行ID
    mode = "full"       # 重试模式：full（全部重试）或 failed（仅重试失败节点）
    response = aihc_client.workflow.RetryPipelineRun(
        runId=run_id,
        mode=mode,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为重试工作流运行

停止工作流运行

使用以下代码可以停止工作流运行。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 停止工作流运行
try:
    __logger.info('--------------------------------StopPipelineRun start--------------------------------')
    run_id = "run-xxx"     # 替换为实际的工作流运行ID
    terminate = False      # 是否强制终止：True 表示强制终止，False 表示优雅停止
    message = "手动停止"   # 停止原因（可选）
    response = aihc_client.workflow.StopPipelineRun(
        runId=run_id,
        terminate=terminate,
        message=message,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为停止工作流运行

删除工作流运行实例

使用以下代码可以删除工作流运行实例。

                Python
                
            

                import logging
import os

import dotenv

from baidubce.bce_client_configuration import BceClientConfiguration
from baidubce.auth.bce_credentials import BceCredentials
from baidubce.services.aihc.aihc_client import AihcClient
from baidubce.exception import BceHttpClientError, BceServerError

# 加载环境变量
dotenv.load_dotenv()

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', force=True)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("baidubce").setLevel(logging.INFO)
__logger = logging.getLogger(__name__)
__logger.setLevel(logging.INFO)

# 配置认证信息
HOST = os.getenv('HOST') or 'https://aihc.bj.baidubce.com'
AK = os.getenv('AK') or 'your-access-key-id'
SK = os.getenv('SK') or 'your-secret-access-key'

config = BceClientConfiguration(credentials=BceCredentials(AK, SK), endpoint=HOST)
aihc_client = AihcClient(config)

# 删除工作流运行实例
try:
    __logger.info('--------------------------------DeletePipelineRun start--------------------------------')
    run_id = "run-xxx"  # 替换为实际的工作流运行ID
    response = aihc_client.workflow.DeletePipelineRun(
        runId=run_id,
    )
    print(response)
except BceHttpClientError as e:
    if isinstance(e.last_error, BceServerError):
        __logger.error('send request failed. Response %s, code: %s, msg: %s'
                        % (e.last_error.status_code, e.last_error.code, str(e.last_error)))
    else:
        __logger.error('send request failed. Unknown exception: %s' % e)
            

注意: 根据接口文档去填写具体的访问参数，接口链接为删除工作流运行实例

评价此篇文章

有帮助没帮助

数据集相关

常用工具

百度智能云

百度百舸 · AI计算平台