Dynamic agent workflows

ADK에서 지원Python v2.0.0TypeScript v2.0.0Go v2.0.0

The ADK framework provides a programmatic way to define workflows as a more flexible and powerful alternative to graph-based workflows. Using a graph-based approach provides a convenient way to compose multi-step, static process structures with workflow nodes. However, if the logic path for your workflow is more complex, with iterative loops or complex branching logic, a graph-based approach may not suit your needs, or may become too unwieldy to manage.

Dynamic workflows in ADK allow you to put aside graph-based path structures and use the full power of your chosen programming language to build workflows. With dynamic workflows, you can create workflows with simple decorators (Python) or constructor functions (Go), invoke workflow nodes as functions, and build complex routing logic. Here are some of the benefits of dynamic workflows in ADK:

• Flexible Control Flow: Define execution order dynamically using loops, conditionals, and recursion which are difficult or impossible to represent in static graphs.
• Programmatic Experience: Use familiar constructs like while loops and async/await (Python) or for loops and workflow.RunNode (Go) instead of graph-based routing.
• Automatic Checkpointing: Dynamic workflows track each node execution. Successful sub-nodes are automatically skipped when resuming the workflow, making complex logic durable and resumable by default.
• Encapsulation: Wrap business logic into parent nodes that internally compose lower-level nodes, keeping the overall workflow clean and manageable.

시작하기

The following dynamic workflow code example shows how to define a basic workflow containing a single node with a function:

PythonTypeScriptGo

from google.adk import Context
from google.adk import Workflow
from google.adk.workflow import node
from typing import Any

@node(name="hello_node")
def my_node(node_input: Any):
    return "Hello World"

# define a dynamic workflow node
@node(rerun_on_resume=True)
async def my_workflow(ctx: Context, node_input: str) -> str:
    # run_node executes a node and returns its output
    result = await ctx.run_node(my_node, node_input="hello")
    return result

# Run the workflow
root_agent = Workflow(
    name="root_agent",
    edges=[("START", my_workflow)],
)

This example uses the @node annotation for convenience and to keep the written code as simple as possible. This annotation generates wrappers that allow the code to be run in the context of an ADK dynamic workflow.

TypeScript에는 @node 데코레이터가 없습니다. 대신 node(fn, options) 팩토리 함수를 사용하세요. ctx.runNode() 메서드는 ctx.run_node()에 해당합니다:

import { node, NodeContext, Workflow } from '@google/adk';

const myNode = node(() => 'Hello World', { name: 'hello_node' });

const myWorkflow = node(
  async (ctx: NodeContext, _nodeInput: string) => {
    const result = await ctx.runNode(myNode, 'hello');
    return result.output;
  },
  { name: 'my_workflow', rerunOnResume: true },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', myWorkflow]],
});

오케스트레이터 노드를 작성할 때, 결과를 읽는 방법과 일시 중지 후 워크플로가 작동하는 방식에 영향을 미치는 두 가지 세부사항이 있습니다:

• ctx.runNode() 메서드는 출력 값이 아닌 노드 결과(node result)로 확인(resolve)됩니다. 값을 가져오려면 .output 프로퍼티를 읽으세요.
• ctx.runNode()를 호출하는 오케스트레이터는 반드시 rerunOnResume: true를 설정해야 합니다. 이 설정은 재개 시 노드 본문을 다시 실행하므로, 이미 완료된 하위 노드는 다시 실행되지 않고 체크포인트에서 재생됩니다.

In Go, workflow.NewFunctionNode replaces the @node decorator and workflow.NewDynamicNode replaces the @node(rerun_on_resume=True) async orchestrator. workflow.RunNode is the direct equivalent of ctx.run_node(). workflowagent.New with workflow.Chain replaces Workflow(edges=[...]).

Resume behaviour after a human-in-the-loop pause is controlled by NodeConfig.RerunOnResume — see Nodes below for details.

// helloNode is a simple FunctionNode that returns "Hello World".
// In Python this would be written as:
//
//  @node(name="hello_node")
//  def my_node(node_input: Any):
//      return "Hello World"
//
// In Go, workflow.NewFunctionNode wraps the same logic with the
// required node interface, inferring input and output types from
// the generic parameters.
var helloNode = workflow.NewFunctionNode("hello_node",
    func(_ agent.Context, _ string) (string, error) {
        return "Hello World", nil
    },
    workflow.NodeConfig{},
)

// myWorkflow is a dynamic orchestrator node. It calls workflow.RunNode
// to schedule helloNode as a child and returns its output.
// In Python this would be:
//
//  @node(rerun_on_resume=True)
//  async def my_workflow(ctx: Context, node_input: str) -> str:
//      result = await ctx.run_node(my_node, node_input="hello")
//      return result
//
// workflow.NewDynamicNode defaults RerunOnResume to &true, matching the
// Python @node(rerun_on_resume=True) behaviour.
var myWorkflow = workflow.NewDynamicNode[string, string]("my_workflow",
    func(ctx agent.Context, _ string, _ func(*session.Event) error) (string, error) {
        return workflow.RunNode[string](ctx, helloNode, "hello")
    },
    workflow.NodeConfig{},
)

func runGetStarted() error {
    ctx := context.Background()

    // workflowagent.New creates an agent.Agent backed by the workflow engine.
    // workflow.Chain(workflow.Start, myWorkflow) produces the edges slice
    // equivalent to Python's edges=[("START", my_workflow)].
    wa, err := workflowagent.New(workflowagent.Config{
        Name:        "root_agent",
        Description: "A minimal dynamic workflow.",
        Edges:       workflow.Chain(workflow.Start, myWorkflow),
    })
    if err != nil {
        return fmt.Errorf("workflowagent.New: %w", err)
    }

    l := full.NewLauncher()
    return l.Execute(ctx, &launcher.Config{
        AgentLoader: agent.NewSingleLoader(wa),
    }, os.Args[1:])
}

