eino-compose

## Orchestration Overview The `github.com/cloudwego/eino/compose` package provides three orchestration APIs: | API | Topology | Cycles | Type Alignment | |----------|-------------------|-------|----------------------| | Graph | Directed graph | Yes (Pregel mode) / No (DAG mode) | Whole input/output | | Chain | Linear sequence | No | Whole input/output | | Workflow | DAG | No | Field-level mapping | *Chain is implemented on top of Graph in Pregel mode but enforces linear topology. All three compile into `Runnable[I, O]` which exposes Invoke, Stream, Collect, and Transform. ```go import "github.com/cloudwego/eino/compose" ``` ## Choosing an API - **Chain** -- sequential pipeline: prompt -> model -> parser. Simplest API. - **Graph** -- need branching, loops (ReAct agent), or fan-out/fan-in. - **Workflow** -- need field-level mapping between nodes with different struct types; DAG only. ## Graph Quick Reference ```go g := compose.NewGraph[InputType, OutputType]() // Add nodes g.AddChatModelNode("model", chatModel) g.AddChatTemplateNode("tmpl", tmpl) g.AddToolsNode("tools", toolsNode) g.AddLambdaNode("fn", compose.InvokableLambda(myFunc)) g.AddPassthroughNode("pass") g.AddGraphNode("sub", subGraph) // Connect nodes g.AddEdge(compose.START, "tmpl") g.AddEdge("tmpl", "model") g.AddEdge("model", compose.END) // Branch (conditional routing) branch := compose.NewGraphBranch(conditionFn, map[string]bool{"a": true, "b": true}) g.AddBranch("model", branch) // Compile and run r, err := g.Compile(ctx) out, err := r.Invoke(ctx, input) ``` ## Chain Quick Reference ```go chain := compose.NewChain[InputType, OutputType]() chain. AppendChatTemplate(tmpl). AppendChatModel(model). AppendLambda(compose.InvokableLambda(parseFn)) r, err := chain.Compile(ctx) out, err := r.Invoke(ctx, input) ``` Append methods: `AppendChatModel`, `AppendChatTemplate`, `AppendToolsNode`, `AppendAgenticToolsNode`, `AppendLambda`, `AppendGraph`, `AppendParallel`, `AppendBranch`, `AppendPassthrough`, `AppendRetriever`, `AppendEmbedding`, `AppendLoader`, `AppendIndexer`, `AppendDocumentTransformer`. ## Workflow Quick Reference ```go wf := compose.NewWorkflow[InputStruct, OutputStruct]() wf.AddLambdaNode("node1", compose.InvokableLambda(fn1)). AddInput(compose.START, compose.MapFields("FieldA", "InputField")) wf.AddLambdaNode("node2", compose.InvokableLambda(fn2)). AddInput("node1", compose.ToField("Result")) wf.End().AddInput("node2") r, err := wf.Compile(ctx) ``` Field mapping helpers: `MapFields`, `ToField`, `FromField`, `MapFieldPaths`, `ToFieldPath`, `FromFieldPath`. ## Stream Programming Four interaction modes on `Runnable[I, O]`: | Mode | Input | Output | Lambda Constructor | |-----------|----------------|---------------------|------------------------------| | Invoke | `I` | `O` | `compose.InvokableLambda` | | Stream | `I` | `*StreamReader[O]` | `compose.StreamableLambda` | | Collect | `*StreamReader[I]` | `O` | `compose.CollectableLambda` | | Transform | `*StreamReader[I]` | `*StreamReader[O]` | `compose.TransformableLambda` | Framework auto-converts between modes: - **Invoke** call: all internal nodes run in Invoke mode. - **Stream/Collect/Transform** call: all internal nodes run in Transform mode; missing modes are auto-filled. Stream primitives live in `github.com/cloudwego/eino/schema`: ```go sr, sw := schema.Pipe[T](capacity) // sw.Send(chunk, nil); sw.Close() // chunk, err := sr.Recv(); sr.Close() ``` ## AgenticToolsNode `AgenticToolsNode` is the agentic counterpart to `ToolsNode`. It operates on `*schema.AgenticMessage` instead of `*schema.Message`, extracting `FunctionToolCall` content blocks and returning `FunctionToolResult` content blocks. ```go import "github.com/cloudwego/eino/compose" // Create from the same ToolsNodeConfig agenticToolsNode, err := compose.NewAgenticToolsNode(ctx, &compose.ToolsNodeConfig{ Tools: []tool.BaseTool{myTool1, myTool2}, }) // Use in Graph g := compose.NewGraph[*schema.AgenticMessage, []*schema.AgenticMessage]() g.AddAgenticToolsNode("tools", agenticToolsNode) // Use in Chain chain := compose.NewChain[*schema.AgenticMessage, []*schema.AgenticMessage]() chain.AppendAgenticToolsNode(agenticToolsNode) ``` Key differences from `ToolsNode`: - Input: `*schema.AgenticMessage` (reads `ContentBlockTypeFunctionToolCall` blocks) - Output: `[]*schema.AgenticMessage` (writes `ContentBlockTypeFunctionToolResult` blocks) - Supports multimodal tool results (text, image, audio, video, file) - Handles `ToolSearchFunctionToolResult` for dynamic tool discovery - Reuses the same `ToolsNodeConfig` -- all existing tools work unchanged ## Compile & Run ```go r, err := g.Compile(ctx, compose.WithGraphName("my_graph"), compose.WithNodeTriggerMode(compose.AllPredecessor), // DAG mode ) // Non-streaming out, err := r.Invoke(ctx, input) // Streaming stream, err := r.Stream(ctx, input) defer stream.Close() for { chunk, err := stream.Recv() if err == io.EOF { break } if err != nil { return err } process(chunk) } ``` ## State Graph Share state across nodes within a single request: ```go g := compose.NewGraph[string, string](compose.WithGenLocalState(func(ctx context.Context) *MyState { return &MyState{} })) g.AddLambdaNode("node", lambda, compose.WithStatePreHandler(func(ctx context.Context, in string, state *MyState) (string, error) { // read/write state before node executes return in, nil }), compose.WithStatePostHandler(func(ctx context.Context, out string, state *MyState) (string, error) { // read/write state after node executes return out, nil }), ) ``` ## Instructions to Agent When helping users build orchestration: 1. Default to **Graph** for most use cases. Use Chain only for simple linear pipelines. Use Workflow when field-level mapping between different struct types is needed. 2. Always show the **Compile** step -- `g.Compile(ctx)` returns `Runnable[I,O]`. 3. Always **close StreamReaders** -- use `defer sr.Close()` immediately after obtaining a stream. 4. Upstream output type must match downstream input type (or use `WithInputKey`/`WithOutputKey` for map conversion). 5. For cyclic graphs (e.g., ReAct agent), use default Pregel mode (`AnyPredecessor`). For DAGs, set `AllPredecessor`. 6. Use `compose.WithCallbacks(handler)` to inject logging/tracing at runtime. 7. Use `compose.WithCheckPointStore(store)` with interrupt nodes for pause/resume workflows. ## Reference Files Read these files on-demand for detailed API, examples, and advanced usage: - [reference/graph.md](reference/graph.md) -- Full Graph API, branches, state graph, cyclic graph, complete ReAct example - [reference/chain.md](reference/chain.md) -- Chain API, when to use, parallel/branch in chain - [reference/workflow.md](reference/workflow.md) -- Workflow API, field-level mapping helpers, constraints - [reference/stream.md](reference/stream.md) -- StreamReader/Writer, Pipe/Copy/Merge, lambda constructors, auto-conversion rules - [reference/callback.md](reference/callback.md) -- Callback timings, handler registration, trigger rules, tracing example - [reference/call-option.md](reference/call-option.md) -- Per-request CallOption, component-type options, node targeting - [reference/checkpoint-and-state.md](reference/checkpoint-and-state.md) -- CheckPointStore, interrupt/resume, state management