[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/v1
 
   TAB_NAME: "::PET ADMIN"
   PAGE_SIZE: "20"

.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/v1
 
   TAB_NAME: "::PET ADMIN"
   PAGE_SIZE: "20"

pipe/context.go

@@ -15,17 +15,22 @@ var contexts = sync.Map{}
 func (pipe *Pipe) Create() *Context {
 	id := uuid.NewString()
 	ctx := &Context{
-		id:     id,
+		id:      id,
-		Pipe:   pipe,
+		Pipe:    pipe,
-		in:     map[string][]any{},
+		in:      map[*Node][]any{},
-		out:    map[string]any{},
+		out:     map[*Node]any{},
-		input:  map[string][]any{},
+		history: map[*Node][]Prompt{},
-		output: map[string]any{},
+		current: nil,
+
+		input:  []any{},
+		output: nil,
 	}
 
-	if pipe.Nodes != nil {
+	// Set the current node
-		ctx.current = pipe.Nodes[0].Namespace()
+	if pipe.HasNodes() {
+		ctx.current = &pipe.Nodes[0]
 	}
+
 	contexts.Store(id, ctx)
 	return ctx
 }
@@ -58,42 +63,41 @@ func (ctx *Context) ID() string {
 	return ctx.id
 }
 
-// Current the current node
-func (ctx *Context) Current() (*Node, error) {
-	node, has := ctx.mapping[ctx.current]
-	if !has {
-		return nil, fmt.Errorf("pipe: %s %s", ctx.Name, "node not found")
-	}
-	return node, nil
-}
-
 // Next the next node
-func (ctx *Context) Next() (*Node, error) {
+func (ctx *Context) Next() (*Node, bool, error) {
-	node, err := ctx.Current()
+
-	if err != nil {
+	if ctx.current == nil {
-		return nil, err
+		return nil, true, nil
 	}
 
-	if node.Goto != "" {
+	// if the goto is not empty, then goto the node
-		next, err := ctx.replaceString(node.Goto)
+	if ctx.current.Goto != "" {
+		data := ctx.data(ctx.current)
+		next, err := data.replaceString(ctx.current.Goto)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 
 		if next == "EOF" {
-			return nil, fmt.Errorf("EOF")
+			return nil, true, nil
 		}
 
-		ctx.current = next
+		var has = false
-		return ctx.Current()
+		ctx.current, has = ctx.mapping[next]
+		if !has {
+			return nil, false, ctx.Errorf("node %s not found", next)
+		}
+		return ctx.current, false, nil
 	}
 
-	next := node.index[len(node.index)-1] + 1
+	// continue to the next node
-	if next < len(ctx.Nodes) {
+	next := ctx.current.index + 1
-		ctx.current = ctx.Nodes[next].Namespace()
+	if next >= len(ctx.Nodes) {
-		return ctx.Current()
+		return nil, true, nil
 	}
-	return nil, fmt.Errorf("EOF")
+
+	ctx.current = &ctx.Nodes[next]
+	return ctx.current, false, nil
 }
 
 // IsEOF check if the error is EOF
@@ -101,165 +105,176 @@ func IsEOF(err error) bool {
 	return err != nil && err.Error() == "EOF"
 }
 
-// Exec the pipe
+// Exec this is the entry point of the pipe
 func (ctx *Context) Exec(args ...any) (any, error) {
-	node, err := ctx.Current()
+	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(node, args...)
+
+	return ctx.exec(ctx.current, input)
 }
 
 // Exec and return error
-func (ctx *Context) exec(node *Node, args ...any) (any, error) {
+func (ctx *Context) exec(node *Node, input Input) (output any, err error) {
 
+	var out any
 	switch node.Type {
 
 	case "process":
-		err := node.ExecProcess(ctx, args)
+		out, err = node.YaoProcess(ctx, input)
 		if err != nil {
 			return nil, err
 		}
 
-	case "request":
+	// case "request":
-		err := node.ExecRequest(ctx, args)
+	// 	err := node.ExecRequest(ctx, args)
-		if err != nil {
+	// 	if err != nil {
-			return nil, err
+	// 		return nil, err
-		}
+	// 	}
 
 	case "ai":
-		err := node.ExecAI(ctx, args)
+		out, err = node.AI(ctx, input)
 		if err != nil {
 			return nil, err
 		}
 
 	case "switch":
-		err := node.ExecSwitch(ctx, args)
+		out, err = node.Case(ctx, input)
 		if err != nil {
 			return nil, err
 		}
 
 	case "user-input":
-		err := node.Render(ctx, args)
+		out, err = node.Render(ctx, input)
 		if err != nil {
 			return nil, err
 		}
 
 	default:
-		return nil, fmt.Errorf("pipe: %s %s", ctx.Name, "node type error")
+		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) replace(value any) (any, error) {
+func (ctx *Context) data(node *Node) Data {
 
-	switch v := value.(type) {
-	case string:
-		return ctx.replaceAny(v)
-
-	case []any:
-		return ctx.replaceArray(v)
-
-	case map[string]any:
-		return ctx.replaceMap(v)
-
-	case Input:
-		return ctx.replaceArray(v)
-	}
-
-	return value, nil
-}
-
-func (ctx *Context) replaceAny(value string) (any, error) {
-
-	if !IsExpression(value) {
-		return value, nil
-	}
-
-	data, err := ctx.data()
-	if err != nil {
-		return "", err
-	}
-
-	v, err := data.Exec(value)
-	if err != nil {
-		return "", err
-	}
-	return v, nil
-}
-
-// replaceString replace the string
-func (ctx *Context) replaceString(value string) (string, error) {
-
-	if !IsExpression(value) {
-		return value, nil
-	}
-
-	data, err := ctx.data()
-	if err != nil {
-		return "", err
-	}
-
-	v, err := data.ExecString(value)
-	if err != nil {
-		return "", err
-	}
-	return v, nil
-}
-
-func (ctx *Context) replaceMap(value map[string]any) (map[string]any, error) {
-	newValue := map[string]any{}
-	for k, v := range value {
-		res, err := ctx.replace(v)
-		if err != nil {
-			return nil, err
-		}
-		newValue[k] = res
-	}
-	return newValue, nil
-}
-
-func (ctx *Context) replaceArray(value []any) ([]any, error) {
-	newValue := []any{}
-	for _, v := range value {
-		res, err := ctx.replace(v)
-		if err != nil {
-			return nil, err
-		}
-		newValue = append(newValue, res)
-	}
-
-	return newValue, nil
-}
-
-func (ctx *Context) replaceInput(value Input) (Input, error) {
-	return ctx.replaceArray(value)
-}
-
-func (ctx *Context) data() (Data, error) {
-	node, err := ctx.Current()
-	if err != nil {
-		return Data{}, err
-	}
-
-	name := node.Namespace()
 	data := map[string]any{
 		"$sid":    ctx.sid,
 		"$global": ctx.global,
-		"$in":     ctx.in[name],
-		"$out":    ctx.out[name],
 		"$input":  ctx.input,
 		"$output": ctx.output,
 	}
 
-	if ctx.output != nil {
+	if ctx.in != nil {
-		for k, v := range ctx.output {
+		for k, v := range ctx.in {
-			data[k] = v
+			key := fmt.Sprintf("$node.%s.in", k.Name)
+			data[key] = v
 		}
 	}
-	return data, nil
 
+	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
@@ -279,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

@@ -72,3 +72,124 @@ func (data Data) ExecString(stmt string) (string, error) {
 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/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,16 +88,17 @@ 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)

pipe/node.go

@@ -4,235 +4,246 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/yaoapp/kun/log"
+	jsoniter "github.com/json-iterator/go"
-	"github.com/yaoapp/kun/utils"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/yao/openai"
 	"github.com/yaoapp/yao/pipe/ui/cli"
 )
 
-// ExecProcess Execute the process
+// Case Execute the user input
-func (node Node) ExecProcess(ctx *Context, args []any) error {
+func (node *Node) Case(ctx *Context, input Input) (any, error) {
-	var err error
+
-	name := node.Namespace()
+	if node.Switch == nil || len(node.Switch) == 0 {
-	ctx.in[name] = args
+		return nil, node.Errorf(ctx, "switch case not found")
-	if node.Input != nil {
-		ctx.in[name], err = ctx.replaceInput(node.Input)
-		if err != nil {
-			return err
-		}
 	}
 
-	ctx.input[name] = ctx.in[name]
+	input, err := ctx.parseNodeInput(node, input)
-	res := true
-
-	ctx.out[name] = res
-	ctx.output[name] = res
-	if node.Output != nil {
-		ctx.output[name], err = ctx.replace(node.Output)
-		if err != nil {
-			return err
-		}
-	}
-
-	next, err := ctx.Next()
 	if err != nil {
-		if IsEOF(err) {
+		return nil, err
-			return nil
-		}
-		return err
 	}
 
-	// Execute the next node
+	// Find the case
-	_, err = ctx.exec(next, ctx.output[name])
+	var child *Pipe = node.Switch["default"]
-	if err != nil {
+	data := ctx.data(node)
-		return err
+
+	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
+		}
 	}
-	return nil
+
+	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
 }
 
-// ExecRequest Execute the request
+// YaoProcess Execute the Yao Process
-func (node Node) ExecRequest(ctx *Context, args []any) error {
+func (node *Node) YaoProcess(ctx *Context, input Input) (any, error) {
-	return nil
+
+	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
 }
 
-// ExecAI Execute the AI
+// AI Execute the AI input
-func (node Node) ExecAI(ctx *Context, args []any) error {
+func (node *Node) AI(ctx *Context, input Input) (any, error) {
-	var err error
-	name := node.Namespace()
-	ctx.in[name] = args
-	if node.Input != nil {
-		ctx.in[name], err = ctx.replaceInput(node.Input)
-		if err != nil {
-			return err
-		}
-	}
-	ctx.input[name] = ctx.in[name]
 
-	res := map[string]any{"args": args, "Chinese": "你好", "Arabic": "مرحبا"}
+	if node.Prompts == nil || len(node.Prompts) == 0 {
-	ctx.out[name] = res
+		return nil, node.Errorf(ctx, "prompts not found")
-	ctx.output[name] = res
-	if node.Output != nil {
-		ctx.output[name], err = ctx.replace(node.Output)
-		if err != nil {
-			return err
-		}
 	}
 
-	next, err := ctx.Next()
+	input, err := ctx.parseNodeInput(node, input)
 	if err != nil {
-		if IsEOF(err) {
+		return nil, err
-			return nil
-		}
-		return err
 	}
 
-	// Execute the next node
+	data := ctx.data(node)
-	_, err = ctx.exec(next, ctx.output[name])
+	prompts, err := data.replacePrompts(node.Prompts)
 	if err != nil {
-		return err
+		return nil, err
 	}
-	return nil
+	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
 }
 
-// ExecSwitch Execute the switch
+func (node *Node) chatCompletions(ctx *Context, prompts []Prompt, options map[string]interface{}) (any, error) {
-func (node Node) ExecSwitch(ctx *Context, args []any) error {
+	// moapi call
-	var err error
+	ai, err := openai.NewMoapi(node.Model)
-	name := node.Namespace()
-
-	ctx.in[name] = args
-	if node.Input != nil {
-		ctx.in[name], err = ctx.replaceInput(node.Input)
-		if err != nil {
-			return err
-		}
-	}
-	ctx.input[name] = ctx.in[name]
-
-	data, err := ctx.data()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	section, _ := node.Case["default"]
+	response := []string{}
-	for stmt := range node.Case {
+	content := []string{}
-		if stmt == "default" {
+	_, 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)
-		v, err := data.Exec(stmt)
-		if err != nil {
-			log.Warn("pipe: %s %s", ctx.Name, err)
-			continue
-		}
-
-		// If the result is true, then break the loop
-		if match, ok := v.(bool); ok && match {
-			section = node.Case[stmt]
-			break
-		}
 	}
 
-	// Execute the next node
+	// update the history
-	if section == nil {
+	ctx.history[node] = new
-		return fmt.Errorf("pipe: %s %s", ctx.Name, "node case not matched")
+	return new
-	}
-
-	// Execute The Pipe
-	subCtx := section.Create().
-		With(ctx.context).
-		WithGlobal(ctx.global).
-		WithSid(ctx.sid)
-
-	// Copy the input and output
-	for k, v := range ctx.in {
-		subCtx.in[k] = v
-	}
-
-	for k, v := range ctx.input {
-		subCtx.input[k] = v
-	}
-
-	for k, v := range ctx.out {
-		subCtx.out[k] = v
-	}
-
-	for k, v := range ctx.output {
-		subCtx.output[k] = v
-	}
-
-	_, err = subCtx.Exec(ctx.in[name])
-	if err != nil {
-		return err
-	}
-
-	// Merge the output
-	for k, v := range subCtx.out {
-		ctx.out[k] = v
-	}
-
-	for k, v := range subCtx.output {
-		ctx.output[k] = v
-	}
-
-	utils.Dump(name, ctx.output)
-
-	return nil
 }
 
 // Render Execute the user input
-func (node Node) Render(ctx *Context, args []any) error {
+func (node *Node) Render(ctx *Context, input Input) (any, error) {
 
 	switch node.UI {
 
 	case "cli":
-		return node.renderCli(ctx, args)
+		return node.renderCli(ctx, input)
 
 	case "web":
 
-	default:
-		return fmt.Errorf("pipe: %s %s", ctx.Name, "node ui not supported")
 	}
 
-	return nil
+	return nil, fmt.Errorf("pipe: %s %s", ctx.Name, "node type error")
 }
 
-// Namespace the node namespace
+func (node *Node) renderCli(ctx *Context, input Input) (any, error) {
-func (node Node) Namespace() string {
+	input, err := ctx.parseNodeInput(node, input)
-	name := node.Name
+	if err != nil {
-	if node.namespace != "" {
+		return nil, err
-		name = fmt.Sprintf("%s.%s", node.namespace, name)
 	}
-	return name
-}
-
-func (node Node) renderCli(ctx *Context, args []any) error {
-
-	var err error
-	name := node.Namespace()
-
-	ctx.in[name] = args
-	if node.Input != nil {
-		ctx.in[name], err = ctx.replaceInput(node.Input)
-		if err != nil {
-			return err
-		}
-	}
-	ctx.input[name] = ctx.in[name]
 
 	// Set option
-	label, err := ctx.replaceString(node.Label)
+	data := ctx.data(node)
+	label, err := data.replaceString(node.Label)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	option := &cli.Option{Label: label}
 	if node.AutoFill != nil {
 
 		value := fmt.Sprintf("%v", node.AutoFill.Value)
-		value, err = ctx.replaceString(value)
+		value, err = data.replaceString(value)
 		if value != "" {
 			if err != nil {
-				fmt.Println("cmd", err)
+				return nil, err
-				return err
 			}
 
 			if node.AutoFill.Action == "exit" {
@@ -242,35 +253,21 @@ func (node Node) renderCli(ctx *Context, args []any) error {
 		}
 	}
 
-	userDataLines, err := cli.New(option).Render(args)
+	lines, err := cli.New(option).Render(input)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	ctx.out[name] = userDataLines
+	output, err := ctx.parseNodeOutput(node, lines)
-	ctx.output[name] = userDataLines
-	if node.Output != nil {
-		ctx.output[name], err = ctx.replace(node.Output)
-		if err != nil {
-			return err
-		}
-	}
-
-	// Execute the next node
-	next, err := ctx.Next()
 	if err != nil {
-		if IsEOF(err) {
+		return nil, err
-			return nil
-		}
-		return err
 	}
-
+	return output, nil
-	// Execute the next node
+}
-	_, err = ctx.exec(next, ctx.output[name])
+
-	if err != nil {
+// Errorf format the error message
-		return err
+func (node *Node) Errorf(ctx *Context, format string, a ...any) error {
-	}
+	message := fmt.Sprintf(format, a...)
-
+	pid := ctx.Pipe.ID
-	// Next node
+	return fmt.Errorf("pipe: %s nodes[%d](%s) %s (%s)", pid, node.index, node.Name, message, ctx.id)
-	return nil
 }

pipe/pipe.go

@@ -100,43 +100,42 @@ func Get(id string) (*Pipe, error) {
 
 // Build the pipe
 func (pipe *Pipe) build() error {
-	pipe.mapping = map[string]*Node{}
+
 	if pipe.Nodes == nil || len(pipe.Nodes) == 0 {
 		return fmt.Errorf("pipe: %s nodes is required", pipe.Name)
 	}
 
-	return pipe._build("", pipe.Nodes)
+	return pipe._build()
 }
 
-func (pipe *Pipe) _build(namespace string, nodes []Node) error {
+// HasNodes check if the pipe has nodes
+func (pipe *Pipe) HasNodes() bool {
+	return pipe.Nodes != nil && len(pipe.Nodes) > 0
+}
 
-	for i, node := range nodes {
+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)
 		}
 
-		name := node.Name
+		pipe.Nodes[i].index = i
-		if namespace != "" {
+		pipe.mapping[node.Name] = &pipe.Nodes[i]
-			name = namespace + "." + name
-		}
-
-		// Set the index of the node
-		if nodes[i].index == nil {
-			nodes[i].index = []int{}
-		}
-
-		nodes[i].index = append(nodes[i].index, i)
-		nodes[i].namespace = namespace
-		pipe.mapping[name] = &nodes[i]
 
 		// Set the label of the node
 		if node.Label == "" {
-			nodes[i].Label = strings.ToUpper(node.Name)
+			pipe.Nodes[i].Label = strings.ToUpper(node.Name)
 		}
 
 		// Set the type of the node
 		if node.Process != nil {
-			nodes[i].Type = "process"
+			pipe.Nodes[i].Type = "process"
 
 			// Validate the process
 			if node.Process.Name == "" {
@@ -149,38 +148,42 @@ func (pipe *Pipe) _build(namespace string, nodes []Node) error {
 					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 {
-			nodes[i].Type = "request"
+			pipe.Nodes[i].Type = "request"
+			continue
 
 		} else if node.Prompts != nil {
-			nodes[i].Type = "ai"
+			pipe.Nodes[i].Type = "ai"
-
+			continue
-		} else if node.Case != nil {
-			nodes[i].Type = "switch"
-			for _, sub := range node.Case {
-				// Copy the whitelist to the sub pipe
-				sub.Name = fmt.Sprintf("%s.%s", pipe.Name, node.Name)
-				sub.Whitelist = pipe.Whitelist
-				sub.mapping = map[string]*Node{}
-				sub.namespace = node.Name
-				if sub.Nodes != nil && len(sub.Nodes) > 0 {
-					err := sub._build("", sub.Nodes)
-					if err != nil {
-						return err
-					}
-				}
-			}
 
 		} else if node.UI != "" {
-			nodes[i].Type = "user-input"
+			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 {
+		} else if node.Switch != nil {
-			return fmt.Errorf("pipe: %s nodes[%d] process, request, case, prompts or ui is required at least one", pipe.Name, i)
+			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,15 @@ package pipe
 
 import (
 	"context"
+	"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"
 )
 
@@ -28,13 +31,27 @@ func TestRun(t *testing.T) {
 		WithGlobal(map[string]interface{}{"foo": "bar"}).
 		WithSid(sid)
 	defer Close(ctx.ID())
-	assert.NotPanics(t, func() {
+	output, err := ctx.Exec(map[string]interface{}{"placeholder": "translate\nhello world"})
-		ctx.Run(map[string]interface{}{"placeholder": "translate\nhello 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, "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")
+	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/types.go

@@ -11,27 +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    any       `json:"output,omitempty"`    // the pipe output expression
-	Input     Input     `json:"input,omitempty"`
+	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:namespace.name Value:index
+	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 string                 // current position
+	current *Node                  // current position
-	in      map[string][]any       // $in the node input key:namespace.name Value:[]
+
-	out     map[string]any         // $out the node output key:namespace.name Value:any
+	in      map[*Node][]any    // $in the current node input value
-	input   map[string][]any       // $input the pipe input key:namespace.name Value:[]
+	out     map[*Node]any      // $out the current node output value
-	output  map[string]any         // $output the pipe output key:namespace.name Value:any
+	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
@@ -46,17 +52,17 @@ type Node struct {
 	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
-	Case     map[string]*Pipe `json:"case,omitempty"`     // Switch
+	Switch   map[string]*Pipe `json:"case,omitempty"`     // Switch
-	Input    Input            `json:"input,omitempty"`    //
+	Input    Input            `json:"input,omitempty"`    // the node input expression
-	Output   any              `json:"output,omitempty"`   //
+	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
+	index int // the index of the node
-	namespace string   // the namespace of the node
-	history   []Prompt // history of prompts, this is for the AI and auto merge to the prompts
 }
 
 // Whitelist the Whitelist
@@ -87,7 +93,7 @@ type Case struct {
 // Prompt the switch
 type Prompt struct {
 	Role    string `json:"role,omitempty"`
-	Message string `json:"message,omitempty"`
+	Content string `json:"content,omitempty"`
 }
 
 // Process the switch
@@ -98,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

@@ -43,7 +43,7 @@ func (cli *Cli) Render(args []any) ([]string, error) {
 
 	scanner := bufio.NewScanner(cli.option.Reader)
 	var lines []string
-	color.Green("%s", cli.option.Label)
+	color.Blue("%s", cli.option.Label)
 	fmt.Printf("%s", color.WhiteString("> "))
 	for scanner.Scan() {
 		line := scanner.Text()

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,29 @@
+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)
+	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)
 }