diff --git a/pipe/README.md b/pipe/README.md index 7ff97648..caa53496 100644 --- a/pipe/README.md +++ b/pipe/README.md @@ -1,5 +1,80 @@ # Pipe -**Warning: This function under development and is not yet publicly available.** +Pipe Widget is used for complex logic orchestration, serving as an alternative to Flow. -A new workflow orchestration engine designed to solve complex workflow orchestration problems. This is an alternative to **Flow**. +**Warning**: + +Pipe Widget is an experimental feature and not recommended for production use. + +**Usage Scenario** + +Generating DSL from a graphical interface, implementing simple functional logic extensions on the application side. + +## DSL + +CLI: https://github.com/YaoApp/yao-dev-app/blob/main/pipes/cli/translator.pip.yao + +WEB: https://github.com/YaoApp/yao-dev-app/blob/main/pipes/web/translator.pip.yao + +## Node Types + +| Type | options | Description | +| ----------- | ---------------------------- | --------------------------------------------------------- | +| Yao Process | `name`, `args` | run yao process | +| Switch | | conditional branch | +| AI | `prompts`, `model`, `option` | AI interface | +| Request | | HTTP request (not supported yet, use yao process instead) | +| User Input | `ui` (cli/web/...) | user input interface | + +for more details, refer to the DSL demo. + +## Process + +Refer to unit test programs for examples. + +### pipes. + +Run Pipe + +```bash +yao run pipes. [args...] +``` + +If interrupted by user input interface, it returns a context ID for resuming execution. + +### pipe.run + +Run Pipe, equivalent to `pipes.` + +```bash +yao run pipe.run [args...] +``` + +### pipe.create + +Pass DSL text to create and run Pipe + +```bash +yao run pipe.create [args...] +``` + +### pipe.Resume + +Resume execution, used for context restoration + +```bash +yao run pipe.Resume [args...] +``` + +### pipe.Close + +Close Pipe + +```bash +yao run pipe.Close +``` + +## Todo + +[] Progress report for hook integration +[] Support for Http Request Node diff --git a/pipe/pipe.go b/pipe/pipe.go index a88f1256..5612c16d 100644 --- a/pipe/pipe.go +++ b/pipe/pipe.go @@ -43,14 +43,14 @@ func Load(cfg config.Config) error { // New create Pipe func New(source []byte) (*Pipe, error) { pipe := Pipe{} - err := application.Parse("", source, &pipe) + err := application.Parse(".yao", source, &pipe) if err != nil { - return nil, err + return nil, fmt.Errorf("parse pipe: %s", err) } err = (&pipe).build() if err != nil { - return nil, err + return nil, fmt.Errorf("build pipe: %s", err) } return &pipe, nil diff --git a/pipe/process.go b/pipe/process.go index c691465f..38afd065 100644 --- a/pipe/process.go +++ b/pipe/process.go @@ -9,6 +9,7 @@ func init() { process.Register("pipes", processPipes) process.RegisterGroup("pipe", map[string]process.Handler{ "run": processRun, + "create": processCreate, "resume": processResume, "close": processClose, }) @@ -22,14 +23,26 @@ func processPipes(process *process.Process) interface{} { exception.New("pipes.%s not loaded", 404, process.ID).Throw() return nil } - ctx := pipe.Create().WithGlobal(process.Global).WithSid(process.Sid) - res, err := ctx.Exec(process.Args...) + return ctx.Run(process.Args...) +} + +// processCreate process the create pipe.create [...args] +func processCreate(process *process.Process) interface{} { + process.ValidateArgNums(1) + dsl := process.ArgsString(0) + args := []any{} + if len(process.Args) > 1 { + args = process.Args[1:] + } + + pipe, err := New([]byte(dsl)) if err != nil { exception.New(err.Error(), 500).Throw() } - return res + ctx := pipe.Create().WithGlobal(process.Global).WithSid(process.Sid) + return ctx.Run(args...) } // processRun process the resume pipe.run [...args] diff --git a/pipe/process_test.go b/pipe/process_test.go index 563cddc1..cee81e23 100644 --- a/pipe/process_test.go +++ b/pipe/process_test.go @@ -53,6 +53,39 @@ func TestProcessRun(t *testing.T) { assert.Len(t, res.Get("switch"), 2) } +func TestProcessCreate(t *testing.T) { + + prepare(t) + defer test.Clean() + + dsl := `{ + "whitelist": ["utils.fmt.Print"], + "name": "test", + "label": "Test", + "nodes": [ + { + "name": "print", + "process": {"name":"utils.fmt.Print", "args": "{{ $in }}"}, + "output": "print" + } + ], + "output": {"input": "{{ $input }}" } + }` + + p, err := process.Of("pipe.Create", dsl, "hello world") + if err != nil { + t.Fatal(err) + } + + output, err := p.Exec() + if err != nil { + t.Fatal(err) + } + + res := any.Of(output).Map().MapStrAny.Dot() + assert.Equal(t, "hello world", res.Get("input[0]")) +} + func TestProcessResume(t *testing.T) { prepare(t) defer test.Clean()