Eino Agent

Eino ADK Overview

Import: github.com/cloudwego/eino/adk

The Agent Development Kit (ADK) provides a framework for building agents in Go. The ADK is generically parameterized by MessageType to support both classic *schema.Message and the new *schema.AgenticMessage. Prefer *schema.AgenticMessage for new usage.

type MessageType interface {
    *schema.Message | *schema.AgenticMessage
}

type TypedAgent[M MessageType] interface {
    Name(ctx context.Context) string
    Description(ctx context.Context) string
    Run(ctx context.Context, input *TypedAgentInput[M], options ...AgentRunOption) *AsyncIterator[*TypedAgentEvent[M]]
}

// Convenience aliases for classic message type
type Agent = TypedAgent[*schema.Message]

Agent Types

ChatModelAgent Quick Start

import (
    "context"
    "fmt"
    "log"

    "github.com/cloudwego/eino-ext/components/model/openai"
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/components/tool"
    "github.com/cloudwego/eino/components/tool/utils"
    "github.com/cloudwego/eino/compose"
)

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

    // 1. Create a tool
    searchTool, _ := utils.InferTool("search_book", "Search books by genre",
        func(ctx context.Context, input *struct {
            Genre string `json:"genre" jsonschema_description:"Book genre"`
        }) (string, error) {
            return `{"books": ["The Great Gatsby"]}`, nil
        })

    // 2. Create model (BaseModel[M], not ToolCallingChatModel)
    cm, _ := openai.NewChatModel(ctx, &openai.ChatModelConfig{
        APIKey: "your-key", Model: "gpt-4o",
    })

    // 3. Create agent
    agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
        Name:        "BookRecommender",
        Description: "Recommends books",
        Instruction: "You recommend books using the search_book tool.",
        Model:       cm,
        ToolsConfig: adk.ToolsConfig{
            ToolsNodeConfig: compose.ToolsNodeConfig{
                Tools: []tool.BaseTool{searchTool},
            },
        },
    })

    // 4. Run with Runner
    runner := adk.NewRunner(ctx, adk.RunnerConfig{Agent: agent})
    iter := runner.Query(ctx, "recommend a fiction book")
    for {
        event, ok := iter.Next()
        if !ok {
            break
        }
        if event.Err != nil {
            log.Fatal(event.Err)
        }
        if event.Output != nil && event.Output.MessageOutput != nil {
            msg, _ := event.Output.MessageOutput.GetMessage()
            fmt.Printf("Agent[%s]: %v\n", event.AgentName, msg)
        }
    }
}

Cancel Mechanism

Cancel provides safe, controllable termination of agent execution.

// Create a cancel function alongside the run
cancelOpt, cancelFn := adk.WithCancel()
iter := runner.Query(ctx, "do something", cancelOpt)

// ... iterate events ...

// Cancel at a safe point (cancelFn is non-blocking, Wait blocks until complete)
handle, ok := cancelFn(adk.WithAgentCancelMode(adk.CancelAfterChatModel))
if ok {
    handle.Wait()
}

CancelMode (bitmask):

Cancel options:

• WithAgentCancelMode(mode) -- set safe point
• WithAgentCancelTimeout(d) -- escalate to immediate if safe point not reached in time
• WithRecursive() -- propagate cancel into nested AgentTool agents

Cancel produces a CancelError on the event stream with checkpoint data for later resumption.

Model Retry

Output-based retry with full control over retry decisions.

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    // ...
    ModelRetryConfig: &adk.ModelRetryConfig{
        MaxRetries: 3,
        ShouldRetry: func(ctx context.Context, retryCtx *adk.RetryContext) *adk.RetryDecision {
            // Retry based on output content (e.g., empty response, bad finish reason)
            if retryCtx.Err != nil {
                return &adk.RetryDecision{Retry: true, Backoff: time.Second}
            }
            if retryCtx.OutputMessage == nil || retryCtx.OutputMessage.Content == "" {
                return &adk.RetryDecision{Retry: true, Backoff: time.Second}
            }
            return &adk.RetryDecision{Retry: false}
        },
    },
})

RetryContext provides: RetryAttempt, InputMessages, OutputMessage (full concatenated response), Err.

RetryDecision controls: Retry, ModifiedInputMessages, AdditionalOptions, Backoff, RejectReason.

When streaming, a WillRetryError is emitted on the stream to signal retry is occurring.

Model Failover

Dynamic model switching when primary model fails or produces unsatisfactory output.

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    // ...
    ModelFailoverConfig: &adk.ModelFailoverConfig[*schema.Message]{
        MaxRetries: 2,
        ShouldFailover: func(ctx context.Context, outputMessage *schema.Message, outputErr error) bool {
            return outputErr != nil // failover on any error
        },
        GetFailoverModel: func(ctx context.Context, failoverCtx *adk.FailoverContext[*schema.Message]) (
            model.BaseModel[*schema.Message], []*schema.Message, error) {
            // Return a different model and optionally modified input
            return backupModel, failoverCtx.InputMessages, nil
        },
    },
})

Failover interacts with Retry: when both are configured, FailoverContext.LastErr will be a *RetryExhaustedError if retry was also exhausted.

TurnLoop

Push-based event loop for multi-turn agent execution with preemption, idle timeout, and graceful shutdown.

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

loop := adk.NewTurnLoop(adk.TurnLoopConfig[string, *schema.Message]{
    GenInput: func(ctx context.Context, loop *adk.TurnLoop[string, *schema.Message], items []string) (*adk.GenInputResult[string, *schema.Message], error) {
        // Convert pushed items into agent input
        combined := strings.Join(items, "\n")
        return &adk.GenInputResult[string, *schema.Message]{
            RunCtx:   ctx,
            Input:    &adk.TypedAgentInput[*schema.Message]{Messages: []*schema.Message{schema.UserMessage(combined)}},
            Consumed: items,
        }, nil
    },
    PrepareAgent: func(ctx context.Context, loop *adk.TurnLoop[string, *schema.Message], consumed []string) (adk.Agent, error) {
        return myAgent, nil
    },
    OnAgentEvents: func(ctx context.Context, tc *adk.TurnContext[string, *schema.Message], events *adk.AsyncIterator[*adk.AgentEvent]) error {
        for {
            event, ok := events.Next()
            if !ok {
                break
            }
            if event.Err != nil {
                return event.Err
            }
            // Process events (e.g., send to client)
        }
        return nil
    },
})

// Start the loop
loop.Run(ctx)

// Push items (non-blocking, returns (ok bool, resolved <-chan struct{}))
loop.Push("user message 1")

// Preempt current turn with new input
loop.Push("urgent message", adk.WithPreempt[string, *schema.Message](adk.AfterChatModel))

// Stop gracefully
loop.Stop(adk.WithGraceful())

// Wait for exit and get final state
exitState := loop.Wait()

Key concepts:

• Push() queues items; the loop batches and processes them via GenInput
• Preemption: cancel current turn at a safe point and start new turn with pending items
• Idle timeout: UntilIdleFor(d) auto-stops after no items for duration d
• Graceful shutdown: WithGraceful(), WithGracefulTimeout(d), WithImmediate()
• Exit state: TurnLoopExitState contains ExitReason, UnhandledItems, InterruptedItems