빌딩 블록: 노드 및 워크플로

Nodes and workflows represent the basic building blocks of ADK's dynamic workflows. These types and functions provide the functionality required to wrap your code so it can be integrated into code-based workflows in ADK.

노드

A dynamic workflow in ADK is composed of nodes. A simple version of a usable workflow node wraps a plain function with the metadata required to run within a workflow.

PythonTypeScriptGo

In Python, the @node annotation generates the node wrapper, keeping boilerplate to a minimum:

@node(name="hello_node")
def my_function_node(node_input: Any):
    return "Hello World"

The following code snippet shows the equivalent code without the @node annotation:

# base function
def my_function_node(node_input: Any):
    return "Hello World"

# FunctionNode wrapper with options
success_node = FunctionNode(
    my_function_node,
    name="hello",
    rerun_on_resume=True,
)

Creating the node wrapper code yourself can be useful if you are wrapping functions from an external library, need to create multiple nodes from the same function with different configurations, or if you are managing node references in a registry for advanced orchestration.

노드를 빌드하는 데는 node(fn, options) 팩토리 함수와 명시적인 new FunctionNode(name, fn, config) 생성자의 두 가지 방법이 있습니다. 다른 라이브러리의 함수를 래핑하거나, 하나의 함수에서 서로 다르게 구성된 여러 노드가 필요하거나, 고급 오케스트레이션을 위해 레지스트리에서 노드 참조를 관리할 때는 생성자를 사용하세요.

import { FunctionNode, node, NodeContext, Workflow } from '@google/adk';

/** The plain function both node forms wrap. */
function myFunctionNode(_ctx: NodeContext, nodeInput: unknown): string {
  return `Hello ${nodeInput ?? 'World'}`;
}

const helloNode = node(myFunctionNode, { name: 'hello_node' });

const successNode = new FunctionNode('hello', myFunctionNode, {
  rerunOnResume: true,
});

이 코드 샘플에서 가장 중요한 옵션은 rerunOnResume이며, 이는 인간 참여형(human-in-the-loop) 일시 중지 후 워크플로가 재개될 때 발생하는 동작을 제어합니다:

• true (재진입, re-entry): 노드 본문이 처음부터 다시 실행됩니다. ctx.runNode()를 호출하는 모든 오케스트레이터에 이 설정을 사용하세요. 본문이 다시 실행되며 이미 완료된 하위 노드 활성화는 자동으로 건너뜁니다.
• false (핸드오프, 리프 노드 기본값): 재개 페이로드가 중단된 노드를 우회하여 해당 노드의 후속 노드에 입력으로 라우팅됩니다.

In Go, workflow.NewFunctionNode[IN, OUT] wraps a plain function as a workflow node, inferring input and output types from the generic parameters. There is no decorator syntax; the node is a value that you pass as a child to workflow.RunNode inside a dynamic orchestrator:

// myFunctionNode demonstrates the explicit NewFunctionNode constructor —
// equivalent to wrapping a function in a FunctionNode manually in Python:
//
//  success_node = FunctionNode(my_function_node, name="hello", rerun_on_resume=True)
//
// Creating the node directly (rather than via @node) is useful when you
// need multiple nodes from the same function with different configurations,
// or when wrapping functions from an external library.
var myFunctionNode = workflow.NewFunctionNode("hello",
    func(_ agent.Context, _ any) (string, error) {
        return "Hello World", nil
    },
    workflow.NodeConfig{},
)

// myFormattingNode is a second function node that the dynamic orchestrator
// calls in sequence, mirroring:
//
//  result_formatted = await ctx.run_node(my_formatting_node, node_input=result)
var myFormattingNode = workflow.NewFunctionNode("format",
    func(_ agent.Context, in string) (string, error) {
        return fmt.Sprintf("[formatted] %s", in), nil
    },
    workflow.NodeConfig{},
)

NodeConfig holds the same options as Python's @node arguments. The most important field is RerunOnResume *bool, which controls what happens when a workflow resumes after a human-in-the-loop pause:

• &true (re-entry mode): the interrupted node is re-run from the beginning on resume. Use this for dynamic orchestrator nodes that call workflow.RunNode in a loop — the body re-executes and already-completed child activations are skipped automatically (checkpointing). This mirrors Python's @node(rerun_on_resume=True).
• &false (handoff mode): the resume payload is routed directly to the node's successor as input, bypassing the interrupted node entirely. Use this for leaf nodes that simply emit a pause event and expect the human response to flow to the next step.
• nil: the default depends on node type. workflow.NewDynamicNode automatically sets nil → &true (re-entry mode), because an orchestrator body must be re-entered on resume to deliver cached child results. workflow.NewFunctionNode and other leaf node constructors leave nil as-is, which the engine treats as handoff (&false). Explicit &false is always respected on any node type.

// NewDynamicNode: nil RerunOnResume is automatically set to &true.
// Passing &rerun explicitly is equivalent and makes the intent clear.
rerun := true
orchestratorNode := workflow.NewDynamicNode[string, string]("my_workflow",
    myOrchestratorfn,
    workflow.NodeConfig{RerunOnResume: &rerun}, // re-entry: node body re-runs on resume
)

// NewFunctionNode: nil RerunOnResume stays nil → engine treats as handoff.
handoffNode := workflow.NewFunctionNode("leaf_node",
    myLeafFn,
    workflow.NodeConfig{}, // nil RerunOnResume → handoff for FunctionNode
)

