feat: workflow examples

Change-Id: I6ad7882527ba818ff00bea8b9ac76f0fea08951d

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

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

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

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

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

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