Runner

The Runner manages agent lifecycle, context passing, and interrupt/resume:

runner := adk.NewRunner(ctx, adk.RunnerConfig{
    Agent:           myAgent,
    EnableStreaming:  true,
    CheckPointStore: myStore,  // for interrupt/resume
})

// Query (convenience for single user message)
iter := runner.Query(ctx, "hello")

// Run (full control over input messages)
iter := runner.Run(ctx, []*schema.Message{schema.UserMessage("hello")})

Middleware System

Middleware extends ChatModelAgent behavior. Configure via Handlers field:

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    // ...
    Handlers: []adk.ChatModelAgentMiddleware{fsMiddleware, summarizationMW},
})

Eight built-in middleware types (see reference/middleware.md for details):

AgentAsTool

Wrap any Agent as a Tool for use by another agent:

subAgent := createMySubAgent()
agentTool := adk.NewAgentTool(ctx, subAgent)

parentAgent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    ToolsConfig: adk.ToolsConfig{
        ToolsNodeConfig: compose.ToolsNodeConfig{
            Tools: []tool.BaseTool{agentTool},
        },
    },
})

Human-in-the-Loop

ChatModelAgent supports interrupt and resume for human approval, clarification, and feedback. See reference/human-in-the-loop.md for details.

Key pattern: tool returns compose.Interrupt(ctx, info) to pause, then runner.ResumeWithParams(ctx, checkpointID, params) to continue.

Instructions to Agent

• Default to ChatModelAgent for most use cases (single agent with tools)
• Use Runner to execute agents -- never call agent.Run() directly in production
• Middleware order matters: PatchToolCalls first, then Reduction, then Summarization
• Use DeepAgent (adk/prebuilt/deep) when you need built-in planning + filesystem + sub-agents
• Use AgentAsTool or DeepAgents' SubAgents when a sub-agent needs isolated context (no shared history)
• Use TurnLoop when building interactive applications that need preemption, idle management, or push-based input
• Model field accepts model.BaseModel[M] -- for classic path use any BaseChatModel, for agentic path use AgenticModel
• Cancel, Retry, and Failover can all be combined; failover wraps retry
• WithCancel() is a run option, not a config option -- create fresh per-run

Reference Files

Read these files on-demand for detailed API, examples, and advanced usage:

• reference/chat-model-agent.md -- ChatModelAgentConfig reference, ReAct pattern, ToolsConfig, streaming, Cancel/Retry/Failover details
• reference/deep-agents.md -- DeepAgent concept, config, architecture, comparison with ChatModelAgent
• reference/middleware.md -- All 8 middleware types with interface, config, and examples
• reference/runner-and-events.md -- Runner creation, AgentEvent/AgentOutput, event iteration patterns
• reference/agent-as-tool.md -- Wrapping an Agent as a Tool for use by another agent
• reference/human-in-the-loop.md -- Interrupt APIs, ResumableAgent, CheckPointStore, resume patterns
• reference/filesystem.md -- Filesystem Backend interface, Local and AgentKit implementations, usage with DeepAgent

AgentAsTool wraps an Agent as a Tool, enabling one agent to call another via function calling with isolated message history.

NewAgentTool

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

agentTool := adk.NewAgentTool(ctx, subAgent, options...)

The wrapped agent:

• Receives a fresh task description (not the parent's full history)
• Shares SessionValues with the parent agent
• By default does NOT emit internal AgentEvents to the parent's iterator

To emit internal events, set EmitInternalEvents: true in the parent's ToolsConfig.

Basic Usage

import (
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/components/tool"
    "github.com/cloudwego/eino/compose"
)

// Create a sub-agent
researchAgent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    Name:        "Researcher",
    Description: "Searches and summarizes information on any topic",
    Instruction: "Research the given topic thoroughly and provide a summary.",
    Model:       cm,
    ToolsConfig: adk.ToolsConfig{
        ToolsNodeConfig: compose.ToolsNodeConfig{
            Tools: []tool.BaseTool{webSearchTool},
        },
    },
})

// Wrap as tool
researchTool := adk.NewAgentTool(ctx, researchAgent)

// Use in parent agent
parentAgent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    Name:        "Coordinator",
    Description: "Coordinates research and writing tasks",
    Instruction: "Use the Researcher tool to gather information, then write a report.",
    Model:       cm,
    ToolsConfig: adk.ToolsConfig{
        ToolsNodeConfig: compose.ToolsNodeConfig{
            Tools: []tool.BaseTool{researchTool},
        },
        EmitInternalEvents: true,  // See sub-agent's events
    },
})

When to Use AgentAsTool

• The sub-agent needs only a clear task description, not the parent's full conversation history
• You want the parent to maintain control flow (call sub-agent, get result, continue reasoning)
• You need isolated context between parent and sub-agent (each has its own message history)
• SessionValues are shared between parent and sub-agent for cross-cutting concerns

ChatModelAgent is the core ADK agent that uses the ReAct pattern: LLM reasons, generates tool calls, executes tools, feeds results back, and loops until done.

Import

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

ReAct Pattern

1. Call ChatModel (Reason) 2. LLM returns tool call requests (Action) 3. ChatModelAgent executes tools (Act) 4. Tool results fed back to ChatModel (Observation) 5. Loop until ChatModel decides no more tool calls are needed

When no tools are configured, ChatModelAgent degrades to a single ChatModel call.

ChatModelAgentConfig

type TypedChatModelAgentConfig[M MessageType] struct {
    // Name of the agent. Should be unique across all agents.
    Name string

    // Description of capabilities. Helps other agents decide whether to delegate.
    Description string

    // Instruction used as system prompt. Supports f-string placeholders for session values:
    // "The current time is {Time}. The user is {User}."
    Instruction string

    // The LLM model. Must support tool calling.
    Model model.BaseModel[M] // model.BaseChatModel or model.AgenticModel

    // Tool configuration (value type, not pointer)
    ToolsConfig ToolsConfig

    // Custom function to transform instruction + input into model messages.
    // Optional. Defaults to prepending instruction as system message.
    GenModelInput TypedGenModelInput[M]

    // Max ReAct iterations. Default: 20. Agent errors if exceeded.
    MaxIterations int

    // Retry config for ChatModel failures. Optional.
    ModelRetryConfig *TypedModelRetryConfig[M]

    // Failover config for model failures. Optional.
    ModelFailoverConfig *ModelFailoverConfig[M]

    // Middleware list (replaces deprecated Middlewares field)
    Handlers []TypedChatModelAgentMiddleware[M]
}

type ChatModelAgentConfig = TypedChatModelAgentConfig[*schema.Message]

ToolsConfig

type ToolsConfig struct {
    compose.ToolsNodeConfig

    // Tools whose results cause the agent to return immediately (skip further ReAct loops).
    ReturnDirectly map[string]bool

    // When true, internal events from AgentTool sub-agents are emitted to the parent.
    EmitInternalEvents bool
}