워크플로

In an ADK dynamic workflow, you use a dynamic node as the primary orchestrator for nodes. A dynamic node manages running child nodes and the execution logic (order and paths) for those nodes.

PythonTypeScriptGo

@node(rerun_on_resume=True)
async def my_workflow(ctx):
    # run_node executes a node and returns its output
    result = await ctx.run_node(my_function_node, node_input="Hello")
    result_formatted = await ctx.run_node(my_formatting_node, node_input=result)
    return result_formatted

# Run the workflow
root_agent = Workflow(
    name="root_agent",
    edges=[("START", my_workflow)],
)

오케스트레이터는 각 하위 단계에 대해 ctx.runNode()를 await하는 비동기 함수입니다. 이를 rerunOnResume: true로 노드로 래핑하고 그래프의 유일한 edge로 사용합니다:

const myFormattingNode = node(
  (_ctx: NodeContext, nodeInput: string) => `>> ${nodeInput.trim()} <<`,
  { name: 'my_formatting_node' },
);

const myWorkflow = node(
  async (ctx: NodeContext, nodeInput: unknown) => {
    const greeted = await ctx.runNode(helloNode, nodeInput);
    const again = await ctx.runNode(successNode, greeted.output);
    const formatted = await ctx.runNode(myFormattingNode, again.output);
    return formatted.output;
  },
  { name: 'my_workflow', rerunOnResume: true },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', myWorkflow]],
});

workflow.NewDynamicNode creates an orchestrator whose body calls workflow.RunNode for each child step. workflowagent.New with workflow.Chain(workflow.Start, myWorkflow) is the equivalent of Workflow(edges=[("START", my_workflow)]):

// orchestratorWorkflow is a dynamic node that schedules two children in
// sequence via workflow.RunNode, equivalent to:
//
//  @node(rerun_on_resume=True)
//  async def my_workflow(ctx):
//      result = await ctx.run_node(my_function_node, node_input="Hello")
//      result_formatted = await ctx.run_node(my_formatting_node, node_input=result)
//      return result_formatted
var orchestratorWorkflow = workflow.NewDynamicNode[string, string]("my_workflow",
    func(ctx agent.Context, _ string, _ func(*session.Event) error) (string, error) {
        result, err := workflow.RunNode[string](ctx, myFunctionNode, "Hello")
        if err != nil {
            return "", err
        }
        return workflow.RunNode[string](ctx, myFormattingNode, result)
    },
    workflow.NodeConfig{},
)

데이터 처리

When using dynamic workflows with ADK, passing data is simpler than graph-based workflows because workflow.RunNode returns the child node's output directly as a typed Go value — eliminating the need to manually read and write session state keys for data transfer.

PythonTypeScriptGo

from google.adk import Context
from google.adk.workflow import node

@node(rerun_on_resume=True)
async def editorial_workflow(ctx: Context, user_request: str):
    # Agent Node generates output
    raw_draft = await ctx.run_node(draft_agent, user_request)

    # Function Node formats text
    formatted_text = await ctx.run_node(format_function_node, raw_draft)

    return formatted_text

You can also pass specific data schemas using a defined class and configure input and output schemas, similar to graph-based workflow nodes:

from google.adk import Agent
from google.adk import Context
from google.adk.workflow import node
from pydantic import BaseModel

class CityTime(BaseModel):
    time_info: str  # time information
    city: str       # city name

@node
def city_time_function(city: str):
    """Simulate returning the current time in a specified city."""
    return CityTime(time_info="10:10 AM", city=city)

city_report_agent = Agent(
    name="city_report_agent",
    model="gemini-flash-latest",
    input_schema=CityTime,
    instruction="""output the data provided by the previous node.""",
)

@node # workflow node
async def city_workflow(ctx: Context):
    city_time = await ctx.run_node(city_time_function, "Paris")
    report_text = await ctx.run_node(city_report_agent, city_time)

    return report_text

ctx.runNode() 함수는 하위 노드의 결과를 직접 반환하므로, 값을 한 단계 다운스트림으로 이동하기 위해 세션 상태 키를 읽고 쓸 필요가 없습니다. 이 함수는 LlmAgent를 포함하여 노드 유사(node-like) 값을 먼저 node()로 래핑하지 않고도 허용합니다:

import { LlmAgent, node, NodeContext, Workflow } from '@google/adk';

const draftAgent = new LlmAgent({
  name: 'draft_agent',
  model: 'gemini-flash-latest',
  instruction: 'Write a short draft for the user request.',
});

const formatFunctionNode = node(
  (_ctx: NodeContext, rawDraft: string) =>
    rawDraft
      .split('\n')
      .map((line) => line.trim())
      .filter(Boolean)
      .map((line) => `| ${line}`)
      .join('\n'),
  { name: 'format_function_node' },
);

const editorialWorkflow = node(
  async (ctx: NodeContext, userRequest: string) => {
    const rawDraft = await ctx.runNode(draftAgent, userRequest);

    const formattedText = await ctx.runNode(
      formatFunctionNode,
      rawDraft.output,
    );

    return formattedText.output;
  },
  { name: 'editorial_workflow', rerunOnResume: true },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', editorialWorkflow]],
});

스키마는 그래프에서와 동일하게 작동합니다. 순차 라우트 섹션에 표시된 것처럼 실행하는 노드에 스키마를 연결하세요.

In Go, workflow.NewAgentNode wraps an agent.Agent so it can be invoked via workflow.RunNode inside a dynamic orchestrator. The output of each RunNode call is returned as a typed value — no session state reads are required:

