Custom Workflow Deep Dive: Custom Orchestration Engine, Dynamic Scheduling, and Complex Scenario Adaptation

Custom Workflow is the ultimate flexible solution for an Agent Team. When preset patterns such as Sequential, Parallel, and Loop cannot satisfy business requirements, Custom Workflow lets developers define arbitrarily complex scheduling logic in code. This post dives into the architecture design, state machine model, dynamic orchestration capabilities, and production practice of Custom Workflows.

Why Custom Workflow Is Needed

Limitations of Preset Patterns

Pattern	Capability Boundary	Scenarios It Cannot Handle
Sequential	Fixed order execution	Conditional branches, dynamic skips
Parallel	Execute independent tasks simultaneously	Task dependencies, partial results driving later steps
Loop	Iterative optimization	Multi-stage strategies with different iteration policies
Custom	Unlimited	Any complex flow

Real-world case: the after-sales workflow of an e-commerce system—“receive return request → verify order status → if shipped, query logistics → if logistics shows delivered, inspect product status → decide refund or replacement based on product status → notify user.” This flow contains conditional branches, looped queries, and dynamic Agent selection, which cannot be expressed by preset patterns.

Custom vs Orchestrator

┌─────────────────────────────────────────────────────────────┐
│                    Orchestrator Pattern                      │
│  ┌─────────────┐    LLM decision    ┌─────────────┐         │
│  │ Orchestrator │ ────────────────► │  Next Agent  │         │
│  │   (LLM)      │ ◄──────────────── │              │         │
│  └─────────────┘   execution result └─────────────┘         │
│                                                              │
│  Characteristics: flexible but non-deterministic; every      │
│  decision requires an LLM call, which is costly.             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                   Custom Workflow Pattern                    │
│  ┌─────────────┐    code logic     ┌─────────────┐          │
│  │   Handler    │ ────────────────► │  Next Agent  │          │
│  │   (Code)     │ ◄──────────────── │              │          │
│  └─────────────┘   execution result └─────────────┘          │
│                                                              │
│  Characteristics: deterministic and efficient, suitable for  │
│  complex flows with clear rules.                             │
└─────────────────────────────────────────────────────────────┘

Selection advice:

Flow logic is enumerable → Custom Workflow
Flow logic is open-ended → Orchestrator
Mixed scenarios → Custom Workflow as the backbone, with LLM calls at key decision points

State Machine-Driven Architecture

Core Abstractions

A Custom Workflow is essentially an extended implementation of a Finite State Machine (FSM):

// State machine core definition
type StateMachine struct {
    states       map[string]*State       // state set
    transitions  map[string][]Transition // transition rules
    currentState string                  // current state
    context      *ExecutionContext       // execution context
    history      []StateTransition       // execution history
}

type State struct {
    Name        string
    Type        StateType               // state type
    Agent       *agent.Agent            // associated Agent (optional)
    Action      StateAction             // state action
    OnEnter     Hook                    // enter hook
    OnExit      Hook                    // exit hook
    Timeout     time.Duration           // state timeout
    RetryPolicy *RetryPolicy            // retry policy
}

type StateType int
const (
    StateTypeAction   StateType = iota // execute action
    StateTypeDecision                  // decision branch
    StateTypeParallel                  // parallel execution
    StateTypeLoop                      // loop execution
    StateTypeWait                      // wait for external event
    StateTypeEnd                       // terminal state
)

type Transition struct {
    From      string
    To        string
    Condition ConditionFunc           // transition condition
    Priority  int                     // priority
}

type ExecutionContext struct {
    Variables   map[string]interface{}  // variable store
    StateData   map[string]interface{}  // state data
    Input       string                  // original input
    Output      string                  // current output
    Metadata    map[string]interface{}  // metadata
    CancelFunc  context.CancelFunc      // cancel function
}

State Execution Engine