ToolsNodeConfig comes from github.com/cloudwego/eino/compose:

type ToolsNodeConfig struct {
    Tools               []tool.BaseTool
    ToolAliases         map[string]ToolAliasConfig
    UnknownToolsHandler func(ctx context.Context, name, input string) (string, error)
    ExecuteSequentially bool
    ToolArgumentsHandler func(ctx context.Context, name, arguments string) (string, error)
    ToolCallMiddlewares  []ToolMiddleware
}

Creating Tools

Use utils.InferTool for quick tool creation:

import (
    "github.com/cloudwego/eino/components/tool/utils"
)

type SearchInput struct {
    Query string `json:"query" jsonschema_description:"Search query"`
}

type SearchOutput struct {
    Results []string `json:"results"`
}

searchTool, err := utils.InferTool("web_search", "Search the web",
    func(ctx context.Context, input *SearchInput) (*SearchOutput, error) {
        return &SearchOutput{Results: []string{"result1"}}, nil
    })

For tools that accept options (needed for interrupt/resume):

optionableTool, err := utils.InferOptionableTool("ask_user", "Ask user for input",
    func(ctx context.Context, input *AskInput, opts ...tool.Option) (string, error) {
        o := tool.GetImplSpecificOptions[myOptions](nil, opts...)
        if o.NewInput == nil {
            return "", compose.NewInterruptAndRerunErr(input.Question)
        }
        return *o.NewInput, nil
    })

Complete Example with Streaming

import (
    "context"
    "fmt"
    "io"
    "log"

    "github.com/cloudwego/eino-ext/components/model/openai"
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/components/tool"
    "github.com/cloudwego/eino/components/tool/utils"
    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

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

    // Create model
    cm, _ := openai.NewChatModel(ctx, &openai.ChatModelConfig{
        APIKey: "your-key", Model: "gpt-4o",
    })

    // Create tools
    weatherTool, _ := utils.InferTool("get_weather", "Get weather for a city",
        func(ctx context.Context, input *struct {
            City string `json:"city" jsonschema_description:"City name"`
        }) (string, error) {
            return fmt.Sprintf("25C in %s", input.City), nil
        })

    calcTool, _ := utils.InferTool("calculator", "Basic math operations",
        func(ctx context.Context, input *struct {
            Expression string `json:"expression" jsonschema_description:"Math expression"`
        }) (string, error) {
            return "42", nil
        })

    // Create agent
    agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
        Name:        "Assistant",
        Description: "A helpful assistant with weather and calculator tools",
        Instruction: "You are a helpful assistant. Use tools when needed.",
        Model:       cm,
        ToolsConfig: adk.ToolsConfig{
            ToolsNodeConfig: compose.ToolsNodeConfig{
                Tools: []tool.BaseTool{weatherTool, calcTool},
            },
        },
        MaxIterations: 10,
    })

    // Run with streaming
    runner := adk.NewRunner(ctx, adk.RunnerConfig{
        Agent:          agent,
        EnableStreaming: true,
    })

    iter := runner.Query(ctx, "What's the weather in Tokyo?")
    for {
        event, ok := iter.Next()
        if !ok {
            break
        }
        if event.Err != nil {
            log.Fatal(event.Err)
        }

        if event.Output != nil && event.Output.MessageOutput != nil {
            mv := event.Output.MessageOutput
            if mv.IsStreaming {
                // Handle streaming response
                for {
                    msg, err := mv.MessageStream.Recv()
                    if err == io.EOF {
                        break
                    }
                    if err != nil {
                        log.Fatal(err)
                    }
                    fmt.Print(msg.Content)
                }
                fmt.Println()
            } else {
                // Handle non-streaming response
                fmt.Printf("[%s] %s\n", mv.Role, mv.Message.Content)
            }
        }
    }
}

Middleware

Add middleware via the Handlers field:

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    // ...
    Handlers: []adk.ChatModelAgentMiddleware{
        patchToolCallsMW,
        summarizationMW,
        filesystemMW,
    },
})

ModelRetryConfig

See the "ModelRetryConfig (v0.9 Enhanced)" section below for the full retry API with output-based decisions, modified inputs, and backoff control.

When a streaming response will be retried, the stream emits a WillRetryError via Recv(). Handle it to show retry status to the user:

// WillRetryError is returned by stream.Recv() when an attempt is rejected and will be retried.
// The stream continues after this error — call Recv() again to get the next attempt's chunks.
chunk, err := stream.Recv()
if err != nil {
    var willRetry *adk.WillRetryError
    if errors.As(err, &willRetry) {
        fmt.Printf("Retrying (attempt %d): %s\n", willRetry.RetryAttempt, willRetry.ErrStr)
        reason := willRetry.RejectReason() // custom reason from RetryDecision
        continue // next Recv() will return chunks from the retry attempt
    }
}

ModelRetryConfig (v0.9 Enhanced)

Output-based retry with full decision control:

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    // ...
    ModelRetryConfig: &adk.ModelRetryConfig{
        MaxRetries: 3,
        ShouldRetry: func(ctx context.Context, retryCtx *adk.RetryContext) *adk.RetryDecision {
            // Access the full output message for decision
            if retryCtx.Err != nil {
                return &adk.RetryDecision{Retry: true}
            }
            if retryCtx.OutputMessage == nil || retryCtx.OutputMessage.Content == "" {
                return &adk.RetryDecision{
                    Retry:                 true,
                    ModifiedInputMessages: append(retryCtx.InputMessages, schema.UserMessage("Please provide a non-empty response")),
                    PersistModifiedInputMessages: true,
                    Backoff:               2 * time.Second,
                }
            }
            return &adk.RetryDecision{Retry: false}
        },
        // BackoffFunc provides the default delay between retries.
        // ShouldRetry can override this per-attempt via RetryDecision.Backoff (non-zero takes precedence).
        BackoffFunc: func(ctx context.Context, attempt int) time.Duration {
            return time.Duration(attempt) * time.Second // linear backoff
        },
    },
})

RetryContext fields:

• RetryAttempt int -- current retry attempt (1-based: first retry = 1)
• InputMessages []M -- messages sent to the model
• OutputMessage M -- full concatenated response (stream fully consumed for streaming)
• Err error -- error from model call (nil if output-based retry)
• Options []model.Option -- model options used

RetryDecision fields:

• Retry bool -- whether to retry
• RewriteError error -- replace the original error
• ModifiedInputMessages []M -- modified input for retry
• PersistModifiedInputMessages bool -- keep modified input in agent state
• AdditionalOptions []model.Option -- extra model options for retry
• Backoff time.Duration -- wait before retry (overrides BackoffFunc)
• RejectReason any -- attached to WillRetryError for stream consumers

ModelFailoverConfig

Dynamic model switching when the primary model fails:

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    // ...
    ModelFailoverConfig: &adk.ModelFailoverConfig[*schema.Message]{
        MaxRetries: 2,
        ShouldFailover: func(ctx context.Context, outputMessage *schema.Message, outputErr error) bool {
            return outputErr != nil
        },
        GetFailoverModel: func(ctx context.Context, failoverCtx *adk.FailoverContext[*schema.Message]) (
            model.BaseModel[*schema.Message], []*schema.Message, error) {
            return backupModel, failoverCtx.InputMessages, nil
        },
    },
})