// newDataHandlingWorkflow demonstrates how to pass data between a dynamic
// orchestrator and an LlmAgent-backed node. workflow.NewAgentNode wraps an
// agent.Agent so it can be invoked via workflow.RunNode.
//
// In Python this mirrors:
//
//  city_report_agent = Agent(name="city_report_agent", ...)
//  @node
//  async def city_workflow(ctx: Context):
//      city_time = await ctx.run_node(city_time_function, "Paris")
//      report_text = await ctx.run_node(city_report_agent, city_time)
//      return report_text
func newDataHandlingWorkflow(ctx context.Context) (agent.Agent, error) {
    model, err := gemini.NewModel(ctx, "gemini-flash-latest", &genai.ClientConfig{})
    if err != nil {
        return nil, fmt.Errorf("gemini.NewModel: %w", err)
    }

    // cityTimeNode is a FunctionNode that returns a formatted city-time string.
    cityTimeNode := workflow.NewFunctionNode("city_time_function",
        func(_ agent.Context, city string) (string, error) {
            return fmt.Sprintf("10:10 AM in %s", city), nil
        },
        workflow.NodeConfig{},
    )

    // cityReportAgent is an LlmAgent that receives the city-time string and
    // produces a human-friendly report.
    cityReportAgent, err := llmagent.New(llmagent.Config{
        Name:        "city_report_agent",
        Model:       model,
        Description: "Reports city time information.",
        Instruction: "Output the data provided by the previous node in a friendly sentence.",
    })
    if err != nil {
        return nil, fmt.Errorf("llmagent.New (cityReport): %w", err)
    }

    // workflow.NewAgentNode wraps cityReportAgent so it can be called from
    // inside a dynamic node via workflow.RunNode.
    cityReportNode, err := workflow.NewAgentNode(cityReportAgent, workflow.NodeConfig{})
    if err != nil {
        return nil, fmt.Errorf("workflow.NewAgentNode: %w", err)
    }

    cityWorkflow := workflow.NewDynamicNode[string, string]("city_workflow",
        func(ctx agent.Context, _ string, _ func(*session.Event) error) (string, error) {
            cityTime, err := workflow.RunNode[string](ctx, cityTimeNode, "Paris")
            if err != nil {
                return "", err
            }
            return workflow.RunNode[string](ctx, cityReportNode, cityTime)
        },
        workflow.NodeConfig{},
    )

    return workflowagent.New(workflowagent.Config{
        Name:      "data_handling_workflow",
        SubAgents: []agent.Agent{cityReportAgent},
        Edges:     workflow.Chain(workflow.Start, cityWorkflow),
    })
}

For more information on data handling between workflow nodes, see Data handling for agent workflows.

워크플로 경로

Dynamic workflows in ADK provide more flexibility in terms of routing logic compared to graph-based workflows, including iterative loops or more complex branching logic. This section describes some of the techniques that you can use for routing.

Sequence route

You can create sequential task processing with dynamic workflows in ADK, just as you can with graph-based workflows.

PythonTypeScriptGo

The following code snippet shows a dynamic workflow with an agent, a function node, and a second agent:

@node # workflow node
async def city_workflow(ctx: Context):
    city = await ctx.run_node(city_generator_agent)
    city_time = await ctx.run_node(city_time_function, city)
    report_text = await ctx.run_node(city_report_agent, city_time)

    return report_text

순차 라우트는 ctx.runNode() 호출을 차례로 await합니다. 각 호출은 다음 호출이 시작되기 전에 완료됩니다:

import { LlmAgent, node, NodeContext, Workflow } from '@google/adk';
import { z } from 'zod';

const cityTimeSchema = z.object({
  timeInfo: z.string().describe('Time information.'),
  city: z.string().describe('City name.'),
});
type CityTime = z.infer<typeof cityTimeSchema>;

const cityGeneratorAgent = new LlmAgent({
  name: 'city_generator_agent',
  model: 'gemini-flash-latest',
  instruction: 'Return the name of a random city. Return only the name.',
});

/** Simulates returning the current time in a specified city. */
const cityTimeFunction = node(
  (_ctx: NodeContext, city: string): CityTime => ({
    timeInfo: '10:10 AM',
    city: city.trim(),
  }),
  { name: 'city_time_function', outputSchema: cityTimeSchema },
);

const cityReportAgent = node(
  new LlmAgent({
    name: 'city_report_agent',
    model: 'gemini-flash-latest',
    instruction: 'Output the data provided by the previous node as a sentence.',
  }),
  { inputSchema: cityTimeSchema },
);

const cityWorkflow = node(
  async (ctx: NodeContext) => {
    const city = await ctx.runNode(cityGeneratorAgent);
    const cityTime = await ctx.runNode(cityTimeFunction, city.output);
    const reportText = await ctx.runNode(cityReportAgent, cityTime.output);

    return reportText.output;
  },
  { name: 'city_workflow', rerunOnResume: true },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', cityWorkflow]],
});

Call workflow.RunNode sequentially inside a NewDynamicNode body — each call awaits the child before the next one starts. The data handling example above demonstrates exactly this pattern: cityWorkflow calls workflow.RunNode for cityTimeNode and then cityReportNode in order, passing each node's typed output to the next.

Loop route

For workflows where you want to use an iterative loop for a task, dynamic workflows offer much more flexibility to define the routing logic you need.

PythonTypeScriptGo

The following code example shows how to use dynamic workflows to construct a workflow loop for generating, reviewing, and updating code:

from google.adk import Context
from google.adk import Event
from google.adk.agents import LlmAgent
from google.adk.workflow import node

