/eino-compose

name: eino-compose
description: Eino orchestration with Graph, Chain, and Workflow. Use when a user needs to build multi-step pipelines, compose components into executable graphs, handle streaming between nodes, use branching or parallel execution, manage state with checkpoints, or understand the Runnable abstraction. Covers Graph (directed graph with cycles), Chain (linear sequential), and Workflow (DAG with field mapping).

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.

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

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

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

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`:

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.

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

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:

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) {