当前位置：首页 > ops >正文

扣子（coze）实践指南进阶篇——创建工作流，并将工作流接入智能体

ops 2025/9/1 5:43:33

在这里插入图片描述
大家好，欢迎阅读这份《智能体（AI+Agent）开发指南》！
在大模型和智能体快速发展的今天，很多朋友希望学习如何从零开始搭建一个属于自己的智能体。本教程的特点是完全基于国产大模型与火山推理引擎实现，不用翻墙即可上手，非常适合国内开发者快速实践。
通过循序渐进的讲解，你将学会从环境配置、基础构建、进阶功能到实际案例的完整流程，逐步掌握智能体开发的核心技能。无论你是初学者还是有经验的工程师，相信这份教程都能为你带来启发。

在这里插入图片描述
工作流是一系列可执行指令的集合，用于实现业务逻辑或完成特定任务。它为应用/智能体的数据流动和任务处理提供了一个结构化框架。工作流的核心在于将大模型的强大能力与特定的业务逻辑相结合，通过系统化、流程化的方法来实现高效、可扩展的 AI 应用开发。
扣子提供了一个可视化画布，你可以通过拖拽节点迅速搭建工作流。同时，支持在画布实时调试工作流。在工作流画布中，你可以清晰地看到数据的流转过程和任务的执行顺序。

coze基础篇：扣子（coze）实践指南基础篇——五分钟快速接入 DeepSeek 模型

一. 创建工作流

本小节会创建一个工作流，工作流会根据用户输入在网上搜索、用大模型对搜索的结果进行总结，最后将结果输出，如下图所示。
在这里插入图片描述
创建一个空的工作流

添加插件

插入头条搜索，让工作流具备搜索能力

连接用户输入，并设置输入参数，调试成功

以同样的方法，插入大模型，让工作流具备大模型的能力

将各个模块连接起来，运行测试
在这里插入图片描述
测试成功，就可以发布啦。在资源库里可以看到

当然，也是可以用代码调用工作流的，调用方式如下：

"""
This example describes how to use the workflow interface to stream chat.
"""import os
# Our official coze sdk for Python [cozepy](https://github.com/coze-dev/coze-py)
from cozepy import COZE_CN_BASE_URL# Get an access_token through personal access token or oauth.
coze_api_token = ''
# The default access is api.coze.com, but if you need to access api.coze.cn,
# please use base_url to configure the api endpoint to access
coze_api_base = COZE_CN_BASE_URLfrom cozepy import Coze, TokenAuth, Stream, WorkflowEvent, WorkflowEventType  # noqa# Init the Coze client through the access_token.
coze = Coze(auth=TokenAuth(token=coze_api_token), base_url=coze_api_base)# Create a workflow instance in Coze, copy the last number from the web link as the workflow's ID.
workflow_id = '7541314777105858606'# The stream interface will return an iterator of WorkflowEvent. Developers should iterate
# through this iterator to obtain WorkflowEvent and handle them separately according to
# the type of WorkflowEvent.
def handle_workflow_iterator(stream: Stream[WorkflowEvent]):for event in stream:if event.event == WorkflowEventType.MESSAGE:print("got message", event.message)elif event.event == WorkflowEventType.ERROR:print("got error", event.error)elif event.event == WorkflowEventType.INTERRUPT:handle_workflow_iterator(coze.workflows.runs.resume(workflow_id=workflow_id,event_id=event.interrupt.interrupt_data.event_id,resume_data="hey",interrupt_type=event.interrupt.interrupt_data.type,))handle_workflow_iterator(coze.workflows.runs.stream(workflow_id=workflow_id,)
)