From 82cc9a471d2623a9aa85fe2546b9bd379aca5e50 Mon Sep 17 00:00:00 2001 From: "shentong.martin" Date: Wed, 17 Dec 2025 14:38:55 +0800 Subject: [PATCH] feat(compose): async node example Change-Id: I1e5af39c3ac197017ef809d7e68eb4911266925d --- compose/graph/async_node/README.md | 29 ++++++++ compose/graph/async_node/main.go | 108 ++++++++++++++++++++++++++++ compose/graph/async_node/service.go | 83 +++++++++++++++++++++ 3 files changed, 220 insertions(+) create mode 100644 compose/graph/async_node/README.md create mode 100644 compose/graph/async_node/main.go create mode 100644 compose/graph/async_node/service.go diff --git a/compose/graph/async_node/README.md b/compose/graph/async_node/README.md new file mode 100644 index 0000000..91c97cd --- /dev/null +++ b/compose/graph/async_node/README.md @@ -0,0 +1,29 @@ +# Async Lambda Node in Eino Graph + +This example demonstrates an "async node" implemented as a normal lambda in an Eino graph. +It covers two realistic business scenarios: + +- Report Generation (invokable): a long-running background job that produces a document URL. +- Live Transcription (streamable): a stream of tokens produced over time and converted via `StreamReaderWithConvert`. + +## Files +- `service.go`: mocked services (`generateReport`, `transcribeLive`). +- `main.go`: graph wiring, lambda nodes, and run flows. + +## How It Works +- The invokable lambda starts `generateReport` in a goroutine and blocks on a channel until completion or cancellation. +- The streamable lambda obtains a live `StreamReader[string]` and wraps it with `StreamReaderWithConvert` to transform tokens. + +## Run +```bash +cd compose/graph/async_node +go run . +``` + +You will see: +- A report URL logged from the invokable path. +- A stream of uppercase tokens from the transcription path until `EOF`. + +## Notes +- The services inject errors when inputs contain the word `error`, to showcase error propagation. +- Cancellation is respected via `context.Context`. diff --git a/compose/graph/async_node/main.go b/compose/graph/async_node/main.go new file mode 100644 index 0000000..cf1d1c5 --- /dev/null +++ b/compose/graph/async_node/main.go @@ -0,0 +1,108 @@ +/* + * 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" + "errors" + "io" + "strings" + + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/schema" + + "github.com/cloudwego/eino-examples/internal/logs" +) + +// This example demonstrates an async lambda node inside an Eino graph. +// Scenario 1: Background report generation (invokable). +// Scenario 2: Live transcription stream (streamable) with conversion. +func main() { + ctx := context.Background() + + g := compose.NewGraph[string, string]() + + // async_invokable: starts a long-running job and waits on a channel. + inv := compose.InvokableLambda(func(ctx context.Context, in string) (string, error) { + notify := make(chan reportResult, 1) + logs.Infof("async_invokable: start report job with input=%q", in) + go generateReport(ctx, in, notify) + + logs.Infof("async_invokable: waiting on reportResult notification") + select { + case r := <-notify: + return r.url, r.err + case <-ctx.Done(): + return "", ctx.Err() + } + }) + + // async_streamable: consumes a live stream and converts tokens. + str := compose.StreamableLambda(func(ctx context.Context, in string) (*schema.StreamReader[string], error) { + logs.Infof("async_streamable: start transcription stream with input=%q", in) + upstream := transcribeLive(ctx, in) + converted := schema.StreamReaderWithConvert(upstream, func(tok string) (string, error) { + if tok == "" { + return tok, errors.New("empty token") + } + return strings.ToUpper(tok), nil + }) + logs.Infof("async_streamable: waiting on upstream tokens via StreamReaderWithConvert") + return converted, nil + }) + + _ = g.AddLambdaNode("async_invokable", inv) + _ = g.AddLambdaNode("async_streamable", str) + + _ = g.AddEdge(compose.START, "async_invokable") + _ = g.AddEdge("async_invokable", "async_streamable") + _ = g.AddEdge("async_streamable", compose.END) + + run, err := g.Compile(ctx) + if err != nil { + logs.Errorf("compile error: %v", err) + return + } + + // Invoke: "generate report" path + out, err := run.Invoke(ctx, "Quarterly Sales Report") + if err != nil { + logs.Errorf("invoke error: %v", err) + } else { + logs.Infof("report url: %s", out) + } + + // Stream: "transcription" path + stream, err := run.Stream(ctx, "hello world from async node") + if err != nil { + logs.Errorf("stream error: %v", err) + return + } + + for { + chunk, err := stream.Recv() + if err != nil { + if errors.Is(err, io.EOF) { + break + } + logs.Tokenf("recv error: %v", err) + break + } + logs.Tokenf("%s", chunk) + } + stream.Close() +} diff --git a/compose/graph/async_node/service.go b/compose/graph/async_node/service.go new file mode 100644 index 0000000..c29b7be --- /dev/null +++ b/compose/graph/async_node/service.go @@ -0,0 +1,83 @@ +/* + * 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" + "errors" + "strings" + "time" + "unicode/utf8" + + "github.com/cloudwego/eino/schema" +) + +// generateReport simulates a long-running background report generation job. +// It sends the final URL or an error via the notify channel. +func generateReport(ctx context.Context, content string, notify chan<- reportResult) { + select { + case <-ctx.Done(): + notify <- reportResult{"", ctx.Err()} + close(notify) + return + case <-time.After(2 * time.Second): + if strings.Contains(strings.ToLower(content), "error") { + notify <- reportResult{"", errors.New("report generation failed")} + close(notify) + return + } + url := "https://example.com/report/" + strings.ReplaceAll(strings.ToLower(content), " ", "-") + notify <- reportResult{url, nil} + close(notify) + return + } +} + +// transcribeLive simulates a live transcription service that emits tokens over time. +// It may emit an error mid-stream for demonstration when encountering the word "error". +func transcribeLive(ctx context.Context, phrase string) *schema.StreamReader[string] { + sr, sw := schema.Pipe[string](utf8.RuneCountInString(phrase)) + + go func() { + defer sw.Close() + + splitter := func(r rune) bool { return r == ' ' || r == '-' || r == '/' } + for _, w := range strings.FieldsFunc(phrase, splitter) { + select { + case <-ctx.Done(): + sw.Send("", ctx.Err()) + return + default: + } + + if strings.EqualFold(w, "error") { + sw.Send("", errors.New("transcription stream error")) + return + } + + sw.Send(w, nil) + time.Sleep(300 * time.Millisecond) + } + }() + + return sr +} + +type reportResult struct { + url string + err error +}