FailoverContext fields:

• FailoverAttempt uint -- current failover attempt
• InputMessages []M -- original input messages
• LastOutputMessage M -- last model output (may be nil)
• LastErr error -- last error (may be *RetryExhaustedError if retry is also configured)

Cancel

Cancel provides safe, controllable termination during a run:

cancelOpt, cancelFn := adk.WithCancel()
iter := runner.Query(ctx, "do something", cancelOpt)

// Later: cancel at safe point
handle, ok := cancelFn(
    adk.WithAgentCancelMode(adk.CancelAfterToolCalls),
    adk.WithAgentCancelTimeout(5*time.Second),
    adk.WithRecursive(),
)
if ok {
    err := handle.Wait()
    // err is *CancelError with checkpoint for resume
}

CancelMode values (bitmask, combinable):

• CancelImmediate (0) -- abort now, stream terminated with StreamCanceledError
• CancelAfterChatModel -- wait for model call to complete
• CancelAfterToolCalls -- wait for tool execution to complete

On cancel, a CancelError is delivered via the event stream. It contains InterruptContexts for checkpoint-based resumption.

WithAfterToolCallsHook

Execute a callback after all tool calls in a ReAct iteration complete, before the next ChatModel call:

iter := runner.Query(ctx, "do work",
    adk.WithAfterToolCallsHook(func(ctx context.Context) error {
        // Fires after tool execution finishes, before the next model call.
        // Useful for TurnLoop Push+Preempt patterns where pushed items
        // must be visible to the next turn's GenInput.
        fmt.Println("All tool calls completed")
        return nil
    }),
)

SessionValues

SessionValues provide cross-agent key-value storage within a single run:

// Set before running
runner.Run(ctx, msgs, adk.WithSessionValues(map[string]any{"user": "Alice"}))

// Access in Instruction via f-string
Instruction: "The current user is {user}."

DeepAgent is a pre-built agent on top of ChatModelAgent that provides planning, filesystem access, shell execution, and sub-agent delegation out of the box.

Import

Requires eino >= v0.5.14:

import "github.com/cloudwego/eino/adk/prebuilt/deep"

When to Use

Use DeepAgent instead of plain ChatModelAgent when you need:

• Built-in task planning (WriteTodos tool)
• File system operations (read/write/edit/glob/grep)
• Shell command execution
• Sub-agent delegation with context isolation
• Auto-summarization for long conversations

Use plain ChatModelAgent when:

• You need fine-grained control over tools and prompts
• The task is simple (single tool, no planning needed)
• You want to minimize token cost (DeepAgent's planning adds overhead)

Architecture

MainAgent (ChatModelAgent + built-in tools + prompt)
    |
    +-- WriteTodos tool (task planning)
    +-- Built-in tools (read_file, write_file, edit_file, glob, grep, execute)
    +-- TaskTool -> SubAgents
            |
            +-- GeneralPurpose (same tools as main, no TaskTool)
            +-- Custom SubAgents

• MainAgent receives user input, plans via WriteTodos, delegates via TaskTool
• SubAgents have isolated context (no shared history with MainAgent)
• GeneralPurpose sub-agent is added by default for generic tasks

Configuration

type TypedConfig[M adk.MessageType] struct {
    // Name is the identifier for the Deep agent.
    Name string
    // Description provides a brief explanation of the agent's purpose.
    Description string

    // Required: the LLM model.
    // If tools are used, it must support the model.WithTools call option.
    ChatModel model.BaseModel[M]

    // Optional: tools and tool-calling configuration.
    ToolsConfig adk.ToolsConfig

    // Optional: filesystem backend for file operations.
    Backend filesystem.Backend

    // Optional: shell execution (mutually exclusive with StreamingShell).
    Shell filesystem.Shell

    // Optional: streaming shell execution.
    StreamingShell filesystem.StreamingShell

    // Optional: custom sub-agents.
    // M = *schema.Message accepts standard agents; M = *schema.AgenticMessage accepts agentic agents.
    SubAgents []adk.TypedAgent[M]

    // Optional: custom system prompt (replaces built-in prompt when non-empty).
    Instruction string

    // Optional: max reasoning iterations.
    MaxIteration int

    // Optional: disable WriteTodos tool.
    WithoutWriteTodos bool

    // Optional: disable the default general-purpose sub-agent.
    WithoutGeneralSubAgent bool

    // Optional: custom TaskTool description generator.
    TaskToolDescriptionGenerator func(ctx context.Context, availableAgents []adk.TypedAgent[M]) (string, error)

    // Optional: interface-based middleware (recommended).
    Handlers []adk.TypedChatModelAgentMiddleware[M]

    // Optional: model retry configuration.
    ModelRetryConfig *adk.TypedModelRetryConfig[M]
    // Optional: model failover configuration.
    ModelFailoverConfig *adk.ModelFailoverConfig[M]
}

type Config = TypedConfig[*schema.Message]

Quick Start

import (
    "context"
    "fmt"
    "log"

    "github.com/cloudwego/eino-ext/components/model/openai"
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/adk/filesystem"
    "github.com/cloudwego/eino/adk/prebuilt/deep"
)

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

    cm, _ := openai.NewChatModel(ctx, &openai.ChatModelConfig{
        APIKey: "your-key", Model: "gpt-4o",
    })

    backend := filesystem.NewInMemoryBackend()

    agent, err := deep.New(ctx, &deep.Config{
        ChatModel: cm,
        Backend:   backend,
    })
    if err != nil {
        log.Fatal(err)
    }

    runner := adk.NewRunner(ctx, adk.RunnerConfig{Agent: agent})
    iter := runner.Query(ctx, "Analyze the CSV file at /data/sales.csv and create a summary report")

    for {
        event, ok := iter.Next()
        if !ok {
            break
        }
        if event.Err != nil {
            log.Fatal(event.Err)
        }
        if event.Output != nil && event.Output.MessageOutput != nil {
            msg, _ := event.Output.MessageOutput.GetMessage()
            fmt.Printf("[%s] %s\n", event.AgentName, msg.Content)
        }
    }
}

Built-in Capabilities

WriteTodos

A planning tool that lets the agent create and track a structured task list. The agent calls WriteTodos to decompose complex tasks, then updates progress as it works.

Disable with WithoutWriteTodos: true.

File System Tools

When Backend is configured, the agent gets: read_file, write_file, edit_file, glob, grep.

When Shell or StreamingShell is configured, the agent also gets: execute.

Task Delegation (TaskTool)

When SubAgents are configured (or the default general-purpose sub-agent is enabled), the agent gets a task tool. The agent specifies which sub-agent to call and what task to perform.

Sub-agents run in isolated context -- they only receive the task description, not the full conversation history.

Comparison with Other Patterns

Filesystem Backend

The adk/filesystem package defines a pluggable file backend interface used by DeepAgent and filesystem-related middleware.

