Merge pull request #577 from trheyi/main

[add] Pipe Widget Can be used as an alternative to Flow

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.<Widget.ID>
+
+Run Pipe
+
+```bash
+yao run pipes.<Widget.ID> [args...]
+```
+
+If interrupted by user input interface, it returns a context ID for resuming execution.
+
+### pipe.Run
+
+Run Pipe, equivalent to `pipes.<Widget.ID>`
+
+```bash
+yao run pipe.run <Widget.ID> [args...]
+```
+
+### pipe.Create
+
+Pass DSL text to create and run Pipe
+
+```bash
+yao run pipe.create <DSL> [args...]
+```
+
+### pipe.Resume
+
+Resume execution, used for context restoration
+
+```bash
+yao run pipe.Resume <Context.ID> [args...]
+```
+
+### pipe.Close
+
+Close Pipe
+
+```bash
+yao run pipe.Close <Context.ID>
+```
+
+## 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

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

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 {
-	case "web":
+			return nil, false, err
+		}
+		return output, false, nil
 
+	default:
+		input, err := ctx.parseNodeInput(node, input)
+		if err != nil {
+			return nil, true, err
 		}
 
-	return nil, fmt.Errorf("pipe: %s %s", ctx.Name, "node type error")
+		return ResumeContext{
+			ID:    ctx.id,
+			Input: input,
+			Node:  node,
+			Data:  ctx.data(node),
+			Type:  node.Type,
+			UI:    node.UI,
+		}, true, nil
+
+	}
 }
 
 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
 }

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>", source, &pipe)
+	err := application.Parse("<source>.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

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},

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 <pipe.id> [...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 <pipe.id> [...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 <id> [...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 <id>
+func processClose(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+	id := process.ArgsString(0)
+	Close(id)
+	return nil
 }

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")
+}

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"`

pipe/ui/web/web.go

@@ -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
-}