func (sm *StateMachine) Execute(ctx context.Context, input string) (*WorkflowResult, error) {
    // Initialize context
    sm.context = &ExecutionContext{
        Variables: make(map[string]interface{}),
        StateData: make(map[string]interface{}),
        Input:     input,
        Metadata:  make(map[string]interface{}),
    }
    sm.context.SetVariable("input", input)
    sm.context.SetVariable("start_time", time.Now())

    // Find start state
    current := sm.getStartState()
    if current == nil {
        return nil, fmt.Errorf("no start state defined")
    }

    // State machine main loop
    for current.Type != StateTypeEnd {
        select {
        case <-ctx.Done():
            return nil, fmt.Errorf("workflow cancelled: %w", ctx.Err())
        default:
        }

        sm.currentState = current.Name

        // Record history
        sm.history = append(sm.history, StateTransition{
            State:     current.Name,
            Timestamp: time.Now(),
        })

        // Execute enter hook
        if current.OnEnter != nil {
            if err := current.OnEnter(sm.context); err != nil {
                return nil, fmt.Errorf("enter hook failed for state %s: %w", current.Name, err)
            }
        }

        // Execute state action
        result, err := sm.executeState(ctx, current)
        if err != nil {
            // Check for error transition
            if next := sm.findErrorTransition(current, err); next != nil {
                current = next
                continue
            }
            return nil, fmt.Errorf("state %s execution failed: %w", current.Name, err)
        }

        // Execute exit hook
        if current.OnExit != nil {
            if err := current.OnExit(sm.context); err != nil {
                return nil, fmt.Errorf("exit hook failed for state %s: %w", current.Name, err)
            }
        }

        // Determine next state
        next, err := sm.determineNextState(current, result)
        if err != nil {
            return nil, fmt.Errorf("transition determination failed: %w", err)
        }

        current = next
    }

    // Execute terminal state
    return sm.executeEndState(ctx, current)
}

func (sm *StateMachine) executeState(ctx context.Context, state *State) (*StateResult, error) {
    stateCtx, cancel := context.WithTimeout(ctx, state.Timeout)
    defer cancel()

    switch state.Type {
    case StateTypeAction:
        return sm.executeActionState(stateCtx, state)
    case StateTypeDecision:
        return sm.executeDecisionState(stateCtx, state)
    case StateTypeParallel:
        return sm.executeParallelState(stateCtx, state)
    case StateTypeLoop:
        return sm.executeLoopState(stateCtx, state)
    case StateTypeWait:
        return sm.executeWaitState(stateCtx, state)
    default:
        return nil, fmt.Errorf("unknown state type: %v", state.Type)
    }
}

func (sm *StateMachine) determineNextState(current *State, result *StateResult) (*State, error) {
    // Get all transitions from the current state
    transitions := sm.transitions[current.Name]

    // Sort by priority
    sort.Slice(transitions, func(i, j int) bool {
        return transitions[i].Priority > transitions[j].Priority
    })

    // Evaluate each transition condition
    for _, t := range transitions {
        if t.Condition == nil || t.Condition(sm.context, result) {
            next := sm.states[t.To]
            if next == nil {
                return nil, fmt.Errorf("target state %s not found", t.To)
            }
            return next, nil
        }
    }

    return nil, fmt.Errorf("no valid transition from state %s", current.Name)
}

Hands-On Scenario: Intelligent Customer Service System

Full Implementation

package main

import (
    "context"
    "fmt"
    "log"
    "strings"
    "time"

    "github.com/google/adk-go/agent"
    "github.com/google/adk-go/team"
    "github.com/google/adk-go/tool"
)

// CustomerServiceWorkflow intelligent customer service workflow
type CustomerServiceWorkflow struct {
    stateMachine *team.StateMachine
}