Interface: github.com/cloudwego/eino/adk/filesystem Implementations: github.com/cloudwego/eino-ext/adk/backend/{local,agentkit}

Backend Interface

// github.com/cloudwego/eino/adk/filesystem
type Backend interface {
    LsInfo(ctx context.Context, req *LsInfoRequest) ([]FileInfo, error)
    Read(ctx context.Context, req *ReadRequest) (*FileContent, error)
    GrepRaw(ctx context.Context, req *GrepRequest) ([]GrepMatch, error)
    GlobInfo(ctx context.Context, req *GlobInfoRequest) ([]FileInfo, error)
    Write(ctx context.Context, req *WriteRequest) error
    Edit(ctx context.Context, req *EditRequest) error
}

type Shell interface {
    Execute(ctx context.Context, input *ExecuteRequest) (*ExecuteResponse, error)
}

type StreamingShell interface {
    ExecuteStreaming(ctx context.Context, input *ExecuteRequest) (*schema.StreamReader[*ExecuteResponse], error)
}

Backend provides file operations (ls, read, grep, glob, write, edit). Shell and StreamingShell provide command execution. DeepAgent uses all three.

Implementations

Local

Local filesystem backend. Operates directly on the host's file system.

import "github.com/cloudwego/eino-ext/adk/backend/local"

backend, err := local.NewBackend(ctx, &local.Config{
    RootDir: "/path/to/workspace",
})

AgentKit (Sandbox)

Remote sandbox backend via ByteDance AgentKit. Runs file operations and code execution in an isolated cloud environment.

import "github.com/cloudwego/eino-ext/adk/backend/agentkit"

backend, err := agentkit.NewBackend(ctx, &agentkit.Config{
    AccessKeyID:    "your-access-key",
    SecretAccessKey: "your-secret-key",
    Region:         agentkit.RegionOfBeijing,
    SandboxID:      "sandbox-id",
})

Usage with DeepAgent

import (
    "github.com/cloudwego/eino/adk/prebuilt/deep"
    "github.com/cloudwego/eino-ext/adk/backend/local"
)

backend, _ := local.NewBackend(ctx, &local.Config{
    RootDir: "./workspace",
})

agent, _ := deep.New(ctx, &deep.Config{
    ChatModel: chatModel,
    Backend:   backend,
})

Key Types

type FileInfo struct {
    Path       string // file/directory path
    IsDir      bool
    Size       int64  // bytes
    ModifiedAt string // ISO 8601 format
}

type ReadRequest struct {
    FilePath string
    Offset   int // 1-based line number (default: 1)
    Limit    int // max lines to read (default: 2000)
}

type GrepRequest struct {
    Pattern         string // regex pattern (ripgrep syntax)
    Path            string // search scope directory
    Glob            string // file path filter (e.g., "*.go")
    FileType        string // file type filter (e.g., "go", "py")
    CaseInsensitive bool
    EnableMultiline bool
    AfterLines      int // context lines after match
    BeforeLines     int // context lines before match
}

type GlobInfoRequest struct {
    Pattern string // glob expression (e.g., "**/*.go")
    Path    string // base directory
}

type WriteRequest struct {
    FilePath string
    Content  string
}

type EditRequest struct {
    FilePath   string
    OldString  string // must be non-empty, matched literally
    NewString  string // must differ from OldString
    ReplaceAll bool   // false: fail if not exactly one match
}

Human-in-the-loop (HITL) enables agents to pause execution, request human input, and resume from where they stopped.

Core Concepts

1. Interrupt: Agent pauses and sends info to the user (e.g., "approve this action?") 2. Checkpoint: Framework saves execution state to a CheckPointStore 3. Resume: User provides input, framework restores state and continues

Quick Start: Approval Pattern

1. Create a tool that can interrupt

import (
    "github.com/cloudwego/eino/components/tool/utils"
    "github.com/cloudwego/eino/compose"
)

type bookInput struct {
    Location      string `json:"location"`
    PassengerName string `json:"passenger_name"`
}

// Tool that interrupts for approval before executing
bookTool, _ := utils.InferTool("BookTicket", "Book a ticket",
    func(ctx context.Context, input *bookInput) (string, error) {
        // Check if this is a resume after approval
        if isResume, hasData, data := compose.GetResumeContext[bool](ctx); isResume && hasData {
            if data {
                // User approved, execute the action
                return fmt.Sprintf("Booked ticket to %s for %s", input.Location, input.PassengerName), nil
            }
            return "Booking rejected by user", nil
        }

        // First run: interrupt for approval
        return "", compose.Interrupt(ctx, fmt.Sprintf("Approve booking to %s for %s?", input.Location, input.PassengerName))
    })

2. Create agent and runner with CheckPointStore

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    Name:        "TicketBooker",
    Instruction: "Book tickets using the BookTicket tool.",
    Model:       cm,
    ToolsConfig: adk.ToolsConfig{
        ToolsNodeConfig: compose.ToolsNodeConfig{
            Tools: []tool.BaseTool{bookTool},
        },
    },
})

runner := adk.NewRunner(ctx, adk.RunnerConfig{
    Agent:           agent,
    EnableStreaming:  true,
    CheckPointStore: store.NewInMemoryStore(), // Use Redis in production
})

3. Run and handle interrupt

iter := runner.Query(ctx, "Book a ticket to Beijing for Martin", adk.WithCheckPointID("session-1"))

var interruptID string
for {
    event, ok := iter.Next()
    if !ok {
        break
    }
    if event.Err != nil {
        log.Fatal(event.Err)
    }
    if event.Action != nil && event.Action.Interrupted != nil {
        // Save the interrupt ID for resuming later
        interruptID = event.Action.Interrupted.InterruptContexts[0].ID
        fmt.Printf("Interrupt: %v\n", event.Action.Interrupted.InterruptContexts[0].Info)
        break
    }
    // Handle normal output...
}

4. Resume with user decision

// User approves: pass true as resume data
iter, err := runner.ResumeWithParams(ctx, "session-1", &adk.ResumeParams{
    Targets: map[string]any{
        interruptID: true,  // matches compose.GetResumeContext[bool]
    },
})
if err != nil {
    log.Fatal(err)
}

// Continue processing events
for {
    event, ok := iter.Next()
    if !ok {
        break
    }
    // Handle events...
}

Interrupt APIs (ADK Layer)

Simple Interrupt

// In a custom agent's Run method:
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
go func() {
    defer gen.Close()
    gen.Send(adk.Interrupt(ctx, "Please clarify your request."))
}()
return iter

Stateful Interrupt

// In a custom agent's Run method:
state := &MyState{ProcessedItems: 42}
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
go func() {
    defer gen.Close()
    gen.Send(adk.StatefulInterrupt(ctx, "Need user feedback", state))
}()
return iter

Composite Interrupt (for multi-agent)

// In a custom agent's Run method (multi-agent):
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
go func() {
    defer gen.Close()
    gen.Send(adk.CompositeInterrupt(ctx, "Sub-agent needs attention", parentState, subInterruptSignals...))
}()
return iter

Interrupt APIs (Compose Layer -- for tools/graph nodes)

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