coder_agent = LlmAgent(
    name="generator_agent",
    model="gemini-flash-latest",
    instruction="Write python code for user request.",
)

@node(name="lint_reviewer")
async def compile_lint_check(ctx: Context, code: str):
    # Simulate API call or lint check
    class Response:
        findings = ""
    return Response()

fixer_agent = LlmAgent(
    name="fixer_agent",
    model="gemini-flash-latest",
    instruction="""Refactor current code {code}.
        Based on compile & lint review: {findings}""",
)

@node # workflow node
async def code_workflow(ctx: Context, user_request: str):
  code = await ctx.run_node(coder_agent, user_request)
  check_resp = await ctx.run_node(compile_lint_check, code)

  while check_resp.findings:
    yield Event(state={"code": code, "findings": check_resp.findings})
    code = await ctx.run_node(fixer_agent, {"code": code, "findings": check_resp.findings})

    check_resp = await ctx.run_node(compile_lint_check, code)

  yield Event(output=code)

동적 워크플로는 그래프의 역방향 에지 대신 일반 루프로 반복을 정의하여 워크플로 로직을 단순하게 유지하는 데 도움이 됩니다. 값은 로컬 변수에 유지되며, 상태는 에이전트의 지침 템플릿에서 다시 읽어야 하는 경우에만 기록됩니다. 그래프 사이클과 달리 루프는 루프 조건에 의해 바운딩(제한)됩니다:

import { LlmAgent, node, NodeContext, Workflow } from '@google/adk';

/** Safety bound on the refine loop. */
const MAX_FIX_ROUNDS = 3;

const coderAgent = new LlmAgent({
  name: 'generator_agent',
  model: 'gemini-flash-latest',
  instruction: 'Write TypeScript code for the user request. Output code only.',
});

/** Simulates a compile / lint pass. Empty findings means "clean". */
const compileLintCheck = node(
  (_ctx: NodeContext, code: string) => {
    const findings: string[] = [];
    if (!/\/\*\*/.test(code)) {
      findings.push('every function needs a JSDoc comment');
    }
    if (!/\)\s*:\s*\w/.test(code)) {
      findings.push('add return type annotations');
    }
    return { findings: findings.join('; ') };
  },
  { name: 'lint_reviewer' },
);

const fixerAgent = new LlmAgent({
  name: 'fixer_agent',
  model: 'gemini-flash-latest',
  instruction: `Refactor current code {code}.
      Based on compile & lint review: {findings}
      Output code only.`,
});

const codeWorkflow = node(
  async (ctx: NodeContext, userRequest: string) => {
    let code = (await ctx.runNode(coderAgent, userRequest)).output as string;
    let checkResp = (await ctx.runNode(compileLintCheck, code)).output as {
      findings: string;
    };

    for (let round = 0; checkResp.findings && round < MAX_FIX_ROUNDS; round++) {
      ctx.state.set('code', code);
      ctx.state.set('findings', checkResp.findings);

      code = (
        await ctx.runNode(fixerAgent, { code, findings: checkResp.findings })
      ).output as string;
      checkResp = (await ctx.runNode(compileLintCheck, code)).output as {
        findings: string;
      };
    }

    return code;
  },
  { name: 'code_workflow', rerunOnResume: true },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', codeWorkflow]],
});

In Go, the loop is a plain for loop inside the dynamic node body. The lint check node returns an empty string when there are no findings, which signals the loop to exit:

// newLoopWorkflow demonstrates an iterative loop inside a dynamic node.
// The orchestrator body uses a plain Go for loop to keep calling the
// lintCheckNode until there are no findings — equivalent to Python's:
//
//  @node
//  async def code_workflow(ctx: Context, user_request: str):
//      code = await ctx.run_node(coder_agent, user_request)
//      check_resp = await ctx.run_node(compile_lint_check, code)
//      while check_resp.findings:
//          code = await ctx.run_node(fixer_agent, ...)
//          check_resp = await ctx.run_node(compile_lint_check, code)
//      return code
func newLoopWorkflow(ctx context.Context) (agent.Agent, error) {
    model, err := gemini.NewModel(ctx, "gemini-flash-latest", &genai.ClientConfig{})
    if err != nil {
        return nil, fmt.Errorf("gemini.NewModel: %w", err)
    }

    coderAgent, err := llmagent.New(llmagent.Config{
        Name:        "generator_agent",
        Model:       model,
        Description: "Writes Go code for the user request.",
        Instruction: "Write Go code for the user request. Output only the code.",
        OutputKey:   "generated_code",
    })
    if err != nil {
        return nil, fmt.Errorf("llmagent.New (coder): %w", err)
    }

    coderNode, err := workflow.NewAgentNode(coderAgent, workflow.NodeConfig{})
    if err != nil {
        return nil, fmt.Errorf("workflow.NewAgentNode (coder): %w", err)
    }

    // lintCheckNode simulates a lint/compile check. It returns an empty
    // string when there are no findings, signalling the loop to exit.
    lintCheckNode := workflow.NewFunctionNode("lint_reviewer",
        func(_ agent.Context, code string) (string, error) {
            // Simulate a lint check: return findings or empty string when clean.
            if len(code) < 50 {
                return "Code is too short; add error handling.", nil
            }
            return "", nil // no findings — loop exits
        },
        workflow.NodeConfig{},
    )

    fixerAgent, err := llmagent.New(llmagent.Config{
        Name:        "fixer_agent",
        Model:       model,
        Description: "Refactors code based on lint findings.",
        Instruction: "Refactor the provided code to address the review findings. Output only the improved code.",
    })
    if err != nil {
        return nil, fmt.Errorf("llmagent.New (fixer): %w", err)
    }

    fixerNode, err := workflow.NewAgentNode(fixerAgent, workflow.NodeConfig{})
    if err != nil {
        return nil, fmt.Errorf("workflow.NewAgentNode (fixer): %w", err)
    }

    codeWorkflow := workflow.NewDynamicNode[string, string]("code_workflow",
        func(ctx agent.Context, userRequest string, _ func(*session.Event) error) (string, error) {
            code, err := workflow.RunNode[string](ctx, coderNode, userRequest)
            if err != nil {
                return "", err
            }

            findings, err := workflow.RunNode[string](ctx, lintCheckNode, code)
            if err != nil {
                return "", err
            }

            // Loop until the lint check reports no findings.
            for findings != "" {
                code, err = workflow.RunNode[string](ctx, fixerNode, code)
                if err != nil {
                    return "", err
                }
                findings, err = workflow.RunNode[string](ctx, lintCheckNode, code)
                if err != nil {
                    return "", err
                }
            }
            return code, nil
        },
        workflow.NodeConfig{},
    )

    return workflowagent.New(workflowagent.Config{
        Name:      "code_pipeline",
        SubAgents: []agent.Agent{coderAgent, fixerAgent},
        Edges:     workflow.Chain(workflow.Start, codeWorkflow),
    })
}