func NewCustomerServiceWorkflow(model agent.Model) (*CustomerServiceWorkflow, error) {
    sm := team.NewStateMachine()

    // 1. Intent recognition state
    intentState := &team.State{
        Name: "intent_recognition",
        Type: team.StateTypeAction,
        Agent: mustCreateAgent(agent.Config{
            Name:        "intent-classifier",
            Model:       model,
            Instruction: `You are an intent recognition expert. Analyze user input and determine the intent type.
Output must be one of: ORDER_QUERY, LOGISTICS_QUERY, REFUND_REQUEST, COMPLAINT, GENERAL`,
            Timeout: 10 * time.Second,
        }),
        OnEnter: func(ctx *team.ExecutionContext) error {
            log.Printf("[Workflow] Starting intent recognition: %s", ctx.GetVariable("input"))
            return nil
        },
    }
    sm.AddState(intentState)

    // 2. Order query branch
    orderQueryState := &team.State{
        Name: "order_query",
        Type: team.StateTypeAction,
        Agent: mustCreateAgent(agent.Config{
            Name:        "order-agent",
            Model:       model,
            Instruction: `You are an order query expert. Help users check order status and modify order information.`,
            Tools: []tool.Tool{
                tool.NewOrderQueryTool(),
                tool.NewOrderModifyTool(),
            },
            Timeout: 20 * time.Second,
        }),
    }
    sm.AddState(orderQueryState)

    // 3. Logistics query branch
    logisticsState := &team.State{
        Name: "logistics_query",
        Type: team.StateTypeAction,
        Agent: mustCreateAgent(agent.Config{
            Name:        "logistics-agent",
            Model:       model,
            Instruction: `You are a logistics query expert. Query package location and estimated delivery time.`,
            Tools: []tool.Tool{
                tool.NewPackageTrackingTool(),
                tool.NewDeliveryEstimateTool(),
            },
            Timeout: 15 * time.Second,
        }),
    }
    sm.AddState(logisticsState)

    // 4. Refund processing branch (complex flow)
    refundState := &team.State{
        Name: "refund_process",
        Type: team.StateTypeDecision,
        Action: func(ctx *team.ExecutionContext) (*team.StateResult, error) {
            // Decide refund flow based on order status
            orderStatus := ctx.GetVariable("order_status")

            if orderStatus == "unshipped" {
                return &team.StateResult{
                    Data: map[string]interface{}{
                        "refund_type": "instant",
                        "next_step":   "process_instant_refund",
                    },
                }, nil
            } else if orderStatus == "delivered" {
                return &team.StateResult{
                    Data: map[string]interface{}{
                        "refund_type": "return_required",
                        "next_step":   "process_return_refund",
                    },
                }, nil
            }

            return &team.StateResult{
                Data: map[string]interface{}{
                    "refund_type": "review_required",
                    "next_step":   "escalate_to_human",
                },
            }, nil
        },
    }
    sm.AddState(refundState)

    // 5. Instant refund
    instantRefundState := &team.State{
        Name: "process_instant_refund",
        Type: team.StateTypeAction,
        Agent: mustCreateAgent(agent.Config{
            Name:        "refund-agent",
            Model:       model,
            Instruction: `Process instant refunds. Confirm refund amount and return to original payment method.`,
            Tools: []tool.Tool{
                tool.NewRefundTool(),
            },
            Timeout: 15 * time.Second,
        }),
    }
    sm.AddState(instantRefundState)

    // 6. Return refund
    returnRefundState := &team.State{
        Name: "process_return_refund",
        Type: team.StateTypeLoop,
        Agent: mustCreateAgent(agent.Config{
            Name:        "return-agent",
            Model:       model,
            Instruction: `Process return-and-refund flow. Guide user return, inspect item, and process refund.`,
            Tools: []tool.Tool{
                tool.NewReturnLabelTool(),
                tool.NewItemInspectionTool(),
            },
            Timeout: 30 * time.Second,
        }),
        // Loop exit condition: return complete or timeout
    }
    sm.AddState(returnRefundState)

    // 7. Human escalation
    humanEscalationState := &team.State{
        Name: "escalate_to_human",
        Type: team.StateTypeAction,
        Action: func(ctx *team.ExecutionContext) (*team.StateResult, error) {
            // Create a ticket
            ticket := createTicket(ctx)

            return &team.StateResult{
                Output: fmt.Sprintf("Your issue has been submitted to a human agent. Ticket ID: %s, expected reply within 2 hours.", ticket.ID),
                Data: map[string]interface{}{
                    "ticket_id": ticket.ID,
                    "escalated": true,
                },
            }, nil
        },
    }
    sm.AddState(humanEscalationState)

    // 8. Complaint handling
    complaintState := &team.State{
        Name: "complaint_handling",
        Type: team.StateTypeParallel,
        // Execute in parallel: record complaint + analyze sentiment + generate compensation plan
    }
    sm.AddState(complaintState)

    // 9. General Q&A
    generalState := &team.State{
        Name: "general_qa",
        Type: team.StateTypeAction,
        Agent: mustCreateAgent(agent.Config{
            Name:        "general-agent",
            Model:       model,
            Instruction: `You are a general customer service assistant. Answer general questions about products and policies.`,
            Timeout: 15 * time.Second,
        }),
    }
    sm.AddState(generalState)

    // 10. End state
    endState := &team.State{
        Name:        "end",
        Type:        team.StateTypeEnd,
        OnEnter: func(ctx *team.ExecutionContext) error {
            log.Printf("[Workflow] Workflow ended, total duration: %v", time.Since(ctx.GetVariable("start_time").(time.Time)))
            return nil
        },
    }
    sm.AddState(endState)

    // Define transition rules
    sm.AddTransition("intent_recognition", "order_query",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return strings.Contains(result.Output, "ORDER_QUERY")
        }, 10)

    sm.AddTransition("intent_recognition", "logistics_query",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return strings.Contains(result.Output, "LOGISTICS_QUERY")
        }, 10)

    sm.AddTransition("intent_recognition", "refund_process",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return strings.Contains(result.Output, "REFUND_REQUEST")
        }, 10)

    sm.AddTransition("intent_recognition", "complaint_handling",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return strings.Contains(result.Output, "COMPLAINT")
        }, 10)

    sm.AddTransition("intent_recognition", "general_qa",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return strings.Contains(result.Output, "GENERAL") || result.Output == ""
        }, 5) // low priority, fallback

    // Refund flow branches
    sm.AddTransition("refund_process", "process_instant_refund",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return result.Data["refund_type"] == "instant"
        }, 10)

    sm.AddTransition("refund_process", "process_return_refund",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return result.Data["refund_type"] == "return_required"
        }, 10)

    sm.AddTransition("refund_process", "escalate_to_human",
        func(ctx *team.ExecutionContext, result *team.StateResult) bool {
            return result.Data["refund_type"] == "review_required"
        }, 10)

    // All processing states transition to end
    for _, stateName := range []string{"order_query", "logistics_query", "process_instant_refund",
        "process_return_refund", "escalate_to_human", "complaint_handling", "general_qa"} {
        sm.AddTransition(stateName, "end", nil, 1)
    }

    return &CustomerServiceWorkflow{stateMachine: sm}, nil
}

