From fe522c09283c50a62c6f221df3a8194bb469e1d7 Mon Sep 17 00:00:00 2001 From: "shentong.martin" Date: Wed, 6 Aug 2025 21:25:09 +0800 Subject: [PATCH] feat: workflow examples Change-Id: I6ad7882527ba818ff00bea8b9ac76f0fea08951d --- compose/workflow/1_simple/main.go | 68 +++++++++ compose/workflow/2_field_mapping/main.go | 95 +++++++++++++ compose/workflow/3_data_only/main.go | 80 +++++++++++ .../workflow/4_control_only_branch/main.go | 71 ++++++++++ compose/workflow/5_static_values/main.go | 81 +++++++++++ compose/workflow/6_stream_field_map/main.go | 133 ++++++++++++++++++ 6 files changed, 528 insertions(+) create mode 100644 compose/workflow/1_simple/main.go create mode 100644 compose/workflow/2_field_mapping/main.go create mode 100644 compose/workflow/3_data_only/main.go create mode 100644 compose/workflow/4_control_only_branch/main.go create mode 100644 compose/workflow/5_static_values/main.go create mode 100644 compose/workflow/6_stream_field_map/main.go diff --git a/compose/workflow/1_simple/main.go b/compose/workflow/1_simple/main.go new file mode 100644 index 0000000..373c2ee --- /dev/null +++ b/compose/workflow/1_simple/main.go @@ -0,0 +1,68 @@ +/* + * Copyright 2025 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "context" + "strconv" + + "github.com/cloudwego/eino/compose" + + "github.com/cloudwego/eino-examples/internal/logs" +) + +// creates and invokes a simple workflow with only a Lambda node. +// Since all field mappings are ALL to ALL mappings +// (by using AddInput without field mappings), +// this simple workflow is equivalent to a Graph: START -> lambda -> END. +func main() { + // create a Workflow, just like creating a Graph + wf := compose.NewWorkflow[int, string]() + + // add a lambda node to the Workflow, just like adding the lambda to a Graph + wf.AddLambdaNode("lambda", compose.InvokableLambda( + func(ctx context.Context, in int) (string, error) { + return strconv.Itoa(in), nil + })). + // add an input to this lambda node from START. + // this means mapping all output of START to the input of the lambda. + // the effect of AddInput is to set both a control dependency + // and a data dependency. + AddInput(compose.START) + + // obtain the compose.END of the workflow for method chaining + wf.End(). + // add an input to compose.END, + // which means 'using ALL output of lambda node as output of END'. + AddInput("lambda") + + // compile the Workflow, just like compiling a Graph + run, err := wf.Compile(context.Background()) + if err != nil { + logs.Errorf("workflow compile error: %v", err) + return + } + + // invoke the Workflow, just like invoking a Graph + result, err := run.Invoke(context.Background(), 1) + if err != nil { + logs.Errorf("workflow run err: %v", err) + return + } + + logs.Infof("%v", result) +} diff --git a/compose/workflow/2_field_mapping/main.go b/compose/workflow/2_field_mapping/main.go new file mode 100644 index 0000000..f138bd9 --- /dev/null +++ b/compose/workflow/2_field_mapping/main.go @@ -0,0 +1,95 @@ +/* + * Copyright 2025 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "context" + "strings" + + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/schema" + + "github.com/cloudwego/eino-examples/internal/logs" +) + +// demonstrates the field mapping ability of eino workflow. +func main() { + type counter struct { + FullStr string // exported because we will do field mapping for this field + SubStr string // exported because we will do field mapping for this field + } + + // wordCounter is a lambda function that count occurrences of SubStr within FullStr + wordCounter := func(ctx context.Context, c counter) (int, error) { + return strings.Count(c.FullStr, c.SubStr), nil + } + + type message struct { + *schema.Message // exported because we will do field mapping for this field + SubStr string // exported because we will do field mapping for this field + } + + // create a workflow just like a Graph + wf := compose.NewWorkflow[message, map[string]any]() + + // add lambda c1 just like in Graph + wf.AddLambdaNode("c1", compose.InvokableLambda(wordCounter)). + AddInput(compose.START, // add an input from START, specifying 2 field mappings + // map START's SubStr field to lambda c1's SubStr field + compose.MapFields("SubStr", "SubStr"), + // map START's Message's Content field to lambda c1's FullStr field + compose.MapFieldPaths([]string{"Message", "Content"}, []string{"FullStr"})) + + // add lambda c2 just like in Graph + wf.AddLambdaNode("c2", compose.InvokableLambda(wordCounter)). + AddInput(compose.START, // add an input from START, specifying 2 field mappings + // map START's SubStr field to lambda c1's SubStr field + compose.MapFields("SubStr", "SubStr"), + // map START's Message's ReasoningContent field to lambda c1's FullStr field + compose.MapFieldPaths([]string{"Message", "ReasoningContent"}, []string{"FullStr"})) + + wf.End(). // Obtain the compose.END for method chaining + // add an input from c1, + // mapping full output of c1 to the map key 'content_count' + AddInput("c1", compose.ToField("content_count")). + // also add an input from c2, + // mapping full output of c2 to the map key 'reasoning_content_count' + AddInput("c2", compose.ToField("reasoning_content_count")) + + // compile the workflow just like compiling a Graph + run, err := wf.Compile(context.Background()) + if err != nil { + logs.Errorf("workflow compile error: %v", err) + return + } + + // invoke the workflow just like invoking a Graph + result, err := run.Invoke(context.Background(), message{ + Message: &schema.Message{ + Role: schema.Assistant, + Content: "Hello world!", + ReasoningContent: "I need to say something meaningful", + }, + SubStr: "o", // would like to count the occurrences of 'o' + }) + if err != nil { + logs.Errorf("workflow run err: %v", err) + return + } + + logs.Infof("%v", result) +} diff --git a/compose/workflow/3_data_only/main.go b/compose/workflow/3_data_only/main.go new file mode 100644 index 0000000..aa5e7f0 --- /dev/null +++ b/compose/workflow/3_data_only/main.go @@ -0,0 +1,80 @@ +/* + * Copyright 2025 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "context" + + "github.com/cloudwego/eino/compose" + + "github.com/cloudwego/eino-examples/internal/logs" +) + +func main() { + type calculator struct { + Add []int + Multiply int + } + + adder := func(ctx context.Context, in []int) (out int, err error) { + for _, i := range in { + out += i + } + return out, nil + } + + type mul struct { + A int + B int + } + + multiplier := func(ctx context.Context, m mul) (int, error) { + return m.A * m.B, nil + } + + wf := compose.NewWorkflow[calculator, int]() + + wf.AddLambdaNode("adder", compose.InvokableLambda(adder)). + AddInput(compose.START, compose.FromField("Add")) + + wf.AddLambdaNode("mul", compose.InvokableLambda(multiplier)). + AddInput("adder", compose.ToField("A")). + AddInputWithOptions(compose.START, []*compose.FieldMapping{compose.MapFields("Multiply", "B")}, + // use WithNoDirectDependency to declare a 'data-only' dependency, + // in this case, START node's execution status will not determine whether 'mul' node can execute. + // START node only passes one field of its output to 'mul' node. + compose.WithNoDirectDependency()) + + wf.End().AddInput("mul") + + runner, err := wf.Compile(context.Background()) + if err != nil { + logs.Errorf("workflow compile error: %v", err) + return + } + + result, err := runner.Invoke(context.Background(), calculator{ + Add: []int{2, 5}, + Multiply: 3, + }) + if err != nil { + logs.Errorf("workflow run err: %v", err) + return + } + + logs.Infof("%d", result) +} diff --git a/compose/workflow/4_control_only_branch/main.go b/compose/workflow/4_control_only_branch/main.go new file mode 100644 index 0000000..97703a6 --- /dev/null +++ b/compose/workflow/4_control_only_branch/main.go @@ -0,0 +1,71 @@ +/* + * Copyright 2025 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "context" + + "github.com/cloudwego/eino/compose" + + "github.com/cloudwego/eino-examples/internal/logs" +) + +func main() { + bidder1 := func(ctx context.Context, in float64) (float64, error) { + return in + 1.0, nil + } + + bidder2 := func(ctx context.Context, in float64) (float64, error) { + return in + 2.0, nil + } + + wf := compose.NewWorkflow[float64, map[string]float64]() + + wf.AddLambdaNode("b1", compose.InvokableLambda(bidder1)). + AddInput(compose.START) + + // add a branch just like adding branch in Graph. + wf.AddBranch("b1", compose.NewGraphBranch(func(ctx context.Context, in float64) (string, error) { + if in > 5.0 { + return compose.END, nil + } + return "b2", nil + }, map[string]bool{compose.END: true, "b2": true})) + + wf.AddLambdaNode("b2", compose.InvokableLambda(bidder2)). + // b2 executes strictly after b1, but does not rely on b1's output, + // which means b2 depends on b1, but no data passing between them. + AddDependency("b1"). + AddInputWithOptions(compose.START, nil, compose.WithNoDirectDependency()) + + wf.End().AddInput("b1", compose.ToField("bidder1")). + AddInput("b2", compose.ToField("bidder2")) + + runner, err := wf.Compile(context.Background()) + if err != nil { + logs.Errorf("workflow compile error: %v", err) + return + } + + result, err := runner.Invoke(context.Background(), 3.0) + if err != nil { + logs.Errorf("workflow run err: %v", err) + return + } + + logs.Infof("%v", result) +} diff --git a/compose/workflow/5_static_values/main.go b/compose/workflow/5_static_values/main.go new file mode 100644 index 0000000..575beb9 --- /dev/null +++ b/compose/workflow/5_static_values/main.go @@ -0,0 +1,81 @@ +/* + * Copyright 2025 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "context" + "math/rand" + + "github.com/cloudwego/eino/compose" + + "github.com/cloudwego/eino-examples/internal/logs" +) + +func main() { + type bidInput struct { + Price float64 + Budget float64 + } + + bidder := func(ctx context.Context, in bidInput) (float64, error) { + if in.Price >= in.Budget { + return in.Budget, nil + } + + return in.Price + rand.Float64()*in.Budget, nil + } + + wf := compose.NewWorkflow[float64, map[string]float64]() + + wf.AddLambdaNode("b1", compose.InvokableLambda(bidder)). + AddInput(compose.START, compose.ToField("Price")). + // set 'Budget' field to 3.0 for b1 + SetStaticValue([]string{"Budget"}, 3.0) + + // add a branch just like adding branch in Graph. + wf.AddBranch("b1", compose.NewGraphBranch(func(ctx context.Context, in float64) (string, error) { + if in > 5.0 { + return compose.END, nil + } + return "b2", nil + }, map[string]bool{compose.END: true, "b2": true})) + + wf.AddLambdaNode("b2", compose.InvokableLambda(bidder)). + // b2 executes strictly after b1, but does not rely on b1's output, + // which means b2 depends on b1, but no data passing between them. + AddDependency("b1"). + AddInputWithOptions(compose.START, []*compose.FieldMapping{compose.ToField("Price")}, compose.WithNoDirectDependency()). + // set 'Budget' field to 4.0 for b2 + SetStaticValue([]string{"Budget"}, 4.0) + + wf.End().AddInput("b1", compose.ToField("bidder1")). + AddInput("b2", compose.ToField("bidder2")) + + runner, err := wf.Compile(context.Background()) + if err != nil { + logs.Errorf("workflow compile error: %v", err) + return + } + + result, err := runner.Invoke(context.Background(), 3.0) + if err != nil { + logs.Errorf("workflow run err: %v", err) + return + } + + logs.Infof("%v", result) +} diff --git a/compose/workflow/6_stream_field_map/main.go b/compose/workflow/6_stream_field_map/main.go new file mode 100644 index 0000000..52a748a --- /dev/null +++ b/compose/workflow/6_stream_field_map/main.go @@ -0,0 +1,133 @@ +/* + * Copyright 2025 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "context" + "io" + "strings" + + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/schema" + + "github.com/cloudwego/eino-examples/internal/logs" +) + +// demonstrates the stream field mapping ability of eino workflow. +// It's modified from 2_field_mapping. +func main() { + type counter struct { + FullStr string // exported because we will do field mapping for this field + SubStr string // exported because we will do field mapping for this field + } + + // wordCounter is a transformable lambda function that + // count occurrences of SubStr within FullStr, for each trunk. + wordCounter := func(ctx context.Context, c *schema.StreamReader[counter]) ( + *schema.StreamReader[int], error) { + var subStr, cachedStr string + return schema.StreamReaderWithConvert(c, func(co counter) (int, error) { + if len(co.SubStr) > 0 { + // static values will not always come in the first chunk, + // so before the static value (SubStr) comes in, + // we need to cache the full string + subStr = co.SubStr + fullStr := cachedStr + co.FullStr + cachedStr = "" + return strings.Count(fullStr, subStr), nil + } + + if len(subStr) > 0 { + return strings.Count(co.FullStr, subStr), nil + } + cachedStr += co.FullStr + return 0, schema.ErrNoValue + }), nil + } + + // create a workflow just like a Graph + wf := compose.NewWorkflow[*schema.Message, map[string]int]() + + // add lambda c1 just like in Graph + wf.AddLambdaNode("c1", compose.TransformableLambda(wordCounter)). + AddInput(compose.START, // add an input from START, specifying 2 field mappings + // map START's Message's Content field to lambda c1's FullStr field + compose.MapFields("Content", "FullStr")). + // we can set static values even if the input will be stream + SetStaticValue([]string{"SubStr"}, "o") + + // add lambda c2 just like in Graph + wf.AddLambdaNode("c2", compose.TransformableLambda(wordCounter)). + AddInput(compose.START, // add an input from START, specifying 2 field mappings + // map START's Message's ReasoningContent field to lambda c1's FullStr field + compose.MapFields("ReasoningContent", "FullStr")). + SetStaticValue([]string{"SubStr"}, "o") + + wf.End(). // Obtain the compose.END for method chaining + // add an input from c1, + // mapping full output of c1 to the map key 'content_count' + AddInput("c1", compose.ToField("content_count")). + // also add an input from c2, + // mapping full output of c2 to the map key 'reasoning_content_count' + AddInput("c2", compose.ToField("reasoning_content_count")) + + // compile the workflow just like compiling a Graph + run, err := wf.Compile(context.Background()) + if err != nil { + logs.Errorf("workflow compile error: %v", err) + return + } + + // call the workflow using Transform just like calling a Graph with Transform + result, err := run.Transform(context.Background(), + schema.StreamReaderFromArray([]*schema.Message{ + { + Role: schema.Assistant, + ReasoningContent: "I need to say something meaningful", + }, + { + Role: schema.Assistant, + Content: "Hello world!", + }, + })) + if err != nil { + logs.Errorf("workflow run err: %v", err) + return + } + + var contentCount, reasoningCount int + for { + chunk, err := result.Recv() + if err != nil { + if err == io.EOF { + result.Close() + break + } + + logs.Errorf("workflow receive err: %v", err) + return + } + + logs.Infof("%v", chunk) + + contentCount += chunk["content_count"] + reasoningCount += chunk["reasoning_content_count"] + } + + logs.Infof("content count: %d", contentCount) + logs.Infof("reasoning count: %d", reasoningCount) +}