Parallel execution routes

Dynamic workflows in ADK can support parallel execution.

PythonTypeScriptGo

In Python, you can use asyncio.gather to build parallel execution:

import asyncio
from typing import Any
from google.adk import Context
from google.adk.workflow import BaseNode, node


@node(rerun_on_resume=True)
async def parallel_supervisor(
    ctx: Context, node_input: list[Any], real_node: BaseNode
):
    """Runs a worker node in parallel for each item in the input list."""
    tasks = []
    for item in node_input:
        # ctx.run_node returns a future. Append instead of awaiting immediately.
        tasks.append(ctx.run_node(real_node, item))

    # Collect all results in parallel
    results = await asyncio.gather(*tasks)
    return results

Tip: Resuming parallel nodes

The workflow framework ensures that if a dynamic workflow is resumed, only failed or interrupted worker nodes are re-executed, including parallel worker nodes.

ctx.runNode() 메서드는 Promise를 반환하므로, 어떤 하위 노드도 await하기 전에 모든 하위 노드를 시작하면 하위 노드들이 동시에 실행되고 Promise.all이 결과를 수집합니다. 실행 ID(Run ID)는 호출 순서대로 할당되므로 재개 전반에서 ID를 결정론적으로 유지하려면 동기 루프에서 하위 노드를 시작하세요:

import { node, NodeContext, Workflow } from '@google/adk';

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

/** The worker run once per list item. */
const realNode = node(
  async (_ctx: NodeContext, item: string) => {
    await sleep(200);
    return { item, length: item.length };
  },
  { name: 'analyze_item' },
);

const parallelSupervisor = node(
  async (ctx: NodeContext, nodeInput: string) => {
    const items = nodeInput
      .split(',')
      .map((item) => item.trim())
      .filter(Boolean);

    const tasks = items.map((item) => ctx.runNode(realNode, item));
    const results = await Promise.all(tasks);

    return results.map((result) => result.output);
  },
  { name: 'parallel_supervisor', rerunOnResume: true },
);

const summarize = node(
  (_ctx: NodeContext, results: Array<{ item: string; length: number }>) =>
    results.map((r) => `${r.item}: ${r.length} chars`).join('\n'),
  { name: 'summarize' },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', parallelSupervisor, summarize]],
});

팁: 내장 병렬 워커를 선호하세요

목록의 각 항목에 대해 하나의 노드를 실행하려면 node(worker, {parallelWorker: true, maxParallelWorkers: 4})를 사용하세요. 이 옵션은 팬아웃을 수행하고 동시성을 제한하며 기본값은 8입니다. 커스텀 스케줄링이나 부분 실패 처리가 필요한 경우 위에 표시된 수동 방식을 사용하세요. 재개 시 두 경우 모두 실패하거나 중단된 워커만 다시 실행됩니다.

In Go, workflow.NewParallelWorker wraps a child node and runs it concurrently for each element of a list input, collecting results into a single output slice. The maxConcurrency parameter caps how many concurrent activations may run simultaneously; 0 means unlimited:

// newParallelWorkflow demonstrates parallel execution using
// workflow.NewParallelWorker. The worker node runs a wrapped child node
// concurrently for each element in a list input, collecting results.
//
// This is the Go equivalent of using asyncio.gather in Python:
//
//  @node(rerun_on_resume=True)
//  async def parallel_supervisor(ctx, node_input, real_node):
//      tasks = [ctx.run_node(real_node, item) for item in node_input]
//      results = await asyncio.gather(*tasks)
//      return results
func newParallelWorkflow() (agent.Agent, error) {
    // workerNode processes a single item. NewParallelWorker will call it
    // once per element of the list input, concurrently.
    workerNode := workflow.NewFunctionNode("worker",
        func(_ agent.Context, item string) (string, error) {
            return fmt.Sprintf("processed: %s", item), nil
        },
        workflow.NodeConfig{},
    )

    // NewParallelWorker wraps workerNode so it runs concurrently for each
    // element of a []string input. maxConcurrency=0 means unlimited.
    parallelWorker, err := workflow.NewParallelWorker(
        "parallel_supervisor",
        workerNode,
        0, // maxConcurrency: 0 = unlimited
        workflow.NodeConfig{},
    )
    if err != nil {
        return nil, fmt.Errorf("workflow.NewParallelWorker: %w", err)
    }

    return workflowagent.New(workflowagent.Config{
        Name:        "parallel_workflow",
        Description: "Runs a worker node in parallel for each item in the input list.",
        Edges:       workflow.Chain(workflow.Start, parallelWorker),
    })
}