func (w *CustomerServiceWorkflow) Handle(ctx context.Context, userInput string) (string, error) {
    result, err := w.stateMachine.Execute(ctx, userInput)
    if err != nil {
        return "", fmt.Errorf("workflow execution failed: %w", err)
    }
    return result.Output, nil
}

func mustCreateAgent(config agent.Config) *agent.Agent {
    a, err := agent.New(config)
    if err != nil {
        panic(err)
    }
    return a
}

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

    workflow, err := NewCustomerServiceWorkflow(model)
    if err != nil {
        log.Fatalf("Failed to create workflow: %v", err)
    }

    // Test different scenarios
    testCases := []string{
        "I want to check my order",
        "Where is my package",
        "I want a refund",
        "Your service is terrible",
        "How do I use this product",
    }

    for _, input := range testCases {
        response, err := workflow.Handle(ctx, input)
        if err != nil {
            log.Printf("Error handling '%s': %v", input, err)
            continue
        }
        fmt.Printf("\nUser: %s\nAgent: %s\n", input, response)
    }
}

Dynamic Flow Orchestration

Runtime Flow Modification

// Support dynamic workflow modification at runtime
type DynamicWorkflow struct {
    stateMachine *StateMachine
    modifiers    []WorkflowModifier
    mu           sync.RWMutex
}