// Simple interrupt from a tool
return "", compose.Interrupt(ctx, "Waiting for approval")

// Stateful interrupt
return "", compose.StatefulInterrupt(ctx, "Need input", myState)

InterruptInfo Structure

When an interrupt occurs, the event contains structured info:

if event.Action != nil && event.Action.Interrupted != nil {
    for _, point := range event.Action.Interrupted.InterruptContexts {
        fmt.Printf("ID: %s\n", point.ID)           // Unique interrupt address
        fmt.Printf("Info: %v\n", point.Info)         // User-facing info
        fmt.Printf("Root cause: %v\n", point.IsRootCause)
    }
}

Resume APIs

ResumeWithParams (recommended)

iter, err := runner.ResumeWithParams(ctx, checkpointID, &adk.ResumeParams{
    Targets: map[string]any{
        interruptID1: userData1,
        interruptID2: userData2,
    },
})

Legacy Resume (with tool options)

iter, err := runner.Resume(ctx, checkpointID,
    adk.WithToolOptions([]tool.Option{WithNewInput("user response")}),
)

ResumableAgent Interface

Agents that support interrupt/resume must implement:

type TypedResumableAgent[M MessageType] interface {
    TypedAgent[M]
    Resume(ctx context.Context, info *ResumeInfo, opts ...AgentRunOption) *AsyncIterator[*TypedAgentEvent[M]]
}

type ResumableAgent = TypedResumableAgent[*schema.Message]

type ResumeInfo struct {
    WasInterrupted bool   // Always true when Resume is called
    InterruptState any    // State saved via StatefulInterrupt
    IsResumeTarget bool   // Whether this agent is the explicit resume target
    ResumeData     any    // User-provided data for this agent
}

ChatModelAgent implements ResumableAgent by default.

CheckPointStore

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

Built-in: store.NewInMemoryStore() (for development). Use Redis or similar for production.

State serialization uses encoding/gob. Register custom types:

func init() {
    gob.RegisterName("mypackage.MyType", &MyType{})
}

Common HITL Patterns

Complete Clarification Example

// Tool that asks user for clarification
type askInput struct {
    Question string `json:"question" jsonschema_description:"Question to ask the user"`
}

askTool, _ := utils.InferTool("ask_user", "Ask user for clarification",
    func(ctx context.Context, input *askInput) (string, error) {
        // Interrupt to ask user
        return "", compose.Interrupt(ctx, input.Question)
    })

// ... configure agent with askTool ...

// Handle interrupt
iter := runner.Query(ctx, "Recommend me some books", adk.WithCheckPointID("session-1"))
var interruptID string
for {
    event, ok := iter.Next()
    if !ok {
        break
    }
    if event.Action != nil && event.Action.Interrupted != nil {
        interruptID = event.Action.Interrupted.InterruptContexts[0].ID
        fmt.Printf("Agent asks: %v\n", event.Action.Interrupted.InterruptContexts[0].Info)
        break
    }
}

// Resume with user's answer
iter, _ = runner.ResumeWithParams(ctx, "session-1", &adk.ResumeParams{
    Targets: map[string]any{
        interruptID: "I want fiction books",
    },
})

ChatModelAgentMiddleware extends ChatModelAgent behavior at various execution stages.

Interface

type TypedChatModelAgentMiddleware[M MessageType] interface {
    BeforeAgent(ctx context.Context, runCtx *ChatModelAgentContext) (context.Context, *ChatModelAgentContext, error)
    AfterAgent(ctx context.Context, state *TypedChatModelAgentState[M]) (context.Context, error)
    BeforeModelRewriteState(ctx context.Context, state *TypedChatModelAgentState[M], mc *TypedModelContext[M]) (context.Context, *TypedChatModelAgentState[M], error)
    AfterModelRewriteState(ctx context.Context, state *TypedChatModelAgentState[M], mc *TypedModelContext[M]) (context.Context, *TypedChatModelAgentState[M], error)
    WrapInvokableToolCall(ctx context.Context, endpoint InvokableToolCallEndpoint, tCtx *ToolContext) (InvokableToolCallEndpoint, error)
    WrapStreamableToolCall(ctx context.Context, endpoint StreamableToolCallEndpoint, tCtx *ToolContext) (StreamableToolCallEndpoint, error)
    WrapEnhancedInvokableToolCall(ctx context.Context, endpoint EnhancedInvokableToolCallEndpoint, tCtx *ToolContext) (EnhancedInvokableToolCallEndpoint, error)
    WrapEnhancedStreamableToolCall(ctx context.Context, endpoint EnhancedStreamableToolCallEndpoint, tCtx *ToolContext) (EnhancedStreamableToolCallEndpoint, error)
    WrapModel(ctx context.Context, m model.BaseModel[M], mc *TypedModelContext[M]) (model.BaseModel[M], error)
}

// Convenience alias for classic message path
type ChatModelAgentMiddleware = TypedChatModelAgentMiddleware[*schema.Message]

Embed *adk.BaseChatModelAgentMiddleware to get default no-op implementations and only override what you need.

Execution Flow

Agent.Run(input)
  -> BeforeAgent (once per run: modify instruction, tools)
  -> [ReAct Loop]
       -> BeforeModelRewriteState (modify messages before model call)
       -> WrapModel (wrap model for logging, metrics, etc.)
       -> Model.Generate/Stream
       -> AfterModelRewriteState (modify messages after model response)
       -> If tool calls:
            -> WrapInvokableToolCall / WrapStreamableToolCall
            -> Tool.Run()
            -> Results added to messages
       -> Continue loop
  -> AfterAgent (once per run: cleanup, final state processing)
  -> Agent.Run() ends

Configuring Middleware

agent, _ := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
    Handlers: []adk.ChatModelAgentMiddleware{mw1, mw2, mw3},
})

---

FileSystem Middleware

Package: github.com/cloudwego/eino/adk/middlewares/filesystem

Provides file system access and shell execution tools to the agent.

import "github.com/cloudwego/eino/adk/middlewares/filesystem"

mw, err := filesystem.New(ctx, &filesystem.MiddlewareConfig{
    Backend: myBackend,           // Required: filesystem.Backend implementation
    Shell:   myShell,             // Optional: shell execution (mutually exclusive with StreamingShell)
    StreamingShell: myStreamShell, // Optional: streaming shell
})

Injected tools: ls, read_file, write_file, edit_file, glob, grep, execute (if Shell/StreamingShell configured).

Backend implementations:

• filesystem.NewInMemoryBackend() -- in-memory (for testing)
• github.com/cloudwego/eino-ext/adk/backend/local -- local filesystem (Unix/macOS)
• github.com/cloudwego/eino-ext/adk/backend/agentkit -- Volcengine sandbox

---

ToolSearch Middleware

Package: github.com/cloudwego/eino/adk/middlewares/dynamictool/toolsearch

Dynamic tool selection for large tool libraries. Adds a tool_search meta-tool that accepts regex to find tools by name.

import "github.com/cloudwego/eino/adk/middlewares/dynamictool/toolsearch"

