From 35bf78b08d1791658465b8bd9940b5837d19af0a Mon Sep 17 00:00:00 2001 From: Max Date: Tue, 13 Feb 2024 20:54:25 +0800 Subject: [PATCH 1/4] [add] pipe widget (dev) --- .gitignore | 1 + pipe/context.go | 219 +++++++++++++++++++++++++++++++++-- pipe/expression.go | 74 ++++++++++++ pipe/interface.go | 1 - pipe/json.go | 26 +++++ pipe/node.go | 276 +++++++++++++++++++++++++++++++++++++++++++++ pipe/pipe.go | 100 ++++++++++++++++ pipe/pipe_test.go | 6 +- pipe/process.go | 28 +++++ pipe/types.go | 57 +++++++--- pipe/ui/cli/cli.go | 62 ++++++++++ pipe/ui/web/web.go | 12 ++ 12 files changed, 834 insertions(+), 28 deletions(-) create mode 100644 pipe/expression.go delete mode 100644 pipe/interface.go create mode 100644 pipe/node.go create mode 100644 pipe/process.go create mode 100644 pipe/ui/cli/cli.go create mode 100644 pipe/ui/web/web.go diff --git a/.gitignore b/.gitignore index 3b4f8a2f..6dc8a3fa 100644 --- a/.gitignore +++ b/.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 \ No newline at end of file diff --git a/pipe/context.go b/pipe/context.go index 8544c59c..ae7a46aa 100644 --- a/pipe/context.go +++ b/pipe/context.go @@ -14,8 +14,18 @@ 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[string][]any{}, + out: map[string]any{}, + input: map[string][]any{}, + output: map[string]any{}, + } + + if pipe.Nodes != nil { + ctx.current = pipe.Nodes[0].Namespace() + } contexts.Store(id, ctx) return ctx } @@ -48,15 +58,210 @@ func (ctx *Context) ID() string { return ctx.id } -// Exec and return error +// 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) { + node, err := ctx.Current() + if err != nil { + return nil, err + } + + if node.Goto != "" { + next, err := ctx.replaceString(node.Goto) + if err != nil { + return nil, err + } + + if next == "EOF" { + return nil, fmt.Errorf("EOF") + } + + ctx.current = next + return ctx.Current() + } + + next := node.index[len(node.index)-1] + 1 + if next < len(ctx.Nodes) { + ctx.current = ctx.Nodes[next].Namespace() + return ctx.Current() + } + return nil, fmt.Errorf("EOF") +} + +// IsEOF check if the error is EOF +func IsEOF(err error) bool { + return err != nil && err.Error() == "EOF" +} + +// Exec 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) + node, err := ctx.Current() + if err != nil { + return nil, err + } + return ctx.exec(node, args...) +} + +// Exec and return error +func (ctx *Context) exec(node *Node, args ...any) (any, error) { + + switch node.Type { + + case "process": + err := node.ExecProcess(ctx, args) + if err != nil { + return nil, err + } + + case "request": + err := node.ExecRequest(ctx, args) + if err != nil { + return nil, err + } + + case "ai": + err := node.ExecAI(ctx, args) + if err != nil { + return nil, err + } + + case "switch": + err := node.ExecSwitch(ctx, args) + if err != nil { + return nil, err + } + + case "user-input": + err := node.Render(ctx, args) + if err != nil { + return nil, err + } + + default: + return nil, fmt.Errorf("pipe: %s %s", ctx.Name, "node type error") + } + return nil, nil } +func (ctx *Context) replace(value any) (any, error) { + + 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 { + for k, v := range ctx.output { + data[k] = v + } + } + return data, nil + +} + // With with the context func (ctx *Context) With(context context.Context) *Context { ctx.context = context diff --git a/pipe/expression.go b/pipe/expression.go new file mode 100644 index 00000000..153764e9 --- /dev/null +++ b/pipe/expression.go @@ -0,0 +1,74 @@ +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) +} diff --git a/pipe/interface.go b/pipe/interface.go deleted file mode 100644 index a7bad566..00000000 --- a/pipe/interface.go +++ /dev/null @@ -1 +0,0 @@ -package pipe diff --git a/pipe/json.go b/pipe/json.go index 777c6836..465bcf68 100644 --- a/pipe/json.go +++ b/pipe/json.go @@ -100,6 +100,32 @@ func (args Args) UnmarshalJSON(data []byte) error { 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 } diff --git a/pipe/node.go b/pipe/node.go new file mode 100644 index 00000000..bae23015 --- /dev/null +++ b/pipe/node.go @@ -0,0 +1,276 @@ +package pipe + +import ( + "fmt" + "strings" + + "github.com/yaoapp/kun/log" + "github.com/yaoapp/kun/utils" + "github.com/yaoapp/yao/pipe/ui/cli" +) + +// ExecProcess Execute the process +func (node Node) ExecProcess(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] + 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 + } + return err + } + + // Execute the next node + _, err = ctx.exec(next, ctx.output[name]) + if err != nil { + return err + } + return nil +} + +// ExecRequest Execute the request +func (node Node) ExecRequest(ctx *Context, args []any) error { + return nil +} + +// ExecAI Execute the AI +func (node Node) ExecAI(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] + + res := map[string]any{"args": args, "Chinese": "你好", "Arabic": "مرحبا"} + 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 + } + return err + } + + // Execute the next node + _, err = ctx.exec(next, ctx.output[name]) + if err != nil { + return err + } + return nil +} + +// ExecSwitch Execute the switch +func (node Node) ExecSwitch(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] + + data, err := ctx.data() + if err != nil { + return err + } + + section, _ := node.Case["default"] + for stmt := range node.Case { + if stmt == "default" { + continue + } + + 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 + if section == nil { + return fmt.Errorf("pipe: %s %s", ctx.Name, "node case not matched") + } + + // 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 { + + switch node.UI { + + case "cli": + return node.renderCli(ctx, args) + + case "web": + + default: + return fmt.Errorf("pipe: %s %s", ctx.Name, "node ui not supported") + } + + return nil +} + +// Namespace the node namespace +func (node Node) Namespace() string { + name := node.Name + if node.namespace != "" { + 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) + if err != nil { + return err + } + + option := &cli.Option{Label: label} + if node.AutoFill != nil { + + value := fmt.Sprintf("%v", node.AutoFill.Value) + value, err = ctx.replaceString(value) + if value != "" { + if err != nil { + fmt.Println("cmd", err) + return err + } + + if node.AutoFill.Action == "exit" { + value = fmt.Sprintf("%s\nexit()\n", value) + } + option.Reader = strings.NewReader(value) + } + } + + userDataLines, err := cli.New(option).Render(args) + if err != nil { + return err + } + + ctx.out[name] = userDataLines + 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 + } + return err + } + + // Execute the next node + _, err = ctx.exec(next, ctx.output[name]) + if err != nil { + return err + } + + // Next node + return nil +} diff --git a/pipe/pipe.go b/pipe/pipe.go index 0aae8971..76d05933 100644 --- a/pipe/pipe.go +++ b/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,91 @@ func Get(id string) (*Pipe, error) { } return nil, fmt.Errorf("pipe %s not found", id) } + +// 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) +} + +func (pipe *Pipe) _build(namespace string, nodes []Node) error { + + for i, node := range nodes { + if node.Name == "" { + return fmt.Errorf("pipe: %s nodes[%d] name is required", pipe.Name, i) + } + + name := node.Name + if namespace != "" { + 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) + } + + // Set the type of the node + if node.Process != nil { + 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) + } + } + + } else if node.Request != nil { + nodes[i].Type = "request" + + } else if node.Prompts != nil { + nodes[i].Type = "ai" + + } 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" + 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) + } + + } else { + return fmt.Errorf("pipe: %s nodes[%d] process, request, case, prompts or ui is required at least one", pipe.Name, i) + } + } + + return nil +} diff --git a/pipe/pipe_test.go b/pipe/pipe_test.go index 7bcf1ce8..a6d2e237 100644 --- a/pipe/pipe_test.go +++ b/pipe/pipe_test.go @@ -22,15 +22,15 @@ 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()) - - assert.NotPanics(t, func() { ctx.Run() }) + assert.NotPanics(t, func() { + ctx.Run(map[string]interface{}{"placeholder": "translate\nhello world"}) + }) } func prepare(t *testing.T) { diff --git a/pipe/process.go b/pipe/process.go new file mode 100644 index 00000000..b088ac5f --- /dev/null +++ b/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 +} diff --git a/pipe/types.go b/pipe/types.go index 1254666a..de0e0d00 100644 --- a/pipe/types.go +++ b/pipe/types.go @@ -1,6 +1,8 @@ package pipe -import "context" +import ( + "context" +) // Pipe the pipe type Pipe struct { @@ -9,10 +11,13 @@ 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"` + Input Input `json:"input,omitempty"` Whitelist Whitelist `json:"whitelist,omitempty"` // the process whitelist + Goto string `json:"goto,omitempty"` // goto node name / EOF + namespace string // the namespace of the pipe + mapping map[string]*Node // the mapping of the nodes Key:namespace.name Value:index } // Context the Context @@ -22,7 +27,11 @@ type Context struct { context context.Context global map[string]interface{} // $global sid string // $sid - current int // current position + current string // 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 + input map[string][]any // $input the pipe input key:namespace.name Value:[] + output map[string]any // $output the pipe output key:namespace.name Value:any } // Hooks the Hooks @@ -32,16 +41,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 + 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 + Input Input `json:"input,omitempty"` // + Output any `json:"output,omitempty"` // + Goto string `json:"goto,omitempty"` // goto node name / EOF + + 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 @@ -53,12 +68,20 @@ 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 diff --git a/pipe/ui/cli/cli.go b/pipe/ui/cli/cli.go new file mode 100644 index 00000000..372757d2 --- /dev/null +++ b/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.Green("%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 +} diff --git a/pipe/ui/web/web.go b/pipe/ui/web/web.go new file mode 100644 index 00000000..e41d899f --- /dev/null +++ b/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 +} From 864b35ecf1cf06eab08d76e7abf7dff92b3c2247 Mon Sep 17 00:00:00 2001 From: Max Date: Wed, 14 Feb 2024 21:33:04 +0800 Subject: [PATCH 2/4] [add] pipe widget (dev 80%) --- .github/workflows/pr-test.yml | 2 + .github/workflows/unit-test.yml | 2 + pipe/context.go | 317 +++++++++++++------------ pipe/expression.go | 121 ++++++++++ pipe/json.go | 24 +- pipe/node.go | 395 ++++++++++++++++---------------- pipe/pipe.go | 83 +++---- pipe/pipe_test.go | 23 +- pipe/types.go | 58 +++-- pipe/ui/cli/cli.go | 2 +- pipe/utils.go | 26 +++ utils/json/json.go | 29 +++ utils/process.go | 4 + 13 files changed, 674 insertions(+), 412 deletions(-) create mode 100644 pipe/utils.go create mode 100644 utils/json/json.go diff --git a/.github/workflows/pr-test.yml b/.github/workflows/pr-test.yml index 8ff75954..8c7d2817 100644 --- a/.github/workflows/pr-test.yml +++ b/.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" diff --git a/.github/workflows/unit-test.yml b/.github/workflows/unit-test.yml index c8355e72..10c88275 100644 --- a/.github/workflows/unit-test.yml +++ b/.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" diff --git a/pipe/context.go b/pipe/context.go index ae7a46aa..74ae3ce4 100644 --- a/pipe/context.go +++ b/pipe/context.go @@ -15,17 +15,22 @@ var contexts = sync.Map{} func (pipe *Pipe) Create() *Context { id := uuid.NewString() ctx := &Context{ - id: id, - Pipe: pipe, - in: map[string][]any{}, - out: map[string]any{}, - input: map[string][]any{}, - output: map[string]any{}, + id: id, + Pipe: pipe, + in: map[*Node][]any{}, + out: map[*Node]any{}, + history: map[*Node][]Prompt{}, + current: nil, + + input: []any{}, + output: nil, } - if pipe.Nodes != nil { - ctx.current = pipe.Nodes[0].Namespace() + // Set the current node + 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) { - node, err := ctx.Current() - if err != nil { - return nil, err +func (ctx *Context) Next() (*Node, bool, error) { + + if ctx.current == nil { + return nil, true, nil } - if node.Goto != "" { - next, err := ctx.replaceString(node.Goto) + // 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, err + return nil, false, err } if next == "EOF" { - return nil, fmt.Errorf("EOF") + return nil, true, nil } - ctx.current = next - return ctx.Current() + 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 } - next := node.index[len(node.index)-1] + 1 - if next < len(ctx.Nodes) { - ctx.current = ctx.Nodes[next].Namespace() - return ctx.Current() + // continue to the next node + next := ctx.current.index + 1 + if next >= len(ctx.Nodes) { + 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": - err := node.ExecRequest(ctx, args) - if err != nil { - return nil, err - } + // case "request": + // err := node.ExecRequest(ctx, args) + // if err != nil { + // 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 { - for k, v := range ctx.output { - data[k] = v + if ctx.in != nil { + for k, v := range ctx.in { + 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) +} diff --git a/pipe/expression.go b/pipe/expression.go index 153764e9..5ffa825a 100644 --- a/pipe/expression.go +++ b/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} + } +} diff --git a/pipe/json.go b/pipe/json.go index 465bcf68..1dc30f00 100644 --- a/pipe/json.go +++ b/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) diff --git a/pipe/node.go b/pipe/node.go index bae23015..af338002 100644 --- a/pipe/node.go +++ b/pipe/node.go @@ -4,235 +4,246 @@ import ( "fmt" "strings" - "github.com/yaoapp/kun/log" - "github.com/yaoapp/kun/utils" + jsoniter "github.com/json-iterator/go" + "github.com/yaoapp/gou/process" + "github.com/yaoapp/yao/openai" "github.com/yaoapp/yao/pipe/ui/cli" ) -// ExecProcess Execute the process -func (node Node) ExecProcess(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 - } +// 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") } - ctx.input[name] = ctx.in[name] - 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() + input, err := ctx.parseNodeInput(node, input) if err != nil { - if IsEOF(err) { - return nil - } - return err + return nil, err } - // Execute the next node - _, err = ctx.exec(next, ctx.output[name]) - if err != nil { - return 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 + } } - 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 -func (node Node) ExecRequest(ctx *Context, args []any) error { - return 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 } -// ExecAI Execute the AI -func (node Node) ExecAI(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] +// AI Execute the AI input +func (node *Node) AI(ctx *Context, input Input) (any, error) { - res := map[string]any{"args": args, "Chinese": "你好", "Arabic": "مرحبا"} - 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 - } + if node.Prompts == nil || len(node.Prompts) == 0 { + return nil, node.Errorf(ctx, "prompts not found") } - next, err := ctx.Next() + input, err := ctx.parseNodeInput(node, input) if err != nil { - if IsEOF(err) { - return nil - } - return err + return nil, err } - // Execute the next node - _, err = ctx.exec(next, ctx.output[name]) + data := ctx.data(node) + 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) ExecSwitch(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] - - data, err := ctx.data() +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 err + return nil, err } - section, _ := node.Case["default"] - for stmt := range node.Case { - if stmt == "default" { + 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 } - - 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 - } + new = append(new, prompt) } - // Execute the next node - if section == nil { - return fmt.Errorf("pipe: %s %s", ctx.Name, "node case not matched") - } - - // 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 + // update the history + ctx.history[node] = new + return new } // 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) Namespace() string { - name := node.Name - if node.namespace != "" { - name = fmt.Sprintf("%s.%s", node.namespace, name) +func (node *Node) renderCli(ctx *Context, input Input) (any, error) { + input, err := ctx.parseNodeInput(node, input) + if err != nil { + return nil, err } - 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 err + return nil, 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 - 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() + output, err := ctx.parseNodeOutput(node, lines) if err != nil { - if IsEOF(err) { - return nil - } - return err + return nil, err } - - // Execute the next node - _, err = ctx.exec(next, ctx.output[name]) - if err != nil { - return err - } - - // Next node - return nil + 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) } diff --git a/pipe/pipe.go b/pipe/pipe.go index 76d05933..a88f1256 100644 --- a/pipe/pipe.go +++ b/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 - if namespace != "" { - 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] + pipe.Nodes[i].index = i + pipe.mapping[node.Name] = &pipe.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" - - } 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 - } - } - } + pipe.Nodes[i].Type = "ai" + continue } 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 { - return fmt.Errorf("pipe: %s nodes[%d] process, request, case, prompts or ui is required at least one", pipe.Name, i) + } 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 diff --git a/pipe/pipe_test.go b/pipe/pipe_test.go index a6d2e237..8de72c62 100644 --- a/pipe/pipe_test.go +++ b/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() { - ctx.Run(map[string]interface{}{"placeholder": "translate\nhello world"}) - }) + output, err := ctx.Exec(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) diff --git a/pipe/types.go b/pipe/types.go index de0e0d00..f27d339a 100644 --- a/pipe/types.go +++ b/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"` - Input Input `json:"input,omitempty"` + 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: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 - 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 - input map[string][]any // $input the pipe input key:namespace.name Value:[] - output map[string]any // $output the pipe output key:namespace.name Value:any + 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 @@ -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 - Input Input `json:"input,omitempty"` // - Output any `json:"output,omitempty"` // + 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 - namespace string // the namespace of the node - history []Prompt // history of prompts, this is for the AI and auto merge to the prompts + index int // the index of the node } // 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"` +} diff --git a/pipe/ui/cli/cli.go b/pipe/ui/cli/cli.go index 372757d2..7e2b37a0 100644 --- a/pipe/ui/cli/cli.go +++ b/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() diff --git a/pipe/utils.go b/pipe/utils.go new file mode 100644 index 00000000..98a358f6 --- /dev/null +++ b/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))) +} diff --git a/utils/json/json.go b/utils/json/json.go new file mode 100644 index 00000000..2af2b66e --- /dev/null +++ b/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 +} diff --git a/utils/process.go b/utils/process.go index 2ecde84d..64120443 100644 --- a/utils/process.go +++ b/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) } From 92a06bd5c26cf3bfc36c76bdfd8819a58617dfbd Mon Sep 17 00:00:00 2001 From: Max Date: Wed, 14 Feb 2024 21:41:12 +0800 Subject: [PATCH 3/4] Fix error handling in TestRun function --- pipe/pipe_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pipe/pipe_test.go b/pipe/pipe_test.go index 8de72c62..54ef7e9e 100644 --- a/pipe/pipe_test.go +++ b/pipe/pipe_test.go @@ -32,6 +32,9 @@ func TestRun(t *testing.T) { WithSid(sid) defer Close(ctx.ID()) output, err := ctx.Exec(map[string]interface{}{"placeholder": "translate\nhello world"}) + if err != nil { + t.Fatal(err) + } res := any.Of(output).Map().MapStrAny.Dot() assert.True(t, res.Has("global")) From 14a97da33084c57eecafd543cd25bb7c859ba47a Mon Sep 17 00:00:00 2001 From: Max Date: Wed, 14 Feb 2024 22:01:27 +0800 Subject: [PATCH 4/4] Update API mirror URL and fix test case --- .github/workflows/pr-test.yml | 4 ++-- .github/workflows/unit-test.yml | 4 ++-- pipe/pipe_test.go | 2 ++ utils/json/json.go | 5 ++++- 4 files changed, 10 insertions(+), 5 deletions(-) diff --git a/.github/workflows/pr-test.yml b/.github/workflows/pr-test.yml index 8c7d2817..81035241 100644 --- a/.github/workflows/pr-test.yml +++ b/.github/workflows/pr-test.yml @@ -40,7 +40,7 @@ env: OPENAI_TEST_KEY: ${{ secrets.OPENAI_TEST_KEY }} TEST_MOAPI_SECRET: ${{ secrets.OPENAI_TEST_KEY }} - TEST_MOAPI_MIRROR: https://api.openai.com/v1 + TEST_MOAPI_MIRROR: https://api.openai.com TAB_NAME: "::PET ADMIN" PAGE_SIZE: "20" @@ -167,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 diff --git a/.github/workflows/unit-test.yml b/.github/workflows/unit-test.yml index 10c88275..51261762 100644 --- a/.github/workflows/unit-test.yml +++ b/.github/workflows/unit-test.yml @@ -44,7 +44,7 @@ env: OPENAI_TEST_KEY: ${{ secrets.OPENAI_TEST_KEY }} TEST_MOAPI_SECRET: ${{ secrets.OPENAI_TEST_KEY }} - TEST_MOAPI_MIRROR: https://api.openai.com/v1 + TEST_MOAPI_MIRROR: https://api.openai.com TAB_NAME: "::PET ADMIN" PAGE_SIZE: "20" @@ -59,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 diff --git a/pipe/pipe_test.go b/pipe/pipe_test.go index 54ef7e9e..27a4f20b 100644 --- a/pipe/pipe_test.go +++ b/pipe/pipe_test.go @@ -2,6 +2,7 @@ package pipe import ( "context" + "fmt" "os" "testing" "time" @@ -51,6 +52,7 @@ func TestRun(t *testing.T) { 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/utils/json/json.go b/utils/json/json.go index 2af2b66e..48904231 100644 --- a/utils/json/json.go +++ b/utils/json/json.go @@ -8,8 +8,11 @@ import ( // **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() + 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 {