type WorkflowModifier interface {
    Modify(sm *StateMachine, ctx *ExecutionContext) error
}

// Add dynamic branch based on context
func (dw *DynamicWorkflow) AddConditionalBranch(
    fromState string,
    condition ConditionFunc,
    newBranch *State,
) error {
    dw.mu.Lock()
    defer dw.mu.Unlock()

    // Add new state
    dw.stateMachine.AddState(newBranch)

    // Add transition rule
    dw.stateMachine.AddTransition(fromState, newBranch.Name, condition, 5)

    return nil
}

// A/B test support
func (dw *DynamicWorkflow) EnableABTesting(
    stateName string,
    variantA, variantB *State,
    splitRatio float64,
) error {
    dw.mu.Lock()
    defer dw.mu.Unlock()

    // Add two variant states
    dw.stateMachine.AddState(variantA)
    dw.stateMachine.AddState(variantB)

    // Branch based on user ID hash
    dw.stateMachine.AddTransition(stateName, variantA.Name,
        func(ctx *ExecutionContext, result *StateResult) bool {
            userID := ctx.GetVariable("user_id").(string)
            hash := hashString(userID)
            return float64(hash%100)/100.0 < splitRatio
        }, 10)

    dw.stateMachine.AddTransition(stateName, variantB.Name,
        func(ctx *ExecutionContext, result *StateResult) bool {
            return true // fallback
        }, 5)

    return nil
}

Nested Workflows

Sub-Workflow Invocation

// Support nesting other workflows inside a Custom Workflow
type NestedWorkflowState struct {
    SubWorkflow  Workflow
    InputMapper  func(*ExecutionContext) string
    OutputMapper func(string, *ExecutionContext)
}

func (s *NestedWorkflowState) Execute(ctx context.Context, execCtx *ExecutionContext) (*StateResult, error) {
    // Map input
    subInput := s.InputMapper(execCtx)

    // Execute sub-workflow
    subResult, err := s.SubWorkflow.Execute(ctx, subInput)
    if err != nil {
        return nil, fmt.Errorf("sub-workflow failed: %w", err)
    }

    // Map output back to main context
    s.OutputMapper(subResult.Output, execCtx)

    return &StateResult{
        Output: subResult.Output,
        Data:   subResult.Metadata,
    }, nil
}

// Example: nest a refund sub-flow in the customer service system
refundSubWorkflow := team.NewSequentialWorkflow(
    team.WithStep(validateOrderStep),
    team.WithStep(checkRefundEligibilityStep),
    team.WithStep(processRefundStep),
)

sm.AddState(&team.State{
    Name: "refund_subflow",
    Type: team.StateTypeAction,
    Action: func(ctx *team.ExecutionContext) (*team.StateResult, error) {
        nested := &NestedWorkflowState{
            SubWorkflow: refundSubWorkflow,
            InputMapper: func(execCtx *team.ExecutionContext) string {
                return execCtx.GetVariable("order_id").(string)
            },
            OutputMapper: func(output string, execCtx *team.ExecutionContext) {
                execCtx.SetVariable("refund_result", output)
            },
        }
        return nested.Execute(context.Background(), ctx)
    },
})

Error Recovery and Compensating Transactions

Saga Pattern Implementation

