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