Eino Compose

Orchestration Overview

The github.com/cloudwego/eino/compose package provides three orchestration APIs:

*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]:

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) {
        // 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 -- Full Graph API, branches, state graph, cyclic graph, complete ReAct example
• reference/chain.md -- Chain API, when to use, parallel/branch in chain
• reference/workflow.md -- Workflow API, field-level mapping helpers, constraints
• reference/stream.md -- StreamReader/Writer, Pipe/Copy/Merge, lambda constructors, auto-conversion rules
• reference/callback.md -- Callback timings, handler registration, trigger rules, tracing example
• reference/call-option.md -- Per-request CallOption, component-type options, node targeting
• reference/checkpoint-and-state.md -- CheckPointStore, interrupt/resume, state management

CallOption Reference

Pass per-request configuration to specific nodes at invocation time via compose.Option.

Component-Type Options

Apply options to all nodes of a given component type:

out, _ := r.Invoke(ctx, input,
    compose.WithChatModelOption(model.WithTemperature(0.5)),
    compose.WithChatModelOption(model.WithMaxTokens(1024)),
    compose.WithEmbeddingOption(embedding.WithModel("text-embedding-3-small")),
)

Available helpers: WithChatModelOption, WithToolOption, WithRetrieverOption, WithEmbeddingOption, WithIndexerOption, WithLoaderOption, WithChatTemplateOption, WithDocTransformerOption.

Node-Targeted Options

Apply options to a specific named node:

out, _ := r.Invoke(ctx, input,
    compose.WithChatModelOption(model.WithTemperature(0.9)).DesignateNode("CreativeModel"),
    compose.WithChatModelOption(model.WithTemperature(0.1)).DesignateNode("FactualModel"),
)

Nested Graph Targeting

Target a node inside a nested sub-graph:

compose.WithChatModelOption(model.WithTemperature(0.5)).
    DesignateNodeWithPath(compose.NewNodePath("SubGraph", "InnerModel"))

Option Scoping

• No `DesignateNode`: applies to all matching component-type nodes, including those in nested graphs.
• `DesignateNode("Key")`: only the named node in the top-level graph.
• `DesignateNodeWithPath(path)`: a specific node at any nesting depth.

Callback Reference

Inject observability (logging, tracing, metrics) into orchestrated graphs.

Five Timing Points

type Handler interface {
    OnStart(ctx context.Context, info *RunInfo, input CallbackInput) context.Context
    OnEnd(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context
    OnError(ctx context.Context, info *RunInfo, err error) context.Context
    OnStartWithStreamInput(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context
    OnEndWithStreamOutput(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context
}

• OnStart / OnEnd: triggered for non-streaming input/output.
• OnStartWithStreamInput / OnEndWithStreamOutput: triggered when input/output is a stream.
• OnError: triggered on error before returning.

Which timing fires depends on the node's actual execution mode at runtime: if the node ultimately runs as a non-streaming endpoint, OnStart/OnEnd fire; if it runs as a streaming endpoint, the streaming variants fire instead.

RunInfo

type RunInfo struct {
    Name      string               // node name (set via WithNodeName)
    Type      string               // implementation type (e.g., "OpenAI")
    Component components.Component // abstract type (e.g., "ChatModel")
}

Building Handlers

Use HandlerBuilder to implement only the timings you care about:

import "github.com/cloudwego/eino/callbacks"

handler := callbacks.NewHandlerBuilder().
    OnStartFn(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context {
        log.Printf("[Start] %s/%s input=%T", info.Component, info.Name, input)
        return ctx
    }).
    OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        log.Printf("[End] %s/%s", info.Component, info.Name)
        return ctx
    }).
    OnErrorFn(func(ctx context.Context, info *callbacks.RunInfo, err error) context.Context {
        log.Printf("[Error] %s/%s err=%v", info.Component, info.Name, err)
        return ctx
    }).
    Build()

HandlerHelper (Component-Typed Callbacks)

Filter callbacks to specific component types with typed input/output:

import ucb "github.com/cloudwego/eino/utils/callbacks"

handler := ucb.NewHandlerHelper().
    ChatModel(&ucb.ModelCallbackHandler{
        OnStart: func(ctx context.Context, info *callbacks.RunInfo, input *model.CallbackInput) context.Context {
            log.Printf("Model input: %d messages", len(input.Messages))
            return ctx
        },
        OnEnd: func(ctx context.Context, info *callbacks.RunInfo, output *model.CallbackOutput) context.Context {
            log.Printf("Model output: %s", output.Message.Content)
            return ctx
        },
    }).
    Handler()

Supported: ChatModel, ChatTemplate, Retriever, Indexer, Embedding, Loader, DocumentTransformer, Tool, ToolsNode, Lambda, Graph.

Registering Handlers

Global handlers (init-time)

callbacks.AppendGlobalHandlers(handler)  // applies to all runs; not concurrency-safe

Per-invocation handlers

out, err := r.Invoke(ctx, input,
    compose.WithCallbacks(handler),  // all nodes
)

Node-targeted handlers

out, err := r.Invoke(ctx, input,
    compose.WithCallbacks(handler).DesignateNode("Model"),           // top-level node
    compose.WithCallbacks(handler).DesignateNodeWithPath(
        compose.NewNodePath("SubGraph", "InnerNode"),                // nested node
    ),
)

Callback Trigger Rules

• Component triggers (inside implementation): if IsCallbacksEnabled() returns true, the component fires its own callbacks with rich typed input/output.
• Node triggers (graph wrapper): if the component does NOT implement callbacks, the graph wraps it and fires callbacks with the component's interface input/output types.
• Graph triggers: the graph itself fires OnStart/OnEnd with the graph's overall input/output.

Stream callback rule

Always close streams in callback handlers. The framework copies the stream for callbacks -- if a callback's copy is not closed, the original stream cannot release resources.

Example: Adding Tracing to a Graph

handler := callbacks.NewHandlerBuilder().
    OnStartFn(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context {
        log.Printf("[TRACE Start] component=%s name=%s", info.Component, info.Name)
        return ctx
    }).
    OnEndFn(func(ctx context.Context, info *callbacks.RunInfo, output callbacks.CallbackOutput) context.Context {
        log.Printf("[TRACE End] component=%s name=%s", info.Component, info.Name)
        return ctx
    }).
    Build()

// Register globally (fires for all graphs)
callbacks.AppendGlobalHandlers(handler)

// Or pass per-invocation
r, _ := g.Compile(ctx)
_, _ = r.Invoke(ctx, input, compose.WithCallbacks(handler))

Chain API Reference

Chain is a simplified wrapper over Graph for building linear sequential pipelines. It uses a fluent builder pattern.

Creating a Chain

import "github.com/cloudwego/eino/compose"

chain := compose.NewChain[InputType, OutputType](opts ...NewGraphOption)

Append Methods

All Append methods return *Chain[I, O] for method chaining:

chain.AppendChatModel(node model.BaseChatModel, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendChatTemplate(node prompt.ChatTemplate, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendToolsNode(node *compose.ToolsNode, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendAgenticToolsNode(node *compose.AgenticToolsNode, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendLambda(node *compose.Lambda, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendRetriever(node retriever.Retriever, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendEmbedding(node embedding.Embedder, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendLoader(node document.Loader, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendIndexer(node indexer.Indexer, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendDocumentTransformer(node document.Transformer, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendGraph(node compose.AnyGraph, opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendPassthrough(opts ...GraphAddNodeOpt) *Chain[I, O]
chain.AppendBranch(b *ChainBranch) *Chain[I, O]
chain.AppendParallel(p *Parallel) *Chain[I, O]

Compile and Run

r, err := chain.Compile(ctx, opts ...GraphCompileOption)
out, err := r.Invoke(ctx, input)
stream, err := r.Stream(ctx, input)

When to Use Chain vs Graph

Use Chain when: nodes flow one after another, possibly with a branch or parallel step. Use Graph when: you need cycles, complex fan-in/fan-out, or explicit edge control.

Complete Example: Prompt -> Model -> Parse

package main

import (
    "context"
    "fmt"

    "github.com/cloudwego/eino/components/model"
    "github.com/cloudwego/eino/components/prompt"
    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

func main() {
    ctx := context.Background()

    tmpl := prompt.FromMessages(schema.FString,
        schema.SystemMessage("You are a helpful assistant."),
        schema.UserMessage("{question}"),
    )

    chain := compose.NewChain[map[string]any, string]()
    chain.
        AppendChatTemplate(tmpl).
        AppendChatModel(chatModel). // chatModel implements model.BaseChatModel
        AppendLambda(compose.InvokableLambda(func(ctx context.Context, msg *schema.Message) (string, error) {
            return msg.Content, nil
        }))

    r, err := chain.Compile(ctx)
    if err != nil {
        panic(err)
    }

    out, err := r.Invoke(ctx, map[string]any{"question": "What is Go?"})
    if err != nil {
        panic(err)
    }
    fmt.Println(out)
}

Parallel in Chain

AppendParallel runs multiple nodes concurrently, all receiving the same input. Output is map[string]any.

parallel := compose.NewParallel()
parallel.
    AddLambda("summary", compose.InvokableLambda(summarizeFn)).
    AddLambda("keywords", compose.InvokableLambda(extractKeywordsFn))

chain := compose.NewChain[string, map[string]any]()
chain.
    AppendParallel(parallel)

r, _ := chain.Compile(ctx)
result, _ := r.Invoke(ctx, "Some long document text...")
// result = map[string]any{"summary": "...", "keywords": "..."}

Branch in Chain

branchCond := func(ctx context.Context, input string) (string, error) {
    if len(input) > 100 {
        return "long", nil
    }
    return "short", nil
}

branch := compose.NewChainBranch(branchCond).
    AddLambda("long", compose.InvokableLambda(longHandler)).
    AddLambda("short", compose.InvokableLambda(shortHandler))

chain := compose.NewChain[string, string]()
chain.
    AppendBranch(branch).
    AppendLambda(compose.InvokableLambda(postProcess))

r, _ := chain.Compile(ctx)

All branch paths must converge to the same next node in the chain (or to END). The output type of all branches must align with the next node's input type.

Nesting

A Chain implements AnyGraph and can be nested inside Graph or another Chain:

inner := compose.NewChain[string, string]()
inner.AppendLambda(compose.InvokableLambda(fn1))

outer := compose.NewChain[string, string]()
outer.AppendGraph(inner).AppendLambda(compose.InvokableLambda(fn2))

Checkpoint and State Reference

How to manage per-request state, interrupt graph execution, and resume from checkpoints.

State Graph

Share state across nodes within a single graph run. State is created per-request and destroyed after the run completes.

Enabling State

type myState struct {
    Messages []*schema.Message
    Counter  int
}

g := compose.NewGraph[string, string](
    compose.WithGenLocalState(func(ctx context.Context) *myState {
        return &myState{}
    }),
)

Reading/Writing State with Handlers

// Before node executes
preHandler := func(ctx context.Context, in string, state *myState) (string, error) {
    state.Counter++
    return in, nil
}

// After node executes
postHandler := func(ctx context.Context, out string, state *myState) (string, error) {
    state.Messages = append(state.Messages, schema.AssistantMessage(out, nil))
    return out, nil
}

g.AddLambdaNode("node", lambda,
    compose.WithStatePreHandler(preHandler),
    compose.WithStatePostHandler(postHandler),
)

Handler type alignment:

• StatePreHandler[I, S]: I must match the node's non-stream input type.
• StatePostHandler[O, S]: O must match the node's non-stream output type.
• StreamStatePreHandler: input is *schema.StreamReader[I].
• StreamStatePostHandler: input is *schema.StreamReader[O].

ProcessState (inside a node)

lambda := compose.InvokableLambda(func(ctx context.Context, in string) (string, error) {
    err := compose.ProcessState[*myState](ctx, func(_ context.Context, s *myState) error {
        s.Counter++
        return nil
    })
    return in, err
})

State access is mutex-protected by the framework.

Checkpoint and Interrupt

Checkpoints save graph execution state so it can be resumed later. Interrupts pause execution at specific points.

CheckpointStore Interface

type CheckpointStore interface {
    Get(ctx context.Context, key string) (value []byte, existed bool, err error)
    Set(ctx context.Context, key string, value []byte) (err error)
}

You must implement this interface (e.g., backed by Redis, a database, or in-memory store).

Enabling Checkpoints

r, err := g.Compile(ctx,
    compose.WithCheckPointStore(store),
    compose.WithInterruptBeforeNodes([]string{"human_review"}),
    compose.WithInterruptAfterNodes([]string{"data_fetch"}),
)

Running with Checkpoint ID

checkpointID := "request-123"

// First run: will interrupt before "human_review"
result, err := r.Invoke(ctx, input, compose.WithCheckPointID(checkpointID))
if info, ok := compose.ExtractInterruptInfo(err); ok {
    // Graph interrupted. info contains state, interrupt nodes, etc.
    // result is zero-value (not meaningful)
}

Resuming

// Resume from checkpoint (input is ignored when resuming)
result, err = r.Invoke(ctx, "", compose.WithCheckPointID(checkpointID))

Modifying State Before Resume

result, err = r.Invoke(ctx, "",
    compose.WithCheckPointID(checkpointID),
    compose.WithStateModifier(func(ctx context.Context, path compose.NodePath, state any) error {
        s := state.(*myState)
        s.Approved = true
        return nil
    }),
)

Dynamic Interrupt

A node can trigger interrupt at runtime by returning a special error:

Basic (v0.7.0+)

lambda := compose.InvokableLambda(func(ctx context.Context, in string) (string, error) {
    if needsApproval(in) {
        return "", compose.Interrupt(ctx, &ApprovalRequest{Content: in})
    }
    return process(in), nil
})

Stateful Interrupt (preserves local state across resume)

return "", compose.StatefulInterrupt(ctx, &ApprovalInfo{
    ToolName: "search",
    Args:     localState,
}, localState) // third param is local state to persist across resume

Checking Interrupt State on Resume

wasInterrupted, _, storedState := compose.GetInterruptState[string](ctx)
if wasInterrupted {
    // Node was previously interrupted; storedState has the persisted local state
}

isResumeTarget, hasData, data := compose.GetResumeContext[*ApprovalResult](ctx)
if isResumeTarget && hasData {
    // This node was explicitly resumed with data
    if data.Approved {
        // proceed
    }
}

External Interrupt (Cancel With Checkpoint)

Interrupt a running graph from outside (e.g., graceful shutdown):

ctx, interrupt := compose.WithGraphInterrupt(context.Background())

go func() {
    // Run graph with interruptible context
    result, err = r.Invoke(ctx, input, compose.WithCheckPointID(id))
}()

// Later, trigger interrupt from outside
interrupt(compose.WithGraphInterruptTimeout(5 * time.Second))

Registering Custom Types for Serialization

Checkpoint serialization requires type registration for custom structs:

import "github.com/cloudwego/eino/schema"

func init() {
    schema.RegisterName[*MyState]("my_state_v1")
}

• The registered name is used as a serialization key. Changing it after checkpoints have been persisted will make those checkpoints unrecoverable.
• Unexported fields are not serialized.
• The default serializer uses JSON (via sonic). For better performance, use a gob-based serializer via compose.WithSerializer(). The ADK layer already uses gob internally.
• schema.RegisterName registers the type with both gob and the internal JSON serializer, so it works with either approach.

Stream Checkpoint

When checkpointing streams, register a concat function so the framework can combine chunks:

compose.RegisterStreamChunkConcatFunc(func(chunks []MyChunk) (MyChunk, error) {
    var result MyChunk
    for _, c := range chunks {
        result.Body += c.Body
    }
    return result, nil
})

Built-in concat: *schema.Message, []*schema.Message, string.

Complete Example: Graph with Checkpoint

package main

import (
    "context"
    "fmt"

    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

type reviewState struct {
    Approved bool
    Content  string
}

func main() {
    ctx := context.Background()
    store := NewMyCheckpointStore() // implements compose.CheckpointStore

    g := compose.NewGraph[string, string](
        compose.WithGenLocalState(func(ctx context.Context) *reviewState {
            return &reviewState{}
        }),
    )

    _ = g.AddLambdaNode("prepare", compose.InvokableLambda(
        func(ctx context.Context, in string) (string, error) {
            compose.ProcessState[*reviewState](ctx, func(_ context.Context, s *reviewState) error {
                s.Content = in
                return nil
            })
            return in, nil
        },
    ))

    _ = g.AddLambdaNode("review", compose.InvokableLambda(
        func(ctx context.Context, in string) (string, error) {
            return "reviewed: " + in, nil
        },
    ))

    _ = g.AddEdge(compose.START, "prepare")
    _ = g.AddEdge("prepare", "review")
    _ = g.AddEdge("review", compose.END)

    r, _ := g.Compile(ctx,
        compose.WithCheckPointStore(store),
        compose.WithInterruptBeforeNodes([]string{"review"}),
    )

    // First run: interrupts before "review"
    id := "session-1"
    _, err := r.Invoke(ctx, "hello", compose.WithCheckPointID(id))
    if info, ok := compose.ExtractInterruptInfo(err); ok {
        fmt.Printf("Interrupted before: %v\n", info.BeforeNodes)
    }

    // Resume with state modification
    result, _ := r.Invoke(ctx, "",
        compose.WithCheckPointID(id),
        compose.WithStateModifier(func(_ context.Context, _ compose.NodePath, state any) error {
            state.(*reviewState).Approved = true
            return nil
        }),
    )
    fmt.Println(result) // "reviewed: hello"
}

Nested Graph Checkpoints

Enable interrupt in sub-graphs via WithGraphCompileOptions:

g.AddGraphNode("sub", subGraph, compose.WithGraphCompileOptions(
    compose.WithInterruptAfterNodes([]string{"inner_node"}),
))

g.Compile(ctx, compose.WithCheckPointStore(store))

The parent graph must have a checkpoint store. When resuming from a sub-graph interrupt, StateModifier receives the sub-graph's state.

Graph API Reference

Full API reference for compose.Graph -- the most flexible orchestration primitive in Eino, supporting directed graphs with optional cycles.

Creating a Graph

import "github.com/cloudwego/eino/compose"

// I = graph input type, O = graph output type
g := compose.NewGraph[I, O](opts ...NewGraphOption)

Adding Nodes

Every Add*Node method takes a string key, the component instance, and optional GraphAddNodeOpt:

// Built-in component nodes
g.AddChatModelNode(key string, node model.BaseChatModel, opts ...GraphAddNodeOpt) error
g.AddChatTemplateNode(key string, node prompt.ChatTemplate, opts ...GraphAddNodeOpt) error
g.AddToolsNode(key string, node *compose.ToolsNode, opts ...GraphAddNodeOpt) error
g.AddAgenticToolsNode(key string, node *compose.AgenticToolsNode, opts ...GraphAddNodeOpt) error
g.AddRetrieverNode(key string, node retriever.Retriever, opts ...GraphAddNodeOpt) error
g.AddEmbeddingNode(key string, node embedding.Embedder, opts ...GraphAddNodeOpt) error
g.AddIndexerNode(key string, node indexer.Indexer, opts ...GraphAddNodeOpt) error
g.AddLoaderNode(key string, node document.Loader, opts ...GraphAddNodeOpt) error
g.AddDocumentTransformerNode(key string, node document.Transformer, opts ...GraphAddNodeOpt) error

// Lambda nodes (custom logic)
g.AddLambdaNode(key string, node *compose.Lambda, opts ...GraphAddNodeOpt) error

// Sub-graph node
g.AddGraphNode(key string, node compose.AnyGraph, opts ...GraphAddNodeOpt) error

// Passthrough node (forwards input unchanged)
g.AddPassthroughNode(key string, opts ...GraphAddNodeOpt) error

Node Options

compose.WithNodeName("display_name")    // Sets RunInfo.Name for callbacks
compose.WithInputKey("key")             // Extract map[string]any value by key as node input
compose.WithOutputKey("key")            // Wrap node output into map[string]any{key: output}
compose.WithStatePreHandler(handler)    // Pre-process with state access
compose.WithStatePostHandler(handler)   // Post-process with state access
compose.WithStreamStatePreHandler(h)    // Stream-aware pre-handler
compose.WithStreamStatePostHandler(h)   // Stream-aware post-handler
compose.WithGraphCompileOptions(opts)   // Pass compile options to sub-graph nodes

Edges

g.AddEdge(startNode, endNode string) error

Use constants compose.START and compose.END for graph entry/exit:

g.AddEdge(compose.START, "first_node")
g.AddEdge("last_node", compose.END)

Type alignment: the output type of startNode must be assignable to the input type of endNode. This is checked at AddEdge time.

Branches

Branches enable conditional routing. Only the selected branch executes.

Single-select branch

condition := func(ctx context.Context, in OutputType) (string, error) {
    if shouldGoLeft(in) {
        return "left", nil
    }
    return "right", nil
}
endNodes := map[string]bool{"left": true, "right": true}

branch := compose.NewGraphBranch(condition, endNodes)
g.AddBranch("source_node", branch)

Multi-select branch

condition := func(ctx context.Context, in OutputType) (map[string]bool, error) {
    return map[string]bool{"a": true, "b": shouldRunB(in)}, nil
}
branch := compose.NewGraphMultiBranch(condition, map[string]bool{"a": true, "b": true})
g.AddBranch("source_node", branch)

Stream-aware branch

Use NewStreamGraphBranch when the branch can decide from partial stream input:

branch := compose.NewStreamGraphBranch(
    func(ctx context.Context, sr *schema.StreamReader[*schema.Message]) (string, error) {
        msg, err := sr.Recv()
        if err != nil { return "", err }
        defer sr.Close()
        if len(msg.ToolCalls) > 0 {
            return "tools", nil
        }
        return compose.END, nil
    },
    map[string]bool{"tools": true, compose.END: true},
)

State Graph

Enable per-request state shared across all nodes:

type agentState struct {
    Messages []*schema.Message
    Step     int
}

g := compose.NewGraph[string, string](
    compose.WithGenLocalState(func(ctx context.Context) *agentState {
        return &agentState{}
    }),
)

StatePreHandler / StatePostHandler

preHandler := func(ctx context.Context, in []*schema.Message, state *agentState) ([]*schema.Message, error) {
    state.Messages = append(state.Messages, in...)
    return in, nil
}

postHandler := func(ctx context.Context, out *schema.Message, state *agentState) (*schema.Message, error) {
    state.Step++
    return out, nil
}

g.AddChatModelNode("model", chatModel,
    compose.WithStatePreHandler(preHandler),
    compose.WithStatePostHandler(postHandler),
)

ProcessState (inside a node)

lambda := compose.InvokableLambda(func(ctx context.Context, in string) (string, error) {
    err := compose.ProcessState[*agentState](ctx, func(_ context.Context, s *agentState) error {
        s.Messages = append(s.Messages, schema.UserMessage(in))
        return nil
    })
    return in, err
})

NodeTriggerMode (Pregel vs DAG)

// Pregel mode (default): supports cycles, AnyPredecessor triggers
r, _ := g.Compile(ctx) // default: Pregel

// DAG mode: no cycles, AllPredecessor triggers
r, _ := g.Compile(ctx, compose.WithNodeTriggerMode(compose.AllPredecessor))

• AnyPredecessor (Pregel): a node runs when any predecessor completes. Supports cycles. Nodes run in SuperSteps.
• AllPredecessor (DAG): a node runs only after all predecessors complete. No cycles allowed.

Compile Options

g.Compile(ctx,
    compose.WithGraphName("my_graph"),
    compose.WithNodeTriggerMode(compose.AllPredecessor),
    compose.WithMaxRunSteps(20),             // Pregel mode only
    compose.WithCheckPointStore(store),       // Enable checkpointing
    compose.WithInterruptBeforeNodes([]string{"node1"}),
    compose.WithInterruptAfterNodes([]string{"node2"}),
    // compose.WithEagerExecution(), // Deprecated: Eager execution is automatically enabled by default when a node's trigger mode is set to AllPredecessor.
)

Complete Example: ReAct Agent Graph

A minimal ReAct loop: model → branch (has tool calls?) → tools → model cycle.

func buildReActGraph(ctx context.Context, cm model.BaseChatModel) (compose.Runnable[map[string]any, *schema.Message], error) {
    // Create ChatTemplate inline
    tpl := prompt.FromMessages(schema.FString,
        schema.SystemMessage("You are a helpful assistant."),
        schema.UserMessage("help me to book a ticket."),
    )

    // Create ToolsNode with tools
    tn, err := compose.NewToolNode(ctx, &compose.ToolsNodeConfig{
        Tools: []tool.BaseTool{bookTicketTool},
    })
    if err != nil {
        return nil, err
    }

    g := compose.NewGraph[map[string]any, *schema.Message]()

    _ = g.AddChatTemplateNode("ChatTemplate", tpl)
    _ = g.AddChatModelNode("ChatModel", cm)
    _ = g.AddToolsNode("ToolsNode", tn)

    // Edges: START -> ChatTemplate -> ChatModel
    _ = g.AddEdge(compose.START, "ChatTemplate")
    _ = g.AddEdge("ChatTemplate", "ChatModel")

    // Loop: ToolsNode -> ChatModel (creates a cycle)
    _ = g.AddEdge("ToolsNode", "ChatModel")

    // Branch after model: if tool calls → ToolsNode; otherwise → END
    // Stream branch: must consume all chunks to determine if tool calls exist,
    // because ToolCalls may appear in any chunk, not necessarily the first one.
    _ = g.AddBranch("ChatModel", compose.NewStreamGraphBranch(
        func(ctx context.Context, sr *schema.StreamReader[*schema.Message]) (string, error) {
            defer sr.Close()
            for {
                msg, err := sr.Recv()
                if err != nil {
                    break
                }
                if len(msg.ToolCalls) > 0 {
                    return "ToolsNode", nil
                }
            }
            return compose.END, nil
        },
        map[string]bool{"ToolsNode": true, compose.END: true},
    ))

    return g.Compile(ctx)
}

For state management and checkpoint/interrupt support in this pattern, see checkpoint-and-state.md.

Fan-in (Multiple Predecessors)

When multiple edges converge on one node, their outputs are merged:

• Default: all predecessors must output map[string]any with non-overlapping keys.
• Use WithOutputKey("key") to wrap a node's output into a map.
• Register custom merge: compose.RegisterValuesMergeFunc[T](func([]T) (T, error)).

Stream Programming Reference

Eino stream primitives and how streaming works within Graph/Chain/Workflow orchestration.

StreamReader and StreamWriter

From github.com/cloudwego/eino/schema:

// Create a stream pair with buffered capacity
sr, sw := schema.Pipe[T](capacity int) (*StreamReader[T], *StreamWriter[T])

// Writer side
closed := sw.Send(chunk T, err error) bool  // returns true if reader closed
sw.Close()                                   // signals EOF to reader

// Reader side
chunk, err := sr.Recv() (T, error)   // io.EOF when stream ends
sr.Close()                           // release resources, signal writer to stop

Usage Pattern

sr, sw := schema.Pipe[string](10)

go func() {
    defer sw.Close()
    sw.Send("hello ", nil)
    sw.Send("world", nil)
}()

defer sr.Close()
for {
    chunk, err := sr.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        return err
    }
    fmt.Print(chunk)
}

Stream Operations

Copy (fan-out)

copies := sr.Copy(n int) []*StreamReader[T]
// Each copy independently reads every element
// Original reader becomes unusable after Copy

Merge (fan-in)

merged := schema.MergeStreamReaders(srs []*StreamReader[T]) *StreamReader[T]
// Interleaves elements from all sources; EOF after all exhausted

// Named merge (emits SourceEOF per-source)
merged := schema.MergeNamedStreamReaders(map[string]*StreamReader[T]{
    "a": srA,
    "b": srB,
})

Convert / Filter

strReader := schema.StreamReaderWithConvert(intReader,
    func(i int) (string, error) {
        if i == 0 {
            return "", schema.ErrNoValue // skip this element
        }
        return fmt.Sprintf("val_%d", i), nil
    },
)

From Array

sr := schema.StreamReaderFromArray([]string{"a", "b", "c"})

Four Interaction Modes

The Runnable[I, O] interface exposes four modes:

type Runnable[I, O any] interface {
    Invoke(ctx context.Context, input I, opts ...Option) (O, error)
    Stream(ctx context.Context, input I, opts ...Option) (*schema.StreamReader[O], error)
    Collect(ctx context.Context, input *schema.StreamReader[I], opts ...Option) (O, error)
    Transform(ctx context.Context, input *schema.StreamReader[I], opts ...Option) (*schema.StreamReader[O], error)
}

Lambda Constructors

// Non-streaming: I -> O
compose.InvokableLambda(func(ctx context.Context, in I) (O, error))

// Streaming output: I -> StreamReader[O]
compose.StreamableLambda(func(ctx context.Context, in I) (*schema.StreamReader[O], error))

// Streaming input: StreamReader[I] -> O
compose.CollectableLambda(func(ctx context.Context, in *schema.StreamReader[I]) (O, error))

// Bidirectional streaming: StreamReader[I] -> StreamReader[O]
compose.TransformableLambda(func(ctx context.Context, in *schema.StreamReader[I]) (*schema.StreamReader[O], error))

// Combine multiple modes
compose.AnyLambda(invokeFn, streamFn, collectFn, transformFn)

Auto-Conversion Rules

The framework automatically converts between streaming modes:

When Graph is called with Invoke

All internal nodes run in Invoke mode. If a node only implements Stream, the framework auto-concats the output stream into a single value.

When Graph is called with Stream/Collect/Transform

All internal nodes run in Transform mode. Missing modes are auto-filled:

Auto-concat built-in types

The framework can automatically concat StreamReader[T] into T for:

• *schema.Message (via schema.ConcatMessages())
• string (concatenation)
• []*schema.Message
• map[string]any (merge by key)

For custom types, register a concat function:

compose.RegisterStreamChunkConcatFunc(func(chunks []MyType) (MyType, error) {
    // combine chunks into one MyType
    return combined, nil
})

Complete Streaming Example

package main

import (
    "context"
    "fmt"
    "io"

    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

func main() {
    ctx := context.Background()

    g := compose.NewGraph[string, string]()

    // StreamableLambda: takes string, returns stream of string chunks
    _ = g.AddLambdaNode("streamer", compose.StreamableLambda(
        func(ctx context.Context, in string) (*schema.StreamReader[string], error) {
            sr, sw := schema.Pipe[string](10)
            go func() {
                defer sw.Close()
                for _, word := range []string{"Hello ", "streaming ", "world!"} {
                    sw.Send(word, nil)
                }
            }()
            return sr, nil
        },
    ))

    // TransformableLambda: transforms stream chunk by chunk
    _ = g.AddLambdaNode("upper", compose.TransformableLambda(
        func(ctx context.Context, in *schema.StreamReader[string]) (*schema.StreamReader[string], error) {
            return schema.StreamReaderWithConvert(in, func(s string) (string, error) {
                return "[" + s + "]", nil
            }), nil
        },
    ))

    _ = g.AddEdge(compose.START, "streamer")
    _ = g.AddEdge("streamer", "upper")
    _ = g.AddEdge("upper", compose.END)

    r, _ := g.Compile(ctx)

    // Stream call: get output as a stream
    stream, _ := r.Stream(ctx, "input")
    defer stream.Close()

    for {
        chunk, err := stream.Recv()
        if err == io.EOF {
            break
        }
        fmt.Print(chunk) // prints: [Hello ][streaming ][world!]
    }
}

ErrNoValue

schema.ErrNoValue is a sentinel used in StreamReaderWithConvert to skip elements:

filtered := schema.StreamReaderWithConvert(sr, func(msg *schema.Message) (*schema.Message, error) {
    if msg.Content == "" {
        return nil, schema.ErrNoValue // skip empty messages
    }
    return msg, nil
})

Do NOT use ErrNoValue in any other context.

Workflow API Reference

Workflow provides DAG orchestration with field-level mapping between nodes. Unlike Graph which requires whole input/output type alignment, Workflow maps individual struct fields between nodes.

Creating a Workflow

import "github.com/cloudwego/eino/compose"

wf := compose.NewWorkflow[InputType, OutputType](opts ...NewGraphOption)

Adding Nodes

Same node types as Graph. Returns *WorkflowNode for method chaining:

wf.AddChatModelNode(key, chatModel, opts...) *WorkflowNode
wf.AddChatTemplateNode(key, tmpl, opts...) *WorkflowNode
wf.AddToolsNode(key, toolsNode, opts...) *WorkflowNode
wf.AddLambdaNode(key, lambda, opts...) *WorkflowNode
wf.AddRetrieverNode(key, retriever, opts...) *WorkflowNode
wf.AddEmbeddingNode(key, embedder, opts...) *WorkflowNode
wf.AddIndexerNode(key, indexer, opts...) *WorkflowNode
wf.AddLoaderNode(key, loader, opts...) *WorkflowNode
wf.AddDocumentTransformerNode(key, transformer, opts...) *WorkflowNode
wf.AddGraphNode(key, graph, opts...) *WorkflowNode
wf.AddPassthroughNode(key, opts...) *WorkflowNode

Field Mapping with AddInput

AddInput declares both a control dependency and a data dependency:

node.AddInput(fromNodeKey string, mappings ...*FieldMapping) *WorkflowNode

Mapping Helpers

// Top-level field to top-level field
compose.MapFields("SourceField", "TargetField")

// Full output to a top-level field
compose.ToField("TargetField")

// Top-level field to full input
compose.FromField("SourceField")

// Nested field paths
compose.MapFieldPaths([]string{"Outer", "Inner"}, []string{"TargetField"})
compose.ToFieldPath([]string{"Target", "Nested"})
compose.FromFieldPath([]string{"Source", "Nested"})

No mapping (whole output -> whole input)

node.AddInput(compose.START) // maps all of START output to this node's input

Control-Only and Data-Only Dependencies

Data-only (no control dependency)

node.AddInputWithOptions(fromNodeKey, []*compose.FieldMapping{
    compose.MapFields("Price", "InputPrice"),
}, compose.WithNoDirectDependency())

The source node's completion does NOT trigger this node. Data is available only if a control path exists through other nodes.

Control-only (no data)

node.AddDependency("Predecessor")

The predecessor must complete before this node runs, but no data is passed.

Setting the END Node

wf.End().AddInput("LastNode")
// or with field mapping:
wf.End().
    AddInput("NodeA", compose.ToField("ResultA")).
    AddInput("NodeB", compose.ToField("ResultB"))

Static Values

Inject constant values into a node's input fields:

wf.AddLambdaNode("Bidder", compose.InvokableLambda(bidderFn)).
    AddInput(compose.START, compose.ToField("Price")).
    SetStaticValue([]string{"Budget"}, 3.0)

SetStaticValue(path FieldPath, value any) sets the value at the given field path.

Branches

Same branch API as Graph, but branches in Workflow are control-only (no data passing):

wf.AddBranch("SourceNode", compose.NewGraphBranch(
    func(ctx context.Context, in float64) (string, error) {
        if in > threshold {
            return compose.END, nil
        }
        return "NextNode", nil
    },
    map[string]bool{compose.END: true, "NextNode": true},
))

Downstream nodes of a branch get their data through AddInput/AddInputWithOptions, not from the branch source.

Compile and Run

r, err := wf.Compile(ctx, opts ...GraphCompileOption)
out, err := r.Invoke(ctx, input)
stream, err := r.Stream(ctx, input)

When to Use Workflow vs Graph

Use Workflow when: nodes have different input/output struct types and you want direct field mapping without glue lambdas.

Complete Example: Parallel Processing with Field Mapping

package main

import (
    "context"
    "strings"

    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

type counterInput struct {
    FullStr string
    SubStr  string
}

func main() {
    ctx := context.Background()

    wordCounter := func(ctx context.Context, c counterInput) (int, error) {
        return strings.Count(c.FullStr, c.SubStr), nil
    }

    type input struct {
        *schema.Message
        SubStr string
    }

    wf := compose.NewWorkflow[input, map[string]any]()

    // C1 counts SubStr in Message.Content
    wf.AddLambdaNode("C1", compose.InvokableLambda(wordCounter)).
        AddInput(compose.START,
            compose.MapFields("SubStr", "SubStr"),
            compose.MapFieldPaths([]string{"Message", "Content"}, []string{"FullStr"}))

    // C2 counts SubStr in Message.ReasoningContent
    wf.AddLambdaNode("C2", compose.InvokableLambda(wordCounter)).
        AddInput(compose.START,
            compose.MapFields("SubStr", "SubStr"),
            compose.MapFieldPaths([]string{"Message", "ReasoningContent"}, []string{"FullStr"}))

    wf.End().
        AddInput("C1", compose.ToField("ContentCount")).
        AddInput("C2", compose.ToField("ReasoningCount"))

    r, err := wf.Compile(ctx)
    if err != nil {
        panic(err)
    }

    result, _ := r.Invoke(ctx, input{
        Message: &schema.Message{
            Content:          "Hello world!",
            ReasoningContent: "I need to say something meaningful",
        },
        SubStr: "o",
    })
    // result = map[string]any{"ContentCount": 2, "ReasoningCount": 1}
}

Constraints

• Map keys must be string or types convertible to string.
• WithNodeTriggerMode and WithMaxRunSteps are not supported (fixed to AllPredecessor, no cycles).
• Cannot map multiple sources to the same target field.
• Struct fields used in mapping must be exported (reflection-based).