// Saga transaction manager
type SagaManager struct {
    steps         []SagaStep
    compensations []Compensation
}

type SagaStep struct {
    Name        string
    Action      func() error
    Compensation func() error
}

func (s *SagaManager) Execute() error {
    completed := make([]int, 0)

    for i, step := range s.steps {
        if err := step.Action(); err != nil {
            // Execute compensations
            for j := len(completed) - 1; j >= 0; j-- {
                if compErr := s.steps[completed[j]].Compensation(); compErr != nil {
                    log.Printf("Compensation failed for step %s: %v", s.steps[completed[j]].Name, compErr)
                }
            }
            return fmt.Errorf("step %s failed: %w", step.Name, err)
        }
        completed = append(completed, i)
    }

    return nil
}

// Apply in Custom Workflow
func (sm *StateMachine) executeWithSaga(ctx context.Context, saga *SagaManager) (*StateResult, error) {
    if err := saga.Execute(); err != nil {
        return nil, err
    }
    return &StateResult{Output: "success"}, nil
}

Common Questions in Depth

Q: What is the difference between Custom Workflow and hard-coded if-else?

A: Custom Workflow provides:

Visualization: the state machine can be exported as a flowchart.
Testability: each state and transition can be tested independently.
Observability: execution history, current state, and transition paths can be traced.
Dynamic modification: adjust the flow at runtime without restarting.
Persistence: execution state can be saved to support resume from checkpoint.

Q: How do I avoid an overly complex state machine?

A: Follow these principles:

Single responsibility: each state does one thing.
Hierarchical decomposition: split complex flows into sub-workflows.
Clear naming: state names reflect business meaning.
Limit depth: state machine nesting should not exceed 3 levels.
Documentation: every transition condition must have a comment explaining it.

Q: How is state machine performance optimized?

State caching: preload frequently accessed states into memory.
Transition precompilation: compile condition functions to bytecode.
Parallel evaluation: evaluate multiple transition conditions in parallel.
State reuse: use the prototype pattern for states with identical logic.

Next Steps

You now have a deep understanding of Custom Workflow flexible orchestration. Next, explore Agent Routing—strategy design for dynamically selecting Agents, load balancing, and intelligent dispatch.

← Loop Workflow | Agent Routing →

Want to learn more Go ADK hands-on? Follow the “Mengshou Programming” channel for weekly updates on Go and AI programming.

Frequently Asked Questions

Why do I need a Custom Workflow?

When preset patterns such as Sequential, Parallel, and Loop cannot meet complex business needs, Custom Workflow lets you define any scheduling logic in code.

What is the difference between Custom Workflow and hard-coded if-else?

Custom Workflow provides visualization, testability, observability, runtime dynamic modification, and persistent resume-from-checkpoint capabilities.

How do I avoid an overly complex state machine?

Follow single responsibility, hierarchical decomposition, clear naming, limit nesting to no more than 3 levels, and add comments for every transition condition.

How can Custom Workflow state machine performance be optimized?

Improve performance through state caching, precompiled transition conditions, parallel evaluation of transition conditions, and state reuse via the prototype pattern.

How does Custom Workflow implement error recovery?

Introduce the Saga pattern to implement compensating transactions, executing compensating actions for already completed steps when a state execution fails.

Why Custom Workflow Is Needed#

Limitations of Preset Patterns#

Custom vs Orchestrator#

State Machine-Driven Architecture#

Core Abstractions#

State Execution Engine#

Hands-On Scenario: Intelligent Customer Service System#

Full Implementation#

Dynamic Flow Orchestration#

Runtime Flow Modification#

Nested Workflows#

Sub-Workflow Invocation#

Error Recovery and Compensating Transactions#

Saga Pattern Implementation#

Common Questions in Depth#

Q: What is the difference between Custom Workflow and hard-coded if-else?#

Q: How do I avoid an overly complex state machine?#

Q: How is state machine performance optimized?#

Next Steps#