mw, err := toolsearch.New(ctx, &toolsearch.Config{
    DynamicTools: []tool.BaseTool{weatherTool, stockTool, currencyTool},
})

Flow: 1. Initially, only tool_search is visible to the model 2. Model calls tool_search(regex_pattern="weather.*") 3. Matched tools become available for subsequent model calls 4. Multiple searches accumulate results

---

ToolReduction Middleware

Package: github.com/cloudwego/eino/adk/middlewares/reduction

Controls token usage from tool results via two strategies:

• Truncation: Immediately truncate oversized tool results, save full content to backend
• Clear: When total tokens exceed threshold, offload old tool results to files

import "github.com/cloudwego/eino/adk/middlewares/reduction"

mw, err := reduction.New(ctx, &reduction.Config{
    Backend:           myBackend,     // Required: storage for offloaded content
    MaxLengthForTrunc: 50000,         // Default: 50000 chars
    MaxTokensForClear: 160000,        // Default: 160000 tokens
    SkipTruncation:    false,         // Set true to skip truncation
    SkipClear:         false,         // Set true to skip clearing
})

---

Summarization Middleware

Package: github.com/cloudwego/eino/adk/middlewares/summarization

Automatically compresses conversation history when token count exceeds a threshold.

import "github.com/cloudwego/eino/adk/middlewares/summarization"

mw, err := summarization.New(ctx, &summarization.Config{
    Model: summarizationModel,  // Required: model used to generate summaries
    Trigger: &summarization.TriggerCondition{
        ContextTokens: 160000,  // Default: 160000
    },
    TranscriptFilePath: "/path/to/transcript.txt",  // Optional: save full transcript
    PreserveUserMessages: &summarization.PreserveUserMessages{
        Enabled:   true,       // Default: true (when nil)
        MaxTokens: 30000,      // Default: 30000. Keep recent user messages up to this limit
    },
})

How it works: 1. In BeforeModelRewriteState, counts tokens in current messages 2. If tokens exceed threshold, calls the summary model to compress history 3. Replaces old messages with a summary message + preserved recent user messages

---

PlanTask Middleware

Package: github.com/cloudwego/eino/adk/middlewares/plantask

Injects task management tools for the agent to create and track tasks.

import "github.com/cloudwego/eino/adk/middlewares/plantask"

mw, err := plantask.New(ctx, &plantask.Config{
    Backend: myBackend,  // Required: storage backend (should be session-scoped)
    BaseDir: "/tasks",   // Required: directory for task files
})

Injected tools:

• TaskCreate -- create a new task with subject, description
• TaskGet -- get task details by ID
• TaskUpdate -- update status, add dependencies, set owner
• TaskList -- list all tasks with status

Task status flow: pending -> in_progress -> completed (or deleted from any state).

Tasks support dependency management (blocks/blockedBy) with circular dependency detection.

---

Skill Middleware

Package: github.com/cloudwego/eino/adk/middlewares/skill

Enables progressive disclosure of skills. At startup, the agent sees skill names and descriptions. When a task matches, it loads the full SKILL.md content.

import (
    "github.com/cloudwego/eino/adk/middlewares/skill"
    "github.com/cloudwego/eino-ext/adk/backend/local"
)

// Create filesystem backend
be, _ := local.NewBackend(ctx, &local.Config{})

// Create skill backend from filesystem
skillBackend, _ := skill.NewBackendFromFilesystem(ctx, &skill.BackendFromFilesystemConfig{
    Backend: be,
    BaseDir: "/path/to/skills",  // Directory containing skill folders
})

// Create middleware
mw, _ := skill.NewMiddleware(ctx, &skill.Config{
    Backend: skillBackend,
})

Skill directory structure:

skills/
  my-skill/
    SKILL.md          # Required: frontmatter (name, description) + instructions
    scripts/          # Optional: executable code
    references/       # Optional: reference docs

Context modes in SKILL.md frontmatter:

• (empty) -- inline: skill content returned as tool result
• fork -- new agent with clean context, discarding parent message history
• fork_with_context -- new agent carrying over parent message history

---

PatchToolCalls Middleware

Package: github.com/cloudwego/eino/adk/middlewares/patchtoolcalls

Fixes "dangling tool calls" -- assistant messages with tool calls that lack corresponding tool response messages. Common in interrupted sessions or human-in-the-loop.

import "github.com/cloudwego/eino/adk/middlewares/patchtoolcalls"

mw, _ := patchtoolcalls.New(ctx, nil)  // nil config uses defaults

// Custom placeholder message
mw, _ := patchtoolcalls.New(ctx, &patchtoolcalls.Config{
    PatchedContentGenerator: func(ctx context.Context, toolName, toolCallID string) (string, error) {
        return fmt.Sprintf("Tool %s (call %s) was cancelled.", toolName, toolCallID), nil
    },
})

Place this middleware first in the chain to ensure clean message history for other middleware.

---

Agents.md Middleware

Package: github.com/cloudwego/eino/adk/middlewares/agentsmd

Injects Agents.md file contents into model input as transient context. The injected content is excluded from summarization to avoid polluting compressed history.

import "github.com/cloudwego/eino/adk/middlewares/agentsmd"

mw, err := agentsmd.New(ctx, &agentsmd.Config{
    Backend:       myBackend,                        // Required: file access backend
    AgentsMDFiles: []string{"/path/to/Agents.md"},   // Ordered list of files to load
})

Use this middleware when you want to provide persistent reference documentation to the agent without it being summarized away. The content is re-injected fresh on each model call.

---

Run-Local State

Middleware can persist key-value state that survives interrupt/resume cycles:

// Inside any middleware method
adk.SetRunLocalValue(ctx, "myKey", myValue)

// Read later (even after resume)
val, ok, err := adk.GetRunLocalValue(ctx, "myKey")

// Delete
adk.DeleteRunLocalValue(ctx, "myKey")

Values must be gob-serializable. Register custom types in init().

---

Event Emission from Middleware

Middleware can emit custom events to the agent's event stream:

adk.SendEvent(ctx, &adk.AgentEvent{
    AgentName: "MyAgent",
    Output: &adk.AgentOutput{
        CustomizedOutput: myData,
    },
})

---

Recommended Middleware Order

Handlers: []adk.ChatModelAgentMiddleware{
    patchToolCallsMW,    // 1. Fix message history first
    agentsMdMW,          // 2. Inject reference docs
    summarizationMW,     // 3. Compress if needed
    reductionMW,         // 4. Handle large tool results
    filesystemMW,        // 5. Add file tools
    skillMW,             // 6. Add skill discovery
    planTaskMW,          // 7. Add task management
}

Language Support

All built-in middleware supports English (default) and Chinese prompts:

adk.SetLanguage(adk.LanguageChinese)  // Switch to Chinese
adk.SetLanguage(adk.LanguageEnglish)  // Switch to English (default)

Runner is the core engine that executes agents, manages multi-agent coordination, context passing, and interrupt/resume.

Creating a Runner

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

runner := adk.NewRunner(ctx, adk.RunnerConfig{
    Agent:           myAgent,
    EnableStreaming:  true,                // Suggest streaming output from components that support it
    CheckPointStore: myCheckPointStore,    // Required for interrupt/resume
})