Tip: Resuming parallel nodes

The workflow framework ensures that if a dynamic workflow is resumed, only failed or interrupted worker nodes are re-executed, including parallel worker nodes managed by NewParallelWorker.

사람의 입력 (Human Input)

Dynamic workflows in ADK can also include human input or human in the loop (HITL) steps.

PythonTypeScriptGo

You build human input into workflows by yielding a RequestInput from a node, which pauses the workflow and waits for user input. The following code example shows how to build a human input node and include it in a workflow:

from typing import Any
from google.adk import Context
from google.adk.events import RequestInput
from google.adk.workflow import node


@node(rerun_on_resume=False)
async def get_user_approval(ctx: Context, node_input: Any):
    """Yields a RequestInput to pause the workflow and wait for user input."""
    yield RequestInput(message="Please approve this request (Yes/No)")


@node(rerun_on_resume=True)
async def handle_process(ctx: Context, node_input: Any):
    """The orchestrator calling the interactive step."""
    user_response = await ctx.run_node(get_user_approval)

    if user_response.lower() == "yes":
        return "Approved"
    return "Denied"

Important: Parent nodes with ctx.run_node

Parent nodes in dynamic workflows that call ctx.run_node must set rerun_on_resume=True to handle interruptions properly.

리프 노드는 워크플로를 일시 중지하기 위해 RequestInput을 반환하고 기본값인 rerunOnResume: false를 유지하여 응답이 출력이 되도록 합니다. 이를 호출하는 오케스트레이터는 반드시 rerunOnResume: true를 설정해야 합니다:

import { node, NodeContext, RequestInput, Workflow } from '@google/adk';

/**
 * Pauses the workflow and waits for user input.
 *
 * `rerunOnResume: false` (the default, spelled out here because it is the
 * point) is what makes this a one-liner: the reply is handed to the node as
 * its output instead of the body running a second time to collect it.
 */
const getUserApproval = node(
  () => new RequestInput({ message: 'Please approve this request (Yes/No)' }),
  { name: 'get_user_approval', rerunOnResume: false },
);

/** The orchestrator calling the interactive step. */
const handleProcess = node(
  async (ctx: NodeContext, nodeInput: unknown) => {
    const approval = await ctx.runNode(getUserApproval, nodeInput);

    if (approval.interruptIds.length > 0) {
      return undefined;
    }

    const userResponse = String(approval.output ?? '')
      .trim()
      .toLowerCase();

    if (userResponse === 'yes') {
      return 'Approved';
    }
    return 'Denied';
  },
  { name: 'handle_process', rerunOnResume: true },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', handleProcess]],
});

중요: 결정을 내리기 전에 interruptIds를 확인하세요

ctx.runNode() 메서드는 하위 노드가 중단될 때 오류를 발생시키지 않습니다. 정상적으로 반환되며 결과의 interruptIds 프로퍼티가 채워지고 output 프로퍼티는 여전히 undefined 상태로 남습니다. 결과를 사용하기 전에 interruptIds를 확인하세요. 이 검사를 건너뛰는 오케스트레이터는 누락된 출력을 답변으로 취급하고 사용자가 제공하지 않은 값으로 계속 진행합니다.

In Go, use workflow.NewEmittingFunctionNode with workflow.ResumeOrRequestInput to implement the re-entry HITL pattern. On the first pass ResumeOrRequestInput emits a session.RequestInput event and returns ErrNodeInterrupted, pausing the workflow. After the human replies, the node is re-run from the top (RerunOnResume: &true) and ResumeOrRequestInput returns the human's reply directly:

// newHITLWorkflow demonstrates the re-entry HITL pattern using
// workflow.ResumeOrRequestInput. On the first pass the node emits a
// RequestInput event and returns ErrNodeInterrupted (pausing the workflow).
// After the human replies, the same node is re-run from the top
// (RerunOnResume=&true) and ResumeOrRequestInput returns the human's reply.
//
// In Python this is equivalent to:
//
//  @node(rerun_on_resume=True)
//  async def get_user_approval(ctx, node_input):
//      yield RequestInput(message="Please approve this request (Yes/No)")
//
//  @node(rerun_on_resume=True)
//  async def handle_process(ctx, node_input):
//      user_response = await ctx.run_node(get_user_approval)
//      if user_response.lower() == "yes":
//          return "Approved"
//      return "Denied"
func newHITLWorkflow() (agent.Agent, error) {
    rerun := true

    // approvalNode pauses on the first pass to ask the user for a Yes/No
    // approval, then resolves their decision on resume.
    // workflow.ResumeOrRequestInput handles both phases.
    approvalNode := workflow.NewEmittingFunctionNode[any, any]("get_user_approval",
        func(nc agent.Context, _ any, emit func(*session.Event) error) (any, error) {
            // ResumeOrRequestInput: on first pass, emits the prompt and
            // returns ErrNodeInterrupted. On re-run after the human replies,
            // it returns the reply payload directly.
            reply, err := workflow.ResumeOrRequestInput(nc, emit, session.RequestInput{
                InterruptID: "user_approval",
                Message:     "Please approve this request (Yes/No)",
            })
            if err != nil {
                return nil, err
            }

            response, _ := reply.(string)
            if response == "" {
                response = "No"
            }
            if response == "yes" || response == "Yes" {
                return "Approved", nil
            }
            return "Denied", nil
        },
        workflow.NodeConfig{RerunOnResume: &rerun},
    )

    return workflowagent.New(workflowagent.Config{
        Name:        "hitl_workflow",
        Description: "Pauses for user approval before completing a task.",
        Edges:       workflow.Chain(workflow.Start, approvalNode),
    })
}

