Merge pull request #576 from trheyi/main

[add] pipe widget (dev 80%)

.github/workflows/pr-test.yml

@@ -39,6 +39,8 @@ env:
   MONGO_TEST_PASS: "123456"
 
   OPENAI_TEST_KEY: ${{ secrets.OPENAI_TEST_KEY }}
+  TEST_MOAPI_SECRET: ${{ secrets.OPENAI_TEST_KEY }}
+  TEST_MOAPI_MIRROR: https://api.openai.com
 
   TAB_NAME: "::PET ADMIN"
   PAGE_SIZE: "20"
@@ -165,7 +167,7 @@ jobs:
         with:
           repository: yaoapp/page-builder-app
           token: ${{ secrets.YAO_TEST_TOKEN }}
-          path: page-builder-app 
+          path: page-builder-app
 
       - name: Checkout Extension
         uses: actions/checkout@v3

.github/workflows/unit-test.yml

@@ -43,6 +43,8 @@ env:
   MONGO_TEST_PASS: "123456"
 
   OPENAI_TEST_KEY: ${{ secrets.OPENAI_TEST_KEY }}
+  TEST_MOAPI_SECRET: ${{ secrets.OPENAI_TEST_KEY }}
+  TEST_MOAPI_MIRROR: https://api.openai.com
 
   TAB_NAME: "::PET ADMIN"
   PAGE_SIZE: "20"
@@ -57,7 +59,7 @@ env:
   YAO_TEST_APPLICATION: ${{ github.WORKSPACE }}/../app
   YAO_EXTENSION_ROOT: ${{ github.WORKSPACE }}/../extension
   YAO_TEST_BUILDER_APPLICATION: ${{ github.WORKSPACE }}/../page-builder-app
-  
+
   ## Runtime
   YAO_RUNTIME_MIN: 3
   YAO_RUNTIME_MAX: 6

.gitignore