Running an Agent

// Convenience: single user message
iter := runner.Query(ctx, "What is the weather in Tokyo?")

// Full control: multiple messages
iter := runner.Run(ctx, []adk.Message{
    schema.UserMessage("Hello"),
    schema.AssistantMessage("Hi! How can I help?", nil),
    schema.UserMessage("What's the weather?"),
}, adk.WithSessionValues(map[string]any{"user": "Alice"}))

AgentInput

type TypedAgentInput[M MessageType] struct {
    Messages       []M   // Conversation messages
    EnableStreaming bool  // Suggest streaming mode for capable components
}

type AgentInput = TypedAgentInput[*schema.Message]
type Message = *schema.Message

EnableStreaming is a suggestion, not a constraint. Components that only support one mode will ignore it. AgentOutput.IsStreaming indicates the actual output mode.

AgentEvent

Every event from AsyncIterator is an AgentEvent:

type TypedAgentEvent[M MessageType] struct {
    AgentName string                // Which agent produced this event
    Output    *TypedAgentOutput[M]  // Message output (may be nil)
    Action    *AgentAction          // Control action (may be nil)
    Err       error                 // Error (may be nil; may be *CancelError or *RetryExhaustedError)
}

type AgentEvent = TypedAgentEvent[*schema.Message]

AgentOutput

type AgentOutput struct {
    MessageOutput    *MessageVariant  // Message content
    CustomizedOutput any              // Custom output data
}

type MessageVariant struct {
    IsStreaming   bool              // true = read from MessageStream, false = read from Message
    Message       Message           // Non-streaming: complete message
    MessageStream MessageStream     // Streaming: message chunk stream
    Role          schema.RoleType   // Assistant or Tool
    ToolName      string            // Set when Role is Tool
}

Reading Messages

// Non-streaming
if !mv.IsStreaming {
    fmt.Println(mv.Message.Content)
}

// Streaming
if mv.IsStreaming {
    for {
        chunk, err := mv.MessageStream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            log.Fatal(err)
        }
        fmt.Print(chunk.Content)
    }
}

// Convenience method (works for both)
msg, err := mv.GetMessage()

AgentAction

type AgentAction struct {
    Exit             bool                  // Immediately exit the multi-agent system
    Interrupted      *InterruptInfo        // Pause execution, save checkpoint
    BreakLoop        *BreakLoopAction      // Break out of a LoopAgent
    CustomizedAction any                   // Custom action
}

Event Iteration Pattern

iter := runner.Query(ctx, "hello")
for {
    event, ok := iter.Next()
    if !ok {
        break  // No more events
    }

    if event.Err != nil {
        log.Fatal(event.Err)
        break
    }

    // Handle actions
    if event.Action != nil {
        if event.Action.Interrupted != nil {
            fmt.Printf("Interrupted: %v\n", event.Action.Interrupted)
            continue
        }
    }

    // Handle output
    if event.Output != nil && event.Output.MessageOutput != nil {
        msg, err := event.Output.MessageOutput.GetMessage()
        if err != nil {
            log.Fatal(err)
        }
        fmt.Printf("[%s][%s] %s\n", event.AgentName, event.Output.MessageOutput.Role, msg.Content)
    }
}

Cancel-Aware Event Handling

When using WithCancel(), the event stream may deliver a CancelError:

cancelOpt, cancelFn := adk.WithCancel()
iter := runner.Query(ctx, "do work", cancelOpt)

// In another goroutine: trigger cancel
go func() {
    time.Sleep(5 * time.Second)
    handle, _ := cancelFn(adk.WithAgentCancelMode(adk.CancelAfterChatModel))
    handle.Wait()
}()

for {
    event, ok := iter.Next()
    if !ok {
        break
    }
    if event.Err != nil {
        if cancelErr, ok := event.Err.(*adk.CancelError); ok {
            fmt.Printf("Cancelled (mode=%v)\n", cancelErr.Info.Mode)
            // cancelErr.InterruptContexts can be used for resume
            break
        }
        log.Fatal(event.Err)
    }
    // Handle normal events...
}

AgentRunOption

Options that modify agent behavior per-run:

// Inject session values accessible to all agents
adk.WithSessionValues(map[string]any{"key": "value"})

// Set checkpoint ID for interrupt/resume
adk.WithCheckPointID("session-123")

// Skip transfer messages in history
adk.WithSkipTransferMessages()

// Custom options (agent-specific)
adk.WrapImplSpecificOptFn(func(t *myOptions) { t.Field = "value" })

// Target option to specific agents
opt := adk.WithSessionValues(vals).DesignateAgent("agent_1", "agent_2")

SessionValues

Cross-agent key-value storage within a single run:

// Inside an agent or tool:
adk.AddSessionValue(ctx, "key", "value")
val, ok := adk.GetSessionValue(ctx, "key")
allVals := adk.GetSessionValues(ctx)

// Before running (must use option, not AddSessionValue):
runner.Run(ctx, msgs, adk.WithSessionValues(map[string]any{"key": "value"}))

AsyncIterator

type AsyncIterator[T any] struct { ... }

func (ai *AsyncIterator[T]) Next() (T, bool)

• Next() blocks until an event is available or the iterator is closed
• Returns (event, true) when an event is available
• Returns (zero, false) when the agent is done

Custom Agent Implementation

type MyAgent struct{}

func (a *MyAgent) Name(ctx context.Context) string        { return "MyAgent" }
func (a *MyAgent) Description(ctx context.Context) string { return "My custom agent" }

func (a *MyAgent) Run(ctx context.Context, input *adk.AgentInput, opts ...adk.AgentRunOption) *adk.AsyncIterator[*adk.AgentEvent] {
    iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
    go func() {
        defer gen.Close()
        // Process input, generate events
        gen.Send(adk.EventFromMessage(
            schema.AssistantMessage("Hello!", nil), nil, schema.Assistant, "",
        ))
    }()
    return iter
}

Typed Runner (AgenticMessage)

For the agentic path using *schema.AgenticMessage:

runner := adk.NewTypedRunner(adk.TypedRunnerConfig[*schema.AgenticMessage]{
    Agent:          myAgenticAgent,
    EnableStreaming: true,
})

iter := runner.Run(ctx, []*schema.AgenticMessage{
    schema.UserAgenticMessage("hello"),
})
for {
    event, ok := iter.Next()
    if !ok {
        break
    }
    // event is *TypedAgentEvent[*schema.AgenticMessage]
    if event.Output != nil && event.Output.MessageOutput != nil {
        msg, _ := event.Output.MessageOutput.GetMessage()
        for _, block := range msg.ContentBlocks {
            if block.Type == schema.ContentBlockTypeAssistantGenText {
                fmt.Print(block.AssistantGenText.Text)
            }
        }
    }
}

Language Setting

// Set language for all ADK built-in prompts (global)
adk.SetLanguage(adk.LanguageChinese)  // Chinese
adk.SetLanguage(adk.LanguageEnglish)  // English (default)