ben
/
coze-studio
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add config for workflow domain (#1847)

This commit is contained in:
Zhj 2025-08-27 17:04:42 +08:00 committed by GitHub
parent 5562800958
commit 77e1931494
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 135 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/api/handler/coze/workflow_service_test.go
View File

@ -250,7 +250,7 @@ func newWfTestRunner(t *testing.T) *wfTestRunner {
mockTos := storageMock.NewMockStorage(ctrl)
mockTos.EXPECT().GetObjectUrl(gomock.Any(), gomock.Any(), gomock.Any()).Return("", nil).AnyTimes()
workflowRepo := service.NewWorkflowRepository(mockIDGen, db, redisClient, mockTos, cpStore, utChatModel)
workflowRepo := service.NewWorkflowRepository(mockIDGen, db, redisClient, mockTos, cpStore, utChatModel, nil)
mockey.Mock(appworkflow.GetWorkflowDomainSVC).Return(service.NewWorkflowService(workflowRepo)).Build()
mockey.Mock(workflow2.GetRepository).Return(workflowRepo).Build()
publishPatcher := mockey.Mock(appworkflow.PublishWorkflowResource).Return(nil).Build()

27
backend/application/workflow/init.go
View File

@ -19,6 +19,9 @@ package workflow
import (
"context"
"gopkg.in/yaml.v3"
"os"
"github.com/cloudwego/eino/callbacks"
"github.com/cloudwego/eino/compose"
"gorm.io/gorm"
@ -30,7 +33,7 @@ import (
plugin "github.com/coze-dev/coze-studio/backend/domain/plugin/service"
search "github.com/coze-dev/coze-studio/backend/domain/search/service"
"github.com/coze-dev/coze-studio/backend/domain/workflow"
"github.com/coze-dev/coze-studio/backend/domain/workflow/config"
"github.com/coze-dev/coze-studio/backend/domain/workflow/service"
workflowservice "github.com/coze-dev/coze-studio/backend/domain/workflow/service"
"github.com/coze-dev/coze-studio/backend/infra/contract/cache"
@ -57,11 +60,29 @@ type ServiceComponents struct {
WorkflowBuildInChatModel chatmodel.BaseChatModel
}
func InitService(ctx context.Context, components *ServiceComponents) (*ApplicationService, error) {
func initWorkflowConfig() (workflow.WorkflowConfig, error) {
configBs, err := os.ReadFile("resources/conf/workflow/config.yaml")
if err != nil {
return nil, err
}
var cfg *config.WorkflowConfig
err = yaml.Unmarshal(configBs, &cfg)
if err != nil {
return nil, err
}
return cfg, nil
}
func InitService(_ context.Context, components *ServiceComponents) (*ApplicationService, error) {
service.RegisterAllNodeAdaptors()
cfg, err := initWorkflowConfig()
if err != nil {
return nil, err
}
workflowRepo := service.NewWorkflowRepository(components.IDGen, components.DB, components.Cache,
components.Tos, components.CPStore, components.WorkflowBuildInChatModel)
components.Tos, components.CPStore, components.WorkflowBuildInChatModel, cfg)
workflow.SetRepository(workflowRepo)
workflowDomainSVC := service.NewWorkflowService(workflowRepo)

4
backend/conf/workflow/config.yaml Normal file
View File

@ -0,0 +1,4 @@
NodeOfCodeConfig:
SupportThirdPartModules:
- httpx
- numpy

5
backend/domain/workflow/component_interface.go
View File

@ -24,6 +24,7 @@ import (
"github.com/cloudwego/eino/schema"
workflowModel "github.com/coze-dev/coze-studio/backend/api/model/crossdomain/workflow"
"github.com/coze-dev/coze-studio/backend/domain/workflow/config"
"github.com/coze-dev/coze-studio/backend/domain/workflow/entity"
"github.com/coze-dev/coze-studio/backend/domain/workflow/entity/vo"
)
@ -91,3 +92,7 @@ type ToolFromWorkflow interface {
TerminatePlan() vo.TerminatePlan
GetWorkflow() *entity.Workflow
}
type WorkflowConfig interface {
GetNodeOfCodeConfig() *config.NodeOfCodeConfig
}

33
backend/domain/workflow/config/workflow_config.go Normal file
View File

@ -0,0 +1,33 @@
/*
* Copyright 2025 coze-dev Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package config
type WorkflowConfig struct {
NodeOfCodeConfig *NodeOfCodeConfig `yaml:"NodeOfCodeConfig"`
}
func (w WorkflowConfig) GetNodeOfCodeConfig() *NodeOfCodeConfig {
return w.NodeOfCodeConfig
}
type NodeOfCodeConfig struct {
SupportThirdPartModules []string `yaml:"SupportThirdPartModules"`
}
func (n *NodeOfCodeConfig) GetSupportThirdPartModules() []string {
return n.SupportThirdPartModules
}

2
backend/domain/workflow/interface.go
View File

@ -99,6 +99,8 @@ type Repository interface {
idgen.IDGenerator
GetKnowledgeRecallChatModel() model.BaseChatModel
WorkflowConfig
}
var repositorySingleton Repository

10
backend/domain/workflow/internal/canvas/adaptor/canvas_test.go
View File

@ -18,6 +18,7 @@ package adaptor
import (
"context"
"github.com/coze-dev/coze-studio/backend/domain/workflow/config"
"io"
"net"
"net/http"
@ -752,6 +753,15 @@ func TestCodeAndPluginNodes(t *testing.T) {
defer ctrl.Finish()
mockCodeRunner := mockcode.NewMockRunner(ctrl)
mockey.Mock(code.GetCodeRunner).Return(mockCodeRunner).Build()
mockRepo := mockWorkflow.NewMockRepository(ctrl)
mockRepo.EXPECT().GetNodeOfCodeConfig().Return(&config.NodeOfCodeConfig{
SupportThirdPartModules: []string{"httpx", "numpy"},
}).AnyTimes()
mockey.Mock(workflow.GetRepository).Return(mockRepo).Build()
mockCodeRunner.EXPECT().Run(gomock.Any(), gomock.Any()).Return(&coderunner.RunResponse{
Result: map[string]any{
"key0": "value0",

2
backend/domain/workflow/internal/compose/test/question_answer_test.go
View File

@ -104,7 +104,7 @@ func TestQuestionAnswer(t *testing.T) {
mockTos := storageMock.NewMockStorage(ctrl)
mockTos.EXPECT().GetObjectUrl(gomock.Any(), gomock.Any(), gomock.Any()).Return("", nil).AnyTimes()
repo := repo2.NewRepository(mockIDGen, db, redisClient, mockTos,
checkpoint.NewRedisStore(redisClient), nil)
checkpoint.NewRedisStore(redisClient), nil, nil)
mockey.Mock(workflow.GetRepository).Return(repo).Build()
t.Run("answer directly, no structured output", func(t *testing.T) {

14
backend/domain/workflow/internal/nodes/code/code.go
View File

@ -26,10 +26,12 @@ import (
"golang.org/x/exp/maps"
code2 "github.com/coze-dev/coze-studio/backend/crossdomain/impl/code"
wf "github.com/coze-dev/coze-studio/backend/domain/workflow"
"github.com/coze-dev/coze-studio/backend/domain/workflow/entity"
"github.com/coze-dev/coze-studio/backend/domain/workflow/internal/canvas/convert"
"github.com/coze-dev/coze-studio/backend/domain/workflow/internal/schema"
"github.com/coze-dev/coze-studio/backend/infra/contract/coderunner"
"github.com/coze-dev/coze-studio/backend/pkg/lang/slices"
"github.com/coze-dev/coze-studio/backend/domain/workflow/entity/vo"
"github.com/coze-dev/coze-studio/backend/domain/workflow/internal/nodes"
@ -107,15 +109,6 @@ var pythonBuiltinBlacklist = map[string]struct{}{
"tty": {},
}
// pythonThirdPartyWhitelist is the whitelist of python third-party modules,
// see: https://www.coze.cn/open/docs/guides/code_node#7f41f073
// If you want to use other third-party libraries, you can add them to this whitelist.
// And you also need to install them in `/scripts/setup/python.sh` and `/backend/Dockerfile` via `pip install`.
var pythonThirdPartyWhitelist = map[string]struct{}{
"httpx": {},
"numpy": {},
}
type Config struct {
Code string
Language coderunner.Language
@ -192,6 +185,9 @@ func validatePythonImports(code string) error {
imports := parsePythonImports(code)
importErrors := make([]string, 0)
pythonThirdPartyWhitelist := slices.ToMap(wf.GetRepository().GetNodeOfCodeConfig().GetSupportThirdPartModules(), func(e string) (string, bool) {
return e, true
})
var blacklistedModules []string
var nonWhitelistedModules []string
for _, imp := range imports {

6
backend/domain/workflow/internal/repo/repository.go
View File

@ -69,10 +69,11 @@ type RepositoryImpl struct {
workflow.CancelSignalStore
workflow.ExecuteHistoryStore
builtinModel cm.BaseChatModel
workflow.WorkflowConfig
}
func NewRepository(idgen idgen.IDGenerator, db *gorm.DB, redis cache.Cmdable, tos storage.Storage,
cpStore einoCompose.CheckPointStore, chatModel cm.BaseChatModel) workflow.Repository {
cpStore einoCompose.CheckPointStore, chatModel cm.BaseChatModel, workflowConfig workflow.WorkflowConfig) workflow.Repository {
return &RepositoryImpl{
IDGenerator: idgen,
query: query.Use(db),
@ -89,7 +90,8 @@ func NewRepository(idgen idgen.IDGenerator, db *gorm.DB, redis cache.Cmdable, to
query: query.Use(db),
redis: redis,
},
builtinModel: chatModel,
builtinModel: chatModel,
WorkflowConfig: workflowConfig,
}
}

7
backend/domain/workflow/service/service_impl.go
View File

@ -22,13 +22,12 @@ import (
"fmt"
"strconv"
einoCompose "github.com/cloudwego/eino/compose"
"github.com/spf13/cast"
"golang.org/x/exp/maps"
"golang.org/x/sync/errgroup"
"gorm.io/gorm"
einoCompose "github.com/cloudwego/eino/compose"
"github.com/coze-dev/coze-studio/backend/api/model/crossdomain/plugin"
workflowModel "github.com/coze-dev/coze-studio/backend/api/model/crossdomain/workflow"
cloudworkflow "github.com/coze-dev/coze-studio/backend/api/model/workflow"
@ -72,8 +71,8 @@ func NewWorkflowService(repo workflow.Repository) workflow.Service {
}
func NewWorkflowRepository(idgen idgen.IDGenerator, db *gorm.DB, redis cache.Cmdable, tos storage.Storage,
cpStore einoCompose.CheckPointStore, chatModel chatmodel.BaseChatModel) workflow.Repository {
return repo.NewRepository(idgen, db, redis, tos, cpStore, chatModel)
cpStore einoCompose.CheckPointStore, chatModel chatmodel.BaseChatModel, workflowConfig workflow.WorkflowConfig) workflow.Repository {
return repo.NewRepository(idgen, db, redis, tos, cpStore, chatModel, workflowConfig)
}
func (i *impl) ListNodeMeta(_ context.Context, nodeTypes map[entity.NodeType]bool) (map[string][]*entity.NodeTypeMeta, []entity.Category, error) {

71
backend/internal/mock/domain/workflow/interface.go
View File

@ -35,6 +35,7 @@ import (
workflow "github.com/coze-dev/coze-studio/backend/api/model/crossdomain/workflow"
workflow0 "github.com/coze-dev/coze-studio/backend/api/model/workflow"
workflow1 "github.com/coze-dev/coze-studio/backend/domain/workflow"
config "github.com/coze-dev/coze-studio/backend/domain/workflow/config"
entity "github.com/coze-dev/coze-studio/backend/domain/workflow/entity"
vo "github.com/coze-dev/coze-studio/backend/domain/workflow/entity/vo"
gomock "go.uber.org/mock/gomock"
@ -65,47 +66,47 @@ func (m *MockService) EXPECT() *MockServiceMockRecorder {
}
// AsyncExecute mocks base method.
func (m *MockService) AsyncExecute(ctx context.Context, config workflow.ExecuteConfig, input map[string]any) (int64, error) {
func (m *MockService) AsyncExecute(ctx context.Context, arg1 workflow.ExecuteConfig, input map[string]any) (int64, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AsyncExecute", ctx, config, input)
ret := m.ctrl.Call(m, "AsyncExecute", ctx, arg1, input)
ret0, _ := ret[0].(int64)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// AsyncExecute indicates an expected call of AsyncExecute.
func (mr *MockServiceMockRecorder) AsyncExecute(ctx, config, input any) *gomock.Call {
func (mr *MockServiceMockRecorder) AsyncExecute(ctx, arg1, input any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AsyncExecute", reflect.TypeOf((*MockService)(nil).AsyncExecute), ctx, config, input)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AsyncExecute", reflect.TypeOf((*MockService)(nil).AsyncExecute), ctx, arg1, input)
}
// AsyncExecuteNode mocks base method.
func (m *MockService) AsyncExecuteNode(ctx context.Context, nodeID string, config workflow.ExecuteConfig, input map[string]any) (int64, error) {
func (m *MockService) AsyncExecuteNode(ctx context.Context, nodeID string, arg2 workflow.ExecuteConfig, input map[string]any) (int64, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AsyncExecuteNode", ctx, nodeID, config, input)
ret := m.ctrl.Call(m, "AsyncExecuteNode", ctx, nodeID, arg2, input)
ret0, _ := ret[0].(int64)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// AsyncExecuteNode indicates an expected call of AsyncExecuteNode.
func (mr *MockServiceMockRecorder) AsyncExecuteNode(ctx, nodeID, config, input any) *gomock.Call {
func (mr *MockServiceMockRecorder) AsyncExecuteNode(ctx, nodeID, arg2, input any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AsyncExecuteNode", reflect.TypeOf((*MockService)(nil).AsyncExecuteNode), ctx, nodeID, config, input)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AsyncExecuteNode", reflect.TypeOf((*MockService)(nil).AsyncExecuteNode), ctx, nodeID, arg2, input)
}
// AsyncResume mocks base method.
func (m *MockService) AsyncResume(ctx context.Context, req *entity.ResumeRequest, config workflow.ExecuteConfig) error {
func (m *MockService) AsyncResume(ctx context.Context, req *entity.ResumeRequest, arg2 workflow.ExecuteConfig) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AsyncResume", ctx, req, config)
ret := m.ctrl.Call(m, "AsyncResume", ctx, req, arg2)
ret0, _ := ret[0].(error)
return ret0
}
// AsyncResume indicates an expected call of AsyncResume.
func (mr *MockServiceMockRecorder) AsyncResume(ctx, req, config any) *gomock.Call {
func (mr *MockServiceMockRecorder) AsyncResume(ctx, req, arg2 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AsyncResume", reflect.TypeOf((*MockService)(nil).AsyncResume), ctx, req, config)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AsyncResume", reflect.TypeOf((*MockService)(nil).AsyncResume), ctx, req, arg2)
}
// Cancel mocks base method.
@ -368,18 +369,18 @@ func (mr *MockServiceMockRecorder) QueryNodeProperties(ctx, id any) *gomock.Call
}
// ReleaseApplicationWorkflows mocks base method.
func (m *MockService) ReleaseApplicationWorkflows(ctx context.Context, appID int64, config *vo.ReleaseWorkflowConfig) ([]*vo.ValidateIssue, error) {
func (m *MockService) ReleaseApplicationWorkflows(ctx context.Context, appID int64, arg2 *vo.ReleaseWorkflowConfig) ([]*vo.ValidateIssue, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ReleaseApplicationWorkflows", ctx, appID, config)
ret := m.ctrl.Call(m, "ReleaseApplicationWorkflows", ctx, appID, arg2)
ret0, _ := ret[0].([]*vo.ValidateIssue)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// ReleaseApplicationWorkflows indicates an expected call of ReleaseApplicationWorkflows.
func (mr *MockServiceMockRecorder) ReleaseApplicationWorkflows(ctx, appID, config any) *gomock.Call {
func (mr *MockServiceMockRecorder) ReleaseApplicationWorkflows(ctx, appID, arg2 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReleaseApplicationWorkflows", reflect.TypeOf((*MockService)(nil).ReleaseApplicationWorkflows), ctx, appID, config)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReleaseApplicationWorkflows", reflect.TypeOf((*MockService)(nil).ReleaseApplicationWorkflows), ctx, appID, arg2)
}
// Save mocks base method.
@ -397,39 +398,39 @@ func (mr *MockServiceMockRecorder) Save(ctx, id, arg2 any) *gomock.Call {
}
// StreamExecute mocks base method.
func (m *MockService) StreamExecute(ctx context.Context, config workflow.ExecuteConfig, input map[string]any) (*schema.StreamReader[*entity.Message], error) {
func (m *MockService) StreamExecute(ctx context.Context, arg1 workflow.ExecuteConfig, input map[string]any) (*schema.StreamReader[*entity.Message], error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "StreamExecute", ctx, config, input)
ret := m.ctrl.Call(m, "StreamExecute", ctx, arg1, input)
ret0, _ := ret[0].(*schema.StreamReader[*entity.Message])
ret1, _ := ret[1].(error)
return ret0, ret1
}
// StreamExecute indicates an expected call of StreamExecute.
func (mr *MockServiceMockRecorder) StreamExecute(ctx, config, input any) *gomock.Call {
func (mr *MockServiceMockRecorder) StreamExecute(ctx, arg1, input any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StreamExecute", reflect.TypeOf((*MockService)(nil).StreamExecute), ctx, config, input)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StreamExecute", reflect.TypeOf((*MockService)(nil).StreamExecute), ctx, arg1, input)
}
// StreamResume mocks base method.
func (m *MockService) StreamResume(ctx context.Context, req *entity.ResumeRequest, config workflow.ExecuteConfig) (*schema.StreamReader[*entity.Message], error) {
func (m *MockService) StreamResume(ctx context.Context, req *entity.ResumeRequest, arg2 workflow.ExecuteConfig) (*schema.StreamReader[*entity.Message], error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "StreamResume", ctx, req, config)
ret := m.ctrl.Call(m, "StreamResume", ctx, req, arg2)
ret0, _ := ret[0].(*schema.StreamReader[*entity.Message])
ret1, _ := ret[1].(error)
return ret0, ret1
}
// StreamResume indicates an expected call of StreamResume.
func (mr *MockServiceMockRecorder) StreamResume(ctx, req, config any) *gomock.Call {
func (mr *MockServiceMockRecorder) StreamResume(ctx, req, arg2 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StreamResume", reflect.TypeOf((*MockService)(nil).StreamResume), ctx, req, config)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StreamResume", reflect.TypeOf((*MockService)(nil).StreamResume), ctx, req, arg2)
}
// SyncExecute mocks base method.
func (m *MockService) SyncExecute(ctx context.Context, config workflow.ExecuteConfig, input map[string]any) (*entity.WorkflowExecution, vo.TerminatePlan, error) {
func (m *MockService) SyncExecute(ctx context.Context, arg1 workflow.ExecuteConfig, input map[string]any) (*entity.WorkflowExecution, vo.TerminatePlan, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "SyncExecute", ctx, config, input)
ret := m.ctrl.Call(m, "SyncExecute", ctx, arg1, input)
ret0, _ := ret[0].(*entity.WorkflowExecution)
ret1, _ := ret[1].(vo.TerminatePlan)
ret2, _ := ret[2].(error)
@ -437,9 +438,9 @@ func (m *MockService) SyncExecute(ctx context.Context, config workflow.ExecuteCo
}
// SyncExecute indicates an expected call of SyncExecute.
func (mr *MockServiceMockRecorder) SyncExecute(ctx, config, input any) *gomock.Call {
func (mr *MockServiceMockRecorder) SyncExecute(ctx, arg1, input any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncExecute", reflect.TypeOf((*MockService)(nil).SyncExecute), ctx, config, input)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncExecute", reflect.TypeOf((*MockService)(nil).SyncExecute), ctx, arg1, input)
}
// SyncRelatedWorkflowResources mocks base method.
@ -923,6 +924,20 @@ func (mr *MockRepositoryMockRecorder) GetNodeExecutionsByWfExeID(ctx, wfExeID an
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNodeExecutionsByWfExeID", reflect.TypeOf((*MockRepository)(nil).GetNodeExecutionsByWfExeID), ctx, wfExeID)
}
// GetNodeOfCodeConfig mocks base method.
func (m *MockRepository) GetNodeOfCodeConfig() *config.NodeOfCodeConfig {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetNodeOfCodeConfig")
ret0, _ := ret[0].(*config.NodeOfCodeConfig)
return ret0
}
// GetNodeOfCodeConfig indicates an expected call of GetNodeOfCodeConfig.
func (mr *MockRepositoryMockRecorder) GetNodeOfCodeConfig() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNodeOfCodeConfig", reflect.TypeOf((*MockRepository)(nil).GetNodeOfCodeConfig))
}
// GetTestRunLatestExeID mocks base method.
func (m *MockRepository) GetTestRunLatestExeID(ctx context.Context, wfID, uID int64) (int64, error) {
m.ctrl.T.Helper()