고급 기능

Dynamic workflows offer some advanced features designed to handle more complex development scenarios. These capabilities allow for finer control over execution and better integration with existing technical infrastructure.

Execution IDs

The ADK framework generates a deterministic identifier (ID) for child node executions based on the parent ID and a counter. ADK workflows use deterministic IDs for each scheduled node to identify previous results. These IDs are generated based on the order of dynamic node schedules, and are used for checkpointing and to re-run tasks in the correct order in the case of a resumed or re-run workflow.

Custom execution IDs

In some rare cases, you may need to have stable identifiers, such as when processing a reorderable list. In general, you should avoid this due to the impacts to workflow task retries and process resumes. Specifically, these IDs are used to check node states and skip execution if a node was already run. If you provide custom IDs, make sure they are deterministic for workflow re-runs and logically remain the same for the input.

Warning: Custom execution IDs

Avoid creating custom execution IDs. Since execution IDs are used to determine the execution order of nodes, custom execution IDs can cause problems when the system attempts to re-run those nodes in your workflow.

PythonTypeScriptGo

from google.adk import Context
from google.adk.workflow import node
from pydantic import BaseModel
from typing import Any
import asyncio

class Order(BaseModel):
  order_id: str
  cart_items: list[Product]

@node(rerun_on_resume=True)
async def process_all_orders(ctx: Context, node_input: Any):
  orders = await get_orders()

  process_tasks = []
  for order in orders:
    # Use run_id to provide a custom identifier.
    # Custom run_ids must contain at least one non-numeric character
    # to avoid collision with auto-generated sequential numeric IDs.
    task = ctx.run_node(process_order, order, run_id=f"order-{order.order_id}")
    process_tasks.append(task)

  results = await asyncio.gather(*process_tasks)
  return results

By default, auto-generated run IDs are sequential integers starting from "1" (represented as strings). Custom run_id values must contain at least one non-numeric character to avoid collisions with these auto-generated IDs.

후행 옵션으로 ctx.runNode()에 runId를 전달하세요. 이 ID는 자동 생성되는 순차 ID와 충돌하지 않도록 하나 이상의 숫자가 아닌 문자를 포함해야 합니다:

import { node, NodeContext, Workflow } from '@google/adk';

interface Order {
  orderId: string;
  cartItems: string[];
}

/** Stands in for loading orders from a database. */
async function getOrders(): Promise<Order[]> {
  return [
    { orderId: 'a91', cartItems: ['keyboard', 'mouse'] },
    { orderId: 'b02', cartItems: ['monitor'] },
    { orderId: 'c73', cartItems: ['dock', 'cable', 'hub'] },
  ];
}

const processOrder = node(
  (_ctx: NodeContext, order: Order) =>
    `order ${order.orderId}: ${order.cartItems.length} item(s) shipped`,
  { name: 'process_order' },
);

const processAllOrders = node(
  async (ctx: NodeContext) => {
    const orders = await getOrders();

    const processTasks = orders.map((order) =>
      ctx.runNode(processOrder, order, { runId: `order-${order.orderId}` }),
    );

    const results = await Promise.all(processTasks);
    return results.map((result) => result.output).join('\n');
  },
  { name: 'process_all_orders', rerunOnResume: true },
);

export const rootAgent = new Workflow({
  name: 'root_agent',
  edges: [['START', processAllOrders]],
});

In Go, pass workflow.WithRunID("order-x") as a trailing option to workflow.RunNode. The ID must contain at least one non-numeric character to avoid collision with the auto-generated sequential counter IDs:

// newCustomIDWorkflow demonstrates supplying stable custom run IDs via
// workflow.WithRunID — equivalent to Python's:
//
//  task = ctx.run_node(process_order, order, run_id=f"order-{order.order_id}")
//
// Custom run IDs must contain at least one non-numeric character to avoid
// collision with auto-generated sequential integer IDs.
func newCustomIDWorkflow() (agent.Agent, error) {
    processOrderNode := workflow.NewFunctionNode("process_order",
        func(_ agent.Context, orderID string) (string, error) {
            return fmt.Sprintf("processed order %s", orderID), nil
        },
        workflow.NodeConfig{},
    )

    orders := []string{"ord-001", "ord-002", "ord-003"}

    processAllOrders := workflow.NewDynamicNode[any, []string]("process_all_orders",
        func(ctx agent.Context, _ any, _ func(*session.Event) error) ([]string, error) {
            results := make([]string, 0, len(orders))
            for _, orderID := range orders {
                // WithRunID supplies a stable, deterministic identifier for
                // each child invocation. IDs must contain at least one
                // non-numeric character to avoid collision with the
                // auto-generated sequential counter IDs.
                result, err := workflow.RunNode[string](
                    ctx,
                    processOrderNode,
                    orderID,
                    workflow.WithRunID(fmt.Sprintf("order-%s", orderID)),
                )
                if err != nil {
                    return nil, fmt.Errorf("process order %s: %w", orderID, err)
                }
                results = append(results, result)
            }
            return results, nil
        },
        workflow.NodeConfig{},
    )

    return workflowagent.New(workflowagent.Config{
        Name:        "custom_id_workflow",
        Description: "Processes orders with stable per-order execution IDs.",
        Edges:       workflow.Chain(workflow.Start, processAllOrders),
    })
}