@@ -37,5 +37,6 @@ xgen/v1.0/*
 !xgen/v1.0/index.html
 !xgen/v1.0/umi.js
 !xgen/v1.0/layouts__index.async.js
+!pipe/ui
 *-unit-test
 docker/build/test

pipe/context.go

@@ -14,8 +14,23 @@ var contexts = sync.Map{}
 // Create create new context
 func (pipe *Pipe) Create() *Context {
 	id := uuid.NewString()
-	ctx := &Context{Pipe: pipe, id: uuid.NewString()}
-	ctx.current = 0
+	ctx := &Context{
+		id:      id,
+		Pipe:    pipe,
+		in:      map[*Node][]any{},
+		out:     map[*Node]any{},
+		history: map[*Node][]Prompt{},
+		current: nil,
+
+		input:  []any{},
+		output: nil,
+	}
+
+	// Set the current node
+	if pipe.HasNodes() {
+		ctx.current = &pipe.Nodes[0]
+	}
+
 	contexts.Store(id, ctx)
 	return ctx
 }
@@ -48,15 +63,220 @@ func (ctx *Context) ID() string {
 	return ctx.id
 }
 
-// Exec and return error
+// 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) {
-	fmt.Printf("name: %v\n", ctx.Name)
-	fmt.Printf("global: %v\n", ctx.global)
-	fmt.Printf("sid: %v\n", ctx.sid)
-	fmt.Printf("whitelist: %v\n", ctx.Whitelist)
+	if ctx.current == nil {
+		return nil, ctx.Errorf("pipe %s has no nodes", ctx.Name)
+	}
+
+	input, err := ctx.parseInput(args)
+	if err != nil {
+		return nil, err
+	}
+
+	return ctx.exec(ctx.current, input)
+}
+
+// Exec and return error
+func (ctx *Context) exec(node *Node, input Input) (output any, err error) {
+
+	var out any
+	switch node.Type {
+
+	case "process":
+		out, err = node.YaoProcess(ctx, input)
+		if err != nil {
+			return nil, err
+		}
+
+	// case "request":
+	// 	err := node.ExecRequest(ctx, args)
+	// 	if err != nil {
+	// 		return nil, err
+	// 	}
+
+	case "ai":
+		out, err = node.AI(ctx, input)
+		if err != nil {
+			return nil, err
+		}
+
+	case "switch":
+		out, err = node.Case(ctx, input)
+		if err != nil {
+			return nil, err
+		}
+
+	case "user-input":
+		out, err = node.Render(ctx, input)
+		if err != nil {
+			return nil, err
+		}
+
+	default:
+		return nil, node.Errorf(ctx, "type '%s' not support", node.Type)
+	}
+
+	// Execute the 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
+	}
+
+	// Execute the next node
+	return ctx.exec(next, anyToInput(out))
+}
+
+// ParseNodeInput parse the node input
+func (ctx *Context) parseNodeInput(node *Node, input Input) (Input, error) {
+	ctx.in[node] = input
+	if node.Input != nil && len(node.Input) > 0 {
+		data := ctx.data(node)
+		input, err := data.replaceArray(node.Input)
+		if err != nil {
+			return nil, err
+		}
+		ctx.in[node] = input
+		return input, nil
+	}
+
+	return input, nil
+}
+
+// ParseNodeOutput parse the node output
+func (ctx *Context) parseNodeOutput(node *Node, output any) (any, error) {
+	ctx.out[node] = output
+	if node.Output != nil {
+		data := ctx.data(node)
+		output, err := data.replace(node.Output)
+		if err != nil {
+			return nil, err
+		}
+		ctx.out[node] = output
+		return output, nil
+	}
+
+	return output, nil
+}
+
+// ParseInput parse the pipe input
+func (ctx *Context) parseInput(input Input) (Input, error) {
+	ctx.input = input
+	if ctx.Input != nil && len(ctx.Input) > 0 {
+		data := ctx.data(nil)
+		input, err := data.replaceArray(ctx.Input)
+		if err != nil {
+			return nil, err
+		}
+		ctx.input = input
+		return input, nil
+	}
+	return input, nil
+}
+
+// ParseOutput parse the pipe output
+func (ctx *Context) parseOutput() (any, error) {
+
+	if ctx.Output != nil {
+		data := ctx.data(nil)
+		output, err := data.replace(ctx.Output)
+		if err != nil {
+			return nil, err
+		}
+		ctx.output = output
+		return output, nil
+	}
+
+	if ctx.current != nil {
+		return ctx.out[ctx.current], nil
+	}
+
 	return nil, nil
 }
 
+func (ctx *Context) data(node *Node) Data {
+
+	data := map[string]any{
+		"$sid":    ctx.sid,
+		"$global": ctx.global,
+		"$input":  ctx.input,
+		"$output": ctx.output,
+	}
+
+	if ctx.in != nil {
+		for k, v := range ctx.in {
+			key := fmt.Sprintf("$node.%s.in", k.Name)
+			data[key] = v
+		}
+	}
+
+	if ctx.out != nil {
+		for k, v := range ctx.out {
+			data[k.Name] = v
+		}
+	}
+
+	if node != nil {
+		data["$in"] = ctx.in[node]
+		data["$out"] = ctx.out[node]
+	}
+
+	return data
+}
+
 // With with the context
 func (ctx *Context) With(context context.Context) *Context {
 	ctx.context = context
@@ -74,3 +294,19 @@ func (ctx *Context) WithSid(sid string) *Context {
 	ctx.sid = sid
 	return ctx
 }
+
+func (ctx *Context) inheritance(parent *Context) *Context {
+	ctx.in = parent.in
+	ctx.out = parent.out
+	ctx.history = parent.history
+	ctx.global = parent.global
+	ctx.sid = parent.sid
+	ctx.parent = parent
+	return ctx
+}
+
+// Errorf format the error message
+func (ctx *Context) Errorf(format string, a ...any) error {
+	message := fmt.Sprintf(format, a...)
+	return fmt.Errorf("pipe: %s(%s) %s %s", ctx.Name, ctx.Pipe.ID, ctx.id, message)
+}

pipe/expression.go

@@ -0,0 +1,195 @@
+package pipe
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/expr-lang/expr"
+	"github.com/expr-lang/expr/vm"
+	"github.com/yaoapp/kun/log"
+)
+
+// If set the map value, should keep the space at the end of the statement
+var stmtRe = regexp.MustCompile(`\{\{([\s\S]*?)\}\}`)
+var options = []expr.Option{
+	expr.AllowUndefinedVariables(),
+}
+
+// New create a new expression
+func (data Data) New(stmt string) (*vm.Program, error) {
+
+	stmt = stmtRe.ReplaceAllStringFunc(stmt, func(stmt string) string {
+		matches := stmtRe.FindStringSubmatch(stmt)
+		if len(matches) > 0 {
+			stmt = strings.ReplaceAll(stmt, matches[0], matches[1])
+		}
+		return stmt
+	})
+
+	stmt = strings.TrimSpace(stmt)
+	// ' => ' " => "
+	stmt = strings.ReplaceAll(stmt, "'", "'")
+	stmt = strings.ReplaceAll(stmt, """, "\"")
+	return expr.Compile(stmt, append([]expr.Option{expr.Env(data)}, options...)...)
+}
+
+// Exec exec statement for the template
+func (data Data) Exec(stmt string) (interface{}, error) {
+	program, err := data.New(stmt)
+	if err != nil {
+		log.Warn("pipe: %s %s", stmt, err)
+		return nil, nil
+	}
+
+	v, err := expr.Run(program, data)
+	if err != nil {
+		log.Warn("pipe: %s %s", stmt, err)
+		return nil, nil
+	}
+	return v, nil
+}
+
+// ExecString exec statement for the template
+func (data Data) ExecString(stmt string) (string, error) {
+
+	res, err := data.Exec(stmt)
+	if err != nil {
+		return "", nil
+	}
+
+	if res == nil {
+		return "", nil
+	}
+
+	if v, ok := res.(string); ok {
+		return v, nil
+	}
+	return fmt.Sprintf("%v", res), nil
+}
+
+// IsExpression check if the statement is an expression
+func IsExpression(stmt string) bool {
+	return stmtRe.MatchString(stmt)
+}
+
+func (data Data) replace(value any) (any, error) {
+
+	switch v := value.(type) {
+	case string:
+		return data.replaceAny(v)
+
+	case []any:
+		return data.replaceArray(v)
+
+	case map[string]any:
+		return data.replaceMap(v)
+
+	case Input:
+		return data.replaceArray(v)
+	}
+
+	return value, nil
+}
+
+func (data Data) replacePrompts(prompts []Prompt) ([]Prompt, error) {
+	newPrompts := []Prompt{}
+	for _, prompt := range prompts {
+		content, err := data.replaceString(prompt.Content)
+		if err != nil {
+			return nil, err
+		}
+		role, err := data.replaceString(prompt.Role)
+		if err != nil {
+			return nil, err
+		}
+		prompt.Role = role
+		prompt.Content = content
+		newPrompts = append(newPrompts, prompt)
+	}
+	return newPrompts, nil
+}
+
+func (data Data) replaceAny(value string) (any, error) {
+
+	if !IsExpression(value) {
+		return value, nil
+	}
+
+	v, err := data.Exec(value)
+	if err != nil {
+		return "", err
+	}
+	return v, nil
+}
+
+// replaceString replace the string
+func (data Data) replaceString(value string) (string, error) {
+
+	if !IsExpression(value) {
+		return value, nil
+	}
+
+	v, err := data.ExecString(value)
+	if err != nil {
+		return "", err
+	}
+	return v, nil
+}
+
+func (data Data) replaceMap(value map[string]any) (map[string]any, error) {
+	newValue := map[string]any{}
+	if value == nil {
+		return newValue, nil
+	}
+
+	for k, v := range value {
+		res, err := data.replace(v)
+		if err != nil {
+			return nil, err
+		}
+		newValue[k] = res
+	}
+	return newValue, nil
+}
+
+func (data Data) replaceArray(value []any) ([]any, error) {
+	newValue := []any{}
+	if value == nil {
+		return newValue, nil
+	}
+
+	for _, v := range value {
+		res, err := data.replace(v)
+		if err != nil {
+			return nil, err
+		}
+		newValue = append(newValue, res)
+	}
+
+	return newValue, nil
+}
+
+func (data Data) replaceInput(value Input) (Input, error) {
+	return data.replaceArray(value)
+}
+
+func anyToInput(v any) Input {
+	switch v := v.(type) {
+	case Input:
+		return v
+
+	case []any:
+		return v
+
+	case []string:
+		input := Input{}
+		for _, s := range v {
+			input = append(input, s)
+		}
+		return input
+
+	default:
+		return Input{v}
+	}
+}

pipe/interface.go

@@ -1 +0,0 @@
-package pipe

pipe/json.go

@@ -45,7 +45,7 @@ func (whitelist *Whitelist) UnmarshalJSON(data []byte) error {
 }
 
 // UnmarshalJSON Custom JSON unmarshal function
-func (input Input) UnmarshalJSON(data []byte) error {
+func (input *Input) UnmarshalJSON(data []byte) error {
 
 	var res any
 	err := jsoniter.Unmarshal(data, &res)
@@ -55,16 +55,19 @@ func (input Input) UnmarshalJSON(data []byte) error {
 
 	switch v := res.(type) {
 	case []string:
-		input = []any{}
+		value := []any{}
 		for _, name := range v {
-			input = append(input, name)
+			value = append(value, name)
 		}
+		*input = value
 
 	case []interface{}:
-		input = v
+		value := []any{}
+		*input = value
 
 	case string:
-		input = []any{v}
+		value := []any{v}
+		*input = value
 
 	default:
 		return fmt.Errorf("input type error: %#v", v)
@@ -75,7 +78,7 @@ func (input Input) UnmarshalJSON(data []byte) error {
 }
 
 // UnmarshalJSON Custom JSON unmarshal function
-func (args Args) UnmarshalJSON(data []byte) error {
+func (args *Args) UnmarshalJSON(data []byte) error {
 
 	var res any
 	err := jsoniter.Unmarshal(data, &res)
@@ -85,21 +88,48 @@ func (args Args) UnmarshalJSON(data []byte) error {
 
 	switch v := res.(type) {
 	case []string:
-		args = []any{}
+		values := []any{}
 		for _, name := range v {
-			args = append(args, name)
+			values = append(values, name)
 		}
+		*args = values
 
 	case []interface{}:
-		args = v
+		*args = v
 
 	case string:
-		args = []any{v}
+		*args = []any{v}
 
 	default:
 		return fmt.Errorf("input type error: %#v", v)
 	}
 
+	return nil
+}
+
+// UnmarshalJSON Custom JSON unmarshal function
+func (autoFill *AutoFill) UnmarshalJSON(data []byte) error {
+
+	var res any
+	err := jsoniter.Unmarshal(data, &res)
+	if err != nil {
+		return err
+	}
+
+	switch v := res.(type) {
+
+	case map[string]interface{}:
+		if value, has := v["value"]; has {
+			autoFill.Value = fmt.Sprint(value)
+		}
+		if action, has := v["action"]; has {
+			autoFill.Action = fmt.Sprint(action)
+		}
+
+	default:
+		autoFill.Value = v
+	}
+
 	return nil
 
 }

pipe/node.go

@@ -0,0 +1,273 @@
+package pipe
+
+import (
+	"fmt"
+	"strings"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/yao/openai"
+	"github.com/yaoapp/yao/pipe/ui/cli"
+)
+
+// Case Execute the user input
+func (node *Node) Case(ctx *Context, input Input) (any, error) {
+
+	if node.Switch == nil || len(node.Switch) == 0 {
+		return nil, node.Errorf(ctx, "switch case not found")
+	}
+
+	input, err := ctx.parseNodeInput(node, input)
+	if err != nil {
+		return nil, err
+	}
+
+	// Find the case
+	var child *Pipe = node.Switch["default"]
+	data := ctx.data(node)
+
+	for expr, pip := range node.Switch {
+
+		expr, err := data.replaceString(expr)
+		if err != nil {
+			return nil, err
+		}
+
+		v, err := data.Exec(expr)
+		if err != nil {
+			return nil, err
+		}
+
+		if v == true {
+			child = pip
+		}
+	}
+
+	if child == nil {
+		return nil, node.Errorf(ctx, "switch case not found")
+	}
+
+	// Execute the child pipe
+	var res any = nil
+	subctx := child.Create().inheritance(ctx)
+	if subctx.current != nil {
+		res, err = subctx.Exec(input...)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	output, err := ctx.parseNodeOutput(node, res)
+	if err != nil {
+		return nil, err
+	}
+
+	return output, nil
+}
+
+// YaoProcess Execute the Yao Process
+func (node *Node) YaoProcess(ctx *Context, input Input) (any, error) {
+
+	if node.Process == nil {
+		return nil, node.Errorf(ctx, "process not set")
+	}
+
+	input, err := ctx.parseNodeInput(node, input)
+	if err != nil {
+		return nil, err
+	}
+
+	data := ctx.data(node)
+	args, err := data.replaceArray(node.Process.Args)
+
+	// Execute the process
+	process, err := process.Of(node.Process.Name, args...)
+	if err != nil {
+		return nil, node.Errorf(ctx, err.Error())
+	}
+
+	res, err := process.WithGlobal(ctx.global).WithSID(ctx.sid).Exec()
+	if err != nil {
+		return nil, node.Errorf(ctx, err.Error())
+	}
+
+	output, err := ctx.parseNodeOutput(node, res)
+	if err != nil {
+		return nil, err
+	}
+
+	return output, nil
+}
+
+// AI Execute the AI input
+func (node *Node) AI(ctx *Context, input Input) (any, error) {
+
+	if node.Prompts == nil || len(node.Prompts) == 0 {
+		return nil, node.Errorf(ctx, "prompts not found")
+	}
+
+	input, err := ctx.parseNodeInput(node, input)
+	if err != nil {
+		return nil, err
+	}
+
+	data := ctx.data(node)
+	prompts, err := data.replacePrompts(node.Prompts)
+	if err != nil {
+		return nil, err
+	}
+	prompts = node.aiMergeHistory(ctx, prompts)
+
+	res, err := node.chatCompletions(ctx, prompts, node.Options)
+	if err != nil {
+		return nil, err
+	}
+
+	output, err := ctx.parseNodeOutput(node, res)
+	if err != nil {
+		return nil, err
+	}
+
+	return output, nil
+}
+
+func (node *Node) chatCompletions(ctx *Context, prompts []Prompt, options map[string]interface{}) (any, error) {
+	// moapi call
+	ai, err := openai.NewMoapi(node.Model)
+	if err != nil {
+		return nil, err
+	}
+
+	response := []string{}
+	content := []string{}
+	_, ex := ai.ChatCompletions(promptsToMap(prompts), node.Options, func(data []byte) int {
+
+		// Prograss Hook
+
+		if len(data) > 5 && string(data[:5]) == "data:" {
+			var res ChatCompletionChunk
+			err := jsoniter.Unmarshal(data[5:], &res)
+			if err != nil {
+				return 0
+			}
+			if len(res.Choices) > 0 {
+				response = append(response, res.Choices[0].Delta.Content)
+			}
+		} else {
+			content = append(content, string(data))
+		}
+
+		return 1
+	})
+
+	if ex != nil {
+		return nil, node.Errorf(ctx, "AI error: %s", ex.Message)
+	}
+
+	if (len(response) == 0) && (len(content) > 0) {
+		return nil, node.Errorf(ctx, "AI error: %s", strings.Join(content, ""))
+	}
+
+	raw := strings.Join(response, "")
+
+	// try to parse the response
+	var res any
+	err = jsoniter.UnmarshalFromString(raw, &res)
+	if err != nil {
+		return raw, nil
+	}
+
+	return res, nil
+}
+
+func (node *Node) aiMergeHistory(ctx *Context, prompts []Prompt) []Prompt {
+	if ctx.history == nil {
+		ctx.history = map[*Node][]Prompt{}
+	}
+	if ctx.history[node] == nil {
+		ctx.history = map[*Node][]Prompt{}
+	}
+	new := []Prompt{}
+	saved := map[string]bool{}
+
+	// filter the prompts
+	for _, prompt := range ctx.history[node] {
+		saved[prompt.finger()] = true
+		new = append(new, prompt)
+	}
+
+	for _, prompt := range prompts {
+		if saved[prompt.finger()] {
+			continue
+		}
+		new = append(new, prompt)
+	}
+
+	// update the history
+	ctx.history[node] = new
+	return new
+}
+
+// Render Execute the user input
+func (node *Node) Render(ctx *Context, input Input) (any, error) {
+
+	switch node.UI {
+
+	case "cli":
+		return node.renderCli(ctx, input)
+
+	case "web":
+
+	}
+
+	return nil, fmt.Errorf("pipe: %s %s", ctx.Name, "node type error")
+}
+
+func (node *Node) renderCli(ctx *Context, input Input) (any, error) {
+	input, err := ctx.parseNodeInput(node, input)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set option
+	data := ctx.data(node)
+	label, err := data.replaceString(node.Label)
+	if err != nil {
+		return nil, err
+	}
+
+	option := &cli.Option{Label: label}
+	if node.AutoFill != nil {
+
+		value := fmt.Sprintf("%v", node.AutoFill.Value)
+		value, err = data.replaceString(value)
+		if value != "" {
+			if err != nil {
+				return nil, err
+			}
+
+			if node.AutoFill.Action == "exit" {
+				value = fmt.Sprintf("%s\nexit()\n", value)
+			}
+			option.Reader = strings.NewReader(value)
+		}
+	}
+
+	lines, err := cli.New(option).Render(input)
+	if err != nil {
+		return nil, err
+	}
+
+	output, err := ctx.parseNodeOutput(node, lines)
+	if err != nil {
+		return nil, err
+	}
+	return output, nil
+}
+
+// Errorf format the error message
+func (node *Node) Errorf(ctx *Context, format string, a ...any) error {
+	message := fmt.Sprintf(format, a...)
+	pid := ctx.Pipe.ID
+	return fmt.Errorf("pipe: %s nodes[%d](%s) %s (%s)", pid, node.index, node.Name, message, ctx.id)
+}

pipe/pipe.go

@@ -3,6 +3,7 @@ package pipe
 import (
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/yaoapp/gou/application"
 	"github.com/yaoapp/yao/config"
@@ -46,6 +47,12 @@ func New(source []byte) (*Pipe, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	err = (&pipe).build()
+	if err != nil {
+		return nil, err
+	}
+
 	return &pipe, nil
 }
 
@@ -63,6 +70,11 @@ func NewFile(file string, root string) (*Pipe, error) {
 		return nil, err
 	}
 
+	err = (&pipe).build()
+	if err != nil {
+		return nil, err
+	}
+
 	return &pipe, nil
 }
 
@@ -85,3 +97,94 @@ func Get(id string) (*Pipe, error) {
 	}
 	return nil, fmt.Errorf("pipe %s not found", id)
 }
+
+// Build the pipe
+func (pipe *Pipe) build() error {
+
+	if pipe.Nodes == nil || len(pipe.Nodes) == 0 {
+		return fmt.Errorf("pipe: %s nodes is required", pipe.Name)
+	}
+
+	return pipe._build()
+}
+
+// HasNodes check if the pipe has nodes
+func (pipe *Pipe) HasNodes() bool {
+	return pipe.Nodes != nil && len(pipe.Nodes) > 0
+}
+
+func (pipe *Pipe) _build() error {
+
+	pipe.mapping = map[string]*Node{}
+	if pipe.Nodes == nil {
+		return nil
+	}
+
+	for i, node := range pipe.Nodes {
+		if node.Name == "" {
+			return fmt.Errorf("pipe: %s nodes[%d] name is required", pipe.Name, i)
+		}
+
+		pipe.Nodes[i].index = i
+		pipe.mapping[node.Name] = &pipe.Nodes[i]
+
+		// Set the label of the node
+		if node.Label == "" {
+			pipe.Nodes[i].Label = strings.ToUpper(node.Name)
+		}
+
+		// Set the type of the node
+		if node.Process != nil {
+			pipe.Nodes[i].Type = "process"
+
+			// Validate the process
+			if node.Process.Name == "" {
+				return fmt.Errorf("pipe: %s nodes[%d] process name is required", pipe.Name, i)
+			}
+
+			// Security check
+			if pipe.Whitelist != nil {
+				if _, has := pipe.Whitelist[node.Process.Name]; !has {
+					return fmt.Errorf("pipe: %s nodes[%d] process %s is not in the whitelist", pipe.Name, i, node.Process.Name)
+				}
+			}
+			continue
+
+		} else if node.Request != nil {
+			pipe.Nodes[i].Type = "request"
+			continue
+
+		} else if node.Prompts != nil {
+			pipe.Nodes[i].Type = "ai"
+			continue
+
+		} else if node.UI != "" {
+			pipe.Nodes[i].Type = "user-input"
+			if node.UI != "cli" && node.UI != "web" && node.UI != "app" && node.UI != "wxapp" { // Vaildate the UI type
+				return fmt.Errorf("pipe: %s nodes[%d] the type of the UI must be cli, web, app, wxapp", pipe.Name, i)
+			}
+			continue
+
+		} else if node.Switch != nil {
+			pipe.Nodes[i].Type = "switch"
+			for key, pip := range node.Switch {
+				key = ref(key)
+				pip.Whitelist = pipe.Whitelist // Copy the whitelist
+				pip.namespace = node.Name
+				pip.parent = pipe
+				if pip.ID == "" {
+					pip.ID = fmt.Sprintf("%s.%s#%s", pipe.ID, node.Name, key)
+				}
+				if pip.Name == "" {
+					pip.Name = fmt.Sprintf("%s(%s#%s)", pipe.Name, node.Name, key)
+				}
+				pip._build()
+			}
+			continue
+		}
+
+		return fmt.Errorf("pipe: %s nodes[%d] process, request, case, prompts or ui is required at least one", pipe.Name, i)
+	}
+
+	return nil
+}

pipe/pipe_test.go

@@ -2,12 +2,16 @@ package pipe
 
 import (
 	"context"
+	"fmt"
+	"os"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/yaoapp/gou/session"
+	"github.com/yaoapp/kun/any"
 	"github.com/yaoapp/yao/config"
+	"github.com/yaoapp/yao/share"
 	"github.com/yaoapp/yao/test"
 )
 
@@ -22,19 +26,37 @@ func TestRun(t *testing.T) {
 	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())
+	output, err := ctx.Exec(map[string]interface{}{"placeholder": "translate\nhello world"})
+	if err != nil {
+		t.Fatal(err)
+	}
 
-	assert.NotPanics(t, func() { ctx.Run() })
+	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, "translate\nhello world", res.Get("input[0].placeholder"))
+	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},
+	}
 	err := Load(config.Conf)
 	if err != nil {
 		t.Fatal(err)

pipe/process.go

@@ -0,0 +1,28 @@
+package pipe
+
+import (
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/exception"
+)
+
+func init() {
+	process.Register("pipes", processPipes)
+}
+
+// processScripts
+func processPipes(process *process.Process) interface{} {
+
+	pipe, err := Get(process.ID)
+	if err != nil {
+		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...)
+	if err != nil {
+		exception.New(err.Error(), 500).Throw()
+	}
+
+	return res
+}

pipe/types.go

@@ -1,6 +1,8 @@
 package pipe
 
-import "context"
+import (
+	"context"
+)
 
 // Pipe the pipe
 type Pipe struct {
@@ -9,20 +11,33 @@ type Pipe struct {
 	Nodes     []Node    `json:"nodes"`
 	Label     string    `json:"label,omitempty"`
 	Hooks     *Hooks    `json:"hooks,omitempty"`
-	Output    any       `json:"output,omitempty"`    // $output
-	Input     Input     `json:"input,omitempty"`     // $input
+	Output    any       `json:"output,omitempty"`    // the pipe output expression
+	Input     Input     `json:"input,omitempty"`     // the pipe input expression
 	Whitelist Whitelist `json:"whitelist,omitempty"` // the process whitelist
+	Goto      string    `json:"goto,omitempty"`      // goto node name / EOF
 
+	parent    *Pipe            // the parent pipe
+	namespace string           // the namespace of the pipe
+	mapping   map[string]*Node // the mapping of the nodes Key:name Value:index
 }
 
 // Context the Context
 type Context struct {
 	*Pipe
-	id      string
+	id     string
+	parent *Context // the parent context id
+
 	context context.Context
 	global  map[string]interface{} // $global
 	sid     string                 // $sid
-	current int                    // current position
+	current *Node                  // current position
+
+	in      map[*Node][]any    // $in the current node input value
+	out     map[*Node]any      // $out the current node output value
+	history map[*Node][]Prompt // history of prompts, this is for the AI and auto merge to the prompts of the node
+
+	input  []any // $input the pipe input value
+	output any   // $output the pipe output value
 }
 
 // Hooks the Hooks
@@ -32,16 +47,22 @@ type Hooks struct {
 
 // Node the pip node
 type Node struct {
-	Name      string                 `json:"name"`
-	Type      string                 `json:"type,omitempty"`      // user-input, ai, process, switch, request
-	Label     string                 `json:"label,omitempty"`     // Display
-	Process   *Process               `json:"process,omitempty"`   // Yao Process
-	Prompts   []Prompt               `json:"prompts,omitempty"`   // AI prompts
-	Request   *Request               `json:"request,omitempty"`   // Http Request
-	Interface string                 `json:"interface,omitempty"` // User Interface command-line, web, app, wxapp ...
-	Case      map[string]CaseSection `json:"case,omitempty"`      // Switch
-	Input     Input                  `json:"input,omitempty"`     // $in
-	Output    any                    `json:"output,omitempty"`    // $out
+	Name     string           `json:"name"`
+	Type     string           `json:"type,omitempty"`     // user-input, ai, process, switch, request
+	Label    string           `json:"label,omitempty"`    // Display
+	Process  *Process         `json:"process,omitempty"`  // Yao Process
+	Prompts  []Prompt         `json:"prompts,omitempty"`  // AI prompts
+	Model    string           `json:"model,omitempty"`    // AI model name (optional)
+	Options  map[string]any   `json:"options,omitempty"`  // AI or Request options (optional)
+	Request  *Request         `json:"request,omitempty"`  // Http Request
+	UI       string           `json:"ui,omitempty"`       // The User Interface cli, web, app, wxapp ...
+	AutoFill *AutoFill        `json:"autofill,omitempty"` // Autofill the user input with the expression
+	Switch   map[string]*Pipe `json:"case,omitempty"`     // Switch
+	Input    Input            `json:"input,omitempty"`    // the node input expression
+	Output   any              `json:"output,omitempty"`   // the node output expression
+	Goto     string           `json:"goto,omitempty"`     // goto node name / EOF
+
+	index int // the index of the node
 }
 
 // Whitelist the Whitelist
@@ -53,18 +74,26 @@ type Input []any
 // Args the args
 type Args []any
 
-// CaseSection the switch case section
-type CaseSection struct {
+// Data data for the template
+type Data map[string]interface{}
+
+// AutoFill the autofill
+type AutoFill struct {
+	Value  any    `json:"value"`
+	Action string `json:"action,omitempty"`
+}
+
+// Case the switch case section
+type Case struct {
 	Input  Input  `json:"input,omitempty"`  // $in
 	Output any    `json:"output,omitempty"` // $out
 	Nodes  []Node `json:"nodes,omitempty"`  // $out
-	Goto   string `json:"goto,omitempty"`   // goto node name / EOF
 }
 
 // Prompt the switch
 type Prompt struct {
 	Role    string `json:"role,omitempty"`
-	Message string `json:"message,omitempty"`
+	Content string `json:"content,omitempty"`
 }
 
 // Process the switch
@@ -75,3 +104,23 @@ type Process struct {
 
 // Request the request
 type Request struct{}
+
+// ChatCompletionChunk the chat completion chunk
+type ChatCompletionChunk struct {
+	ID                string      `json:"id"`
+	Object            string      `json:"object"`
+	Created           int64       `json:"created"`
+	Model             string      `json:"model"`
+	SystemFingerprint interface{} `json:"system_fingerprint"`
+	Choices           []struct {
+		Index        int         `json:"index"`
+		Delta        DeltaStruct `json:"delta"`
+		Logprobs     interface{} `json:"logprobs"`
+		FinishReason interface{} `json:"finish_reason"`
+	} `json:"choices"`
+}
+
+// DeltaStruct the delta struct
+type DeltaStruct struct {
+	Content string `json:"content"`
+}

pipe/ui/cli/cli.go

@@ -0,0 +1,62 @@
+package cli
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/fatih/color"
+)
+
+// Cli the CLI
+type Cli struct {
+	option *Option
+}
+
+// In the input stream
+var reader io.Reader = os.Stdin
+
+// Option the CLI option
+type Option struct {
+	Label  string
+	Reader io.Reader
+}
+
+// SetReader set the reader
+func SetReader(r io.Reader) {
+	reader = r
+}
+
+// New create a new CLI
+func New(option *Option) *Cli {
+	if option.Reader == nil {
+		option.Reader = reader
+	}
+	return &Cli{
+		option: option,
+	}
+}
+
+// Render the CLI UI
+func (cli *Cli) Render(args []any) ([]string, error) {
+
+	scanner := bufio.NewScanner(cli.option.Reader)
+	var lines []string
+	color.Blue("%s", cli.option.Label)
+	fmt.Printf("%s", color.WhiteString("> "))
+	for scanner.Scan() {
+		line := scanner.Text()
+		if line == "exit()" {
+			break
+		}
+		lines = append(lines, line)
+		fmt.Printf("%s", color.WhiteString("> "))
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+
+	return lines, nil
+}

pipe/ui/web/web.go

@@ -0,0 +1,12 @@
+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
+}

pipe/utils.go

@@ -0,0 +1,26 @@
+package pipe
+
+import (
+	"crypto/md5"
+	"fmt"
+)
+
+func ref(s string) string {
+	return fmt.Sprintf("%x", md5.Sum([]byte(s)))[:6]
+}
+
+func promptsToMap(prompts []Prompt) []map[string]interface{} {
+	maps := []map[string]interface{}{}
+	for _, prompt := range prompts {
+		maps = append(maps, map[string]interface{}{
+			"role":    prompt.Role,
+			"content": prompt.Content,
+		})
+	}
+	return maps
+}
+
+func (promt Prompt) finger() string {
+	raw := fmt.Sprintf("%s|%s", promt.Role, promt.Content)
+	return fmt.Sprintf("%x", md5.Sum([]byte(raw)))
+}

utils/json/json.go

@@ -0,0 +1,32 @@
+package json
+
+import (
+	"github.com/yaoapp/gou/process"
+)
+
+// ProcessValidate utils.json.Validate
+// **Warning** This process under developing, do not use it
+func ProcessValidate(process *process.Process) interface{} {
+	process.ValidateArgNums(2)
+	if _, ok := process.Args[0].(map[string]interface{}); !ok {
+		return false
+	}
+
+	data := process.ArgsMap(0, map[string]interface{}{}).Dot()
+	rules := process.ArgsRecords(1)
+	for _, rule := range rules {
+		for method, value := range rule {
+			switch method {
+			case "haskey":
+				key, ok := value.(string)
+				if !ok {
+					return false
+				}
+				if !data.Has(key) {
+					return false
+				}
+			}
+		}
+	}
+	return true
+}

utils/process.go

@@ -3,6 +3,7 @@ package utils
 import (
 	"github.com/yaoapp/gou/process"
 	"github.com/yaoapp/yao/utils/datetime"
+	"github.com/yaoapp/yao/utils/json"
 	"github.com/yaoapp/yao/utils/str"
 	"github.com/yaoapp/yao/utils/tree"
 	"github.com/yaoapp/yao/utils/url"
@@ -89,4 +90,7 @@ func Init() {
 	process.Register("utils.url.ParseQuery", url.ProcessParseQuery)
 	process.Register("utils.url.QueryParam", url.ProcessQueryParam)
 	process.Register("utils.url.ParseURL", url.ProcessParseURL)
+
+	// JSON
+	process.Register("utils.json.Validate", json.ProcessValidate)
 }