diff --git a/pipe/README.md b/pipe/README.md index 7ff97648..48df8b86 100644 --- a/pipe/README.md +++ b/pipe/README.md @@ -1,5 +1,84 @@ # 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 +``` + +## Features + +- [x] **Yao Process Node** Support for running yao process +- [x] **Switch Node** Conditional branch +- [x] **AI Node** AI interface +- [x] **User Input Node** User input interface +- [ ] **Request Node** Support for Http Request +- [ ] **Hooks** Progress report for hook integration diff --git a/pipe/context.go b/pipe/context.go index 74ae3ce4..ecf1a308 100644 --- a/pipe/context.go +++ b/pipe/context.go @@ -36,12 +36,12 @@ func (pipe *Pipe) Create() *Context { } // Open the context -func Open(id string) *Context { +func Open(id string) (*Context, error) { ctx, ok := contexts.Load(id) if !ok { - exception.New("pipe: %s not found", 404, id).Throw() + return nil, fmt.Errorf("context %s not found", id) } - return ctx.(*Context) + return ctx.(*Context), nil } // Close the context @@ -49,6 +49,46 @@ func Close(id string) { contexts.Delete(id) } +// Resume the context by id +func (ctx *Context) Resume(id string, args ...any) any { + v, err := ctx.resume(args...) + if err != nil { + exception.New("pipe: %s %s", 500, ctx.Name, err).Throw() + } + return v +} + +// resume the context by id +func (ctx *Context) resume(args ...any) (any, error) { + if ctx.current == nil { + return nil, ctx.Errorf("pipe %s has no nodes", ctx.Name) + } + + node := ctx.current + output, err := ctx.parseNodeOutput(node, args) + if err != nil { + return nil, node.Errorf(ctx, err.Error()) + } + + // Next node + next, eof, err := ctx.next() + if err != nil { + return nil, err + } + + // End of the pipe + if eof { + defer Close(ctx.id) + output, err := ctx.parseOutput() + if err != nil { + return nil, err + } + return output, nil + } + + return ctx.exec(next, anyToInput(output)) +} + // Run the pipe func (ctx *Context) Run(args ...any) any { v, err := ctx.Exec(args...) @@ -58,53 +98,6 @@ func (ctx *Context) Run(args ...any) any { return v } -// ID the context id -func (ctx *Context) ID() string { - return ctx.id -} - -// Next the next node -func (ctx *Context) Next() (*Node, bool, error) { - - if ctx.current == nil { - return nil, true, nil - } - - // if the goto is not empty, then goto the node - if ctx.current.Goto != "" { - data := ctx.data(ctx.current) - next, err := data.replaceString(ctx.current.Goto) - if err != nil { - return nil, false, err - } - - if next == "EOF" { - return nil, true, nil - } - - var has = false - ctx.current, has = ctx.mapping[next] - if !has { - return nil, false, ctx.Errorf("node %s not found", next) - } - return ctx.current, false, nil - } - - // continue to the next node - next := ctx.current.index + 1 - if next >= len(ctx.Nodes) { - return nil, true, nil - } - - ctx.current = &ctx.Nodes[next] - return ctx.current, false, nil -} - -// IsEOF check if the error is EOF -func IsEOF(err error) bool { - return err != nil && err.Error() == "EOF" -} - // Exec this is the entry point of the pipe func (ctx *Context) Exec(args ...any) (any, error) { if ctx.current == nil { @@ -150,17 +143,23 @@ func (ctx *Context) exec(node *Node, input Input) (output any, err error) { } case "user-input": - out, err = node.Render(ctx, input) + var pause bool = false + out, pause, err = node.Render(ctx, input) if err != nil { return nil, err } + // Pause the pipe waiting for user input + if pause { + return out, nil + } + default: return nil, node.Errorf(ctx, "type '%s' not support", node.Type) } // Execute the next node - next, eof, err := ctx.Next() + next, eof, err := ctx.next() if err != nil { return nil, err } @@ -180,6 +179,43 @@ func (ctx *Context) exec(node *Node, input Input) (output any, err error) { return ctx.exec(next, anyToInput(out)) } +// Next the next node +func (ctx *Context) next() (*Node, bool, error) { + + if ctx.current == nil { + return nil, true, nil + } + + // if the goto is not empty, then goto the node + if ctx.current.Goto != "" { + data := ctx.data(ctx.current) + next, err := data.replaceString(ctx.current.Goto) + if err != nil { + return nil, false, err + } + + if next == "EOF" { + return nil, true, nil + } + + var has = false + ctx.current, has = ctx.mapping[next] + if !has { + return nil, false, ctx.Errorf("node %s not found", next) + } + return ctx.current, false, nil + } + + // continue to the next node + next := ctx.current.index + 1 + if next >= len(ctx.Nodes) { + return nil, true, nil + } + + ctx.current = &ctx.Nodes[next] + return ctx.current, false, nil +} + // ParseNodeInput parse the node input func (ctx *Context) parseNodeInput(node *Node, input Input) (Input, error) { ctx.in[node] = input diff --git a/pipe/node.go b/pipe/node.go index af338002..57cbf5a0 100644 --- a/pipe/node.go +++ b/pipe/node.go @@ -209,18 +209,33 @@ func (node *Node) aiMergeHistory(ctx *Context, prompts []Prompt) []Prompt { } // Render Execute the user input -func (node *Node) Render(ctx *Context, input Input) (any, error) { +func (node *Node) Render(ctx *Context, input Input) (any, bool, error) { switch node.UI { case "cli": - return node.renderCli(ctx, input) + output, err := node.renderCli(ctx, input) + if err != nil { + return nil, false, err + } + return output, false, nil - case "web": + default: + input, err := ctx.parseNodeInput(node, input) + if err != nil { + return nil, true, err + } + + return ResumeContext{ + ID: ctx.id, + Input: input, + Node: node, + Data: ctx.data(node), + Type: node.Type, + UI: node.UI, + }, true, nil } - - return nil, fmt.Errorf("pipe: %s %s", ctx.Name, "node type error") } func (node *Node) renderCli(ctx *Context, input Input) (any, error) { @@ -255,12 +270,12 @@ func (node *Node) renderCli(ctx *Context, input Input) (any, error) { lines, err := cli.New(option).Render(input) if err != nil { - return nil, err + return nil, node.Errorf(ctx, err.Error()) } output, err := ctx.parseNodeOutput(node, lines) if err != nil { - return nil, err + return nil, node.Errorf(ctx, err.Error()) } return output, nil } 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/pipe_test.go b/pipe/pipe_test.go index 27a4f20b..90cc5e38 100644 --- a/pipe/pipe_test.go +++ b/pipe/pipe_test.go @@ -2,7 +2,6 @@ package pipe import ( "context" - "fmt" "os" "testing" "time" @@ -15,10 +14,10 @@ import ( "github.com/yaoapp/yao/test" ) -func TestRun(t *testing.T) { +func TestRunCli(t *testing.T) { prepare(t) defer test.Clean() - translator, err := Get("translator") + translator, err := Get("cli.translator") if err != nil { t.Fatal(err) } @@ -31,7 +30,8 @@ func TestRun(t *testing.T) { With(context). WithGlobal(map[string]interface{}{"foo": "bar"}). WithSid(sid) - defer Close(ctx.ID()) + defer Close(ctx.id) + output, err := ctx.Exec(map[string]interface{}{"placeholder": "translate\nhello world"}) if err != nil { t.Fatal(err) @@ -43,16 +43,53 @@ func TestRun(t *testing.T) { assert.True(t, res.Has("output")) assert.True(t, res.Has("sid")) assert.True(t, res.Has("switch")) - assert.Equal(t, "bar", res.Get("global.foo")) assert.Equal(t, "translate\nhello world", res.Get("input[0].placeholder")) assert.Len(t, res.Get("switch"), 2) } +func TestRunWeb(t *testing.T) { + prepare(t) + defer test.Clean() + translator, err := Get("web.translator") + if err != nil { + t.Fatal(err) + } + + sid := session.ID() + context, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + ctx := translator. + Create(). + With(context). + WithGlobal(map[string]interface{}{"foo": "bar"}). + WithSid(sid) + defer Close(ctx.id) + + web := ctx.Run("hello web world") + resume := web.(ResumeContext) + assert.Equal(t, Input{"hello web world"}, resume.Input) + + ctx, err = Open(resume.ID) + if err != nil { + t.Fatal(err) + } + output := ctx.Resume(resume.ID, "translate", "hello web world") + + res := any.Of(output).Map().MapStrAny.Dot() + assert.True(t, res.Has("global")) + assert.True(t, res.Has("input")) + assert.True(t, res.Has("output")) + assert.True(t, res.Has("sid")) + assert.True(t, res.Has("switch")) + assert.Equal(t, "bar", res.Get("global.foo")) + assert.Equal(t, "hello web world", res.Get("input[0]")) + assert.Len(t, res.Get("switch"), 2) +} + func prepare(t *testing.T) { test.Prepare(t, config.Conf) mirror := os.Getenv("TEST_MOAPI_MIRROR") - fmt.Println(mirror) secret := os.Getenv("TEST_MOAPI_SECRET") share.App = share.AppInfo{ Moapi: share.Moapi{Channel: "stable", Mirrors: []string{mirror}, Secret: secret}, diff --git a/pipe/process.go b/pipe/process.go index b088ac5f..38afd065 100644 --- a/pipe/process.go +++ b/pipe/process.go @@ -7,6 +7,12 @@ import ( func init() { process.Register("pipes", processPipes) + process.RegisterGroup("pipe", map[string]process.Handler{ + "run": processRun, + "create": processCreate, + "resume": processResume, + "close": processClose, + }) } // processScripts @@ -17,12 +23,69 @@ 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] +func processRun(process *process.Process) interface{} { + process.ValidateArgNums(1) + pid := process.ArgsString(0) + args := []any{} + if len(process.Args) > 1 { + args = process.Args[1:] + } + pipe, err := Get(pid) + if err != nil { + exception.New("pipes.%s not loaded", 404, process.ID).Throw() + } + + ctx := pipe.Create().WithGlobal(process.Global).WithSid(process.Sid) + return ctx.Run(args...) +} + +// processResume process the resume pipe.resume [...args] +func processResume(process *process.Process) interface{} { + process.ValidateArgNums(1) + id := process.ArgsString(0) + args := []any{} + if len(process.Args) > 1 { + args = process.Args[1:] + } + + ctx, err := Open(id) + if err != nil { + exception.New("pipes.%s not found", 404, id).Throw() + } + + return ctx. + WithGlobal(process.Global). + WithSid(process.Sid). + Resume(id, args...) +} + +// processClose process the close pipe.close +func processClose(process *process.Process) interface{} { + process.ValidateArgNums(1) + id := process.ArgsString(0) + Close(id) + return nil } diff --git a/pipe/process_test.go b/pipe/process_test.go new file mode 100644 index 00000000..cee81e23 --- /dev/null +++ b/pipe/process_test.go @@ -0,0 +1,136 @@ +package pipe + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/gou/process" + "github.com/yaoapp/kun/any" + "github.com/yaoapp/yao/test" +) + +func TestProcessPipes(t *testing.T) { + prepare(t) + defer test.Clean() + + p, err := process.Of("pipes.cli.translator", map[string]interface{}{"placeholder": "translate\nhello world"}) + if err != nil { + t.Fatal(err) + } + + output, err := p.Exec() + res := any.Of(output).Map().MapStrAny.Dot() + assert.True(t, res.Has("global")) + assert.True(t, res.Has("input")) + assert.True(t, res.Has("output")) + assert.True(t, res.Has("sid")) + assert.True(t, res.Has("switch")) + assert.Equal(t, "translate\nhello world", res.Get("input[0].placeholder")) + assert.Len(t, res.Get("switch"), 2) +} + +func TestProcessRun(t *testing.T) { + prepare(t) + defer test.Clean() + + p, err := process.Of("pipe.Run", "cli.translator", map[string]interface{}{"placeholder": "translate\nhello 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.True(t, res.Has("global")) + assert.True(t, res.Has("input")) + assert.True(t, res.Has("output")) + assert.True(t, res.Has("sid")) + assert.True(t, res.Has("switch")) + assert.Equal(t, "translate\nhello world", res.Get("input[0].placeholder")) + 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() + + p, err := process.Of("pipe.Run", "web.translator", "hello web world") + if err != nil { + t.Fatal(err) + } + + web, err := p.Exec() + resume := web.(ResumeContext) + + p, err = process.Of("pipe.Resume", resume.ID, "translate", "hello web world") + output, err := p.Exec() + if err != nil { + t.Fatal(err) + } + + res := any.Of(output).Map().MapStrAny.Dot() + assert.True(t, res.Has("global")) + assert.True(t, res.Has("input")) + assert.True(t, res.Has("output")) + assert.True(t, res.Has("sid")) + assert.True(t, res.Has("switch")) + assert.Equal(t, "hello web world", res.Get("input[0]")) + assert.Len(t, res.Get("switch"), 2) +} + +func TestProcessClose(t *testing.T) { + prepare(t) + defer test.Clean() + + p, err := process.Of("pipe.Run", "web.translator", "hello web world") + if err != nil { + t.Fatal(err) + } + + web, err := p.Exec() + resume := web.(ResumeContext) + + p, err = process.Of("pipe.Close", resume.ID) + p.Exec() + + p, err = process.Of("pipe.Resume", resume.ID, "translate", "hello web world") + _, err = p.Exec() + assert.NotNil(t, err) + assert.Contains(t, err.Error(), "not found") +} diff --git a/pipe/types.go b/pipe/types.go index f27d339a..b25be278 100644 --- a/pipe/types.go +++ b/pipe/types.go @@ -77,6 +77,16 @@ type Args []any // Data data for the template type Data map[string]interface{} +// ResumeContext the resume context +type ResumeContext struct { + ID string `json:"__id"` + Type string `json:"__type"` + UI string `json:"__ui"` + Input Input `json:"input"` + Node *Node `json:"node"` + Data Data `json:"data"` +} + // AutoFill the autofill type AutoFill struct { Value any `json:"value"` diff --git a/pipe/ui/web/web.go b/pipe/ui/web/web.go deleted file mode 100644 index e41d899f..00000000 --- a/pipe/ui/web/web.go +++ /dev/null @@ -1,12 +0,0 @@ -package web - -// Web the web UI -type Web struct{} - -// Option the web option -type Option struct{} - -// Render the Web UI -func (web *Web) Render(args []any, option Option) error { - return nil -}