diff --git a/cmd/verifyexamples/examples.go b/cmd/verifyexamples/examples.go index 01f97b86..bd001987 100644 --- a/cmd/verifyexamples/examples.go +++ b/cmd/verifyexamples/examples.go @@ -639,6 +639,17 @@ var workflowExamples = []ExampleDefinition{ "ReverseTextExecutor: !DLROW ,OLLEH", }, }, + { + Name: "03_workflows_observability_executor_io", + ProjectPath: "examples/03-workflows/observability/executor_io", + IsDeterministic: true, + MustContain: []string{ + "invoked UppercaseExecutor: Hello, World!", + "completed UppercaseExecutor: HELLO, WORLD!", + "invoked ReverseTextExecutor: HELLO, WORLD!", + "completed ReverseTextExecutor: !DLROW ,OLLEH", + }, + }, { Name: "03_workflows_01_start_here_02_agents_in_workflows", ProjectPath: "examples/03-workflows/01-start-here/02_agents_in_workflows", diff --git a/examples/03-workflows/observability/executor_io/main.go b/examples/03-workflows/observability/executor_io/main.go new file mode 100644 index 00000000..7465bde5 --- /dev/null +++ b/examples/03-workflows/observability/executor_io/main.go @@ -0,0 +1,65 @@ +// Copyright (c) Microsoft. All rights reserved. + +package main + +import ( + "context" + "slices" + "strings" + + "github.com/microsoft/agent-framework-go/examples/internal/demo" + "github.com/microsoft/agent-framework-go/workflow" + "github.com/microsoft/agent-framework-go/workflow/inproc" +) + +var _ = demo.NewLogger( + "Workflow Executor I/O Observation", + "This sample observes per-executor input and output by handling ExecutorInvokedEvent and ExecutorCompletedEvent.", +) + +func main() { + uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string { + return strings.ToUpper(input) + }).Bind() + + reverse := workflow.NewExecutor("ReverseTextExecutor", func(input string) string { + runes := []rune(input) + slices.Reverse(runes) + return string(runes) + }).Bind() + + wf, err := workflow.NewBuilder(uppercase). + AddEdge(uppercase, reverse). + WithOutputFrom(reverse). + Build() + if err != nil { + demo.Panic(err) + } + + run, err := inproc.Default.RunStreaming(context.Background(), wf, "Hello, World!") + if err != nil { + demo.Panic(err) + } + defer func() { _ = run.Close(context.Background()) }() + + // Unlike 01_streaming, this loop surfaces the input each executor receives via + // ExecutorInvokedEvent, not just the output via ExecutorCompletedEvent. Pairing + // the two events gives a full per-node I/O trace of the workflow. + for evt, err := range run.WatchStream(context.Background()) { + if err != nil { + demo.Panic(err) + } + switch e := evt.(type) { + case workflow.ExecutorInvokedEvent: + demo.Assistantf("invoked %s: %v", e.ExecutorID, e.Message) + case workflow.ExecutorCompletedEvent: + demo.Assistantf("completed %s: %v", e.ExecutorID, e.Result) + case workflow.OutputEvent: + demo.Assistantf("Output: %v", e.Output) + case workflow.ExecutorFailedEvent: + demo.Panic(e.Error) + case workflow.ErrorEvent: + demo.Panic(e.Error) + } + } +}