Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions core/src/main/java/com/google/adk/agents/Context.java
Original file line number Diff line number Diff line change
Expand Up @@ -323,9 +323,9 @@ public final void routes(List<? extends Route> routes) {
}

/**
* Returns the user's answers to the long-running calls that this workflow node or its descendants
* made in an earlier turn, keyed by interrupt ID. A node that reruns on resume reads these
* answers. A null answer has a null value in the map.
* Returns the user's answers to the requests for input and long-running calls that this workflow
* node or its descendants made in an earlier turn, keyed by interrupt ID. A node that reruns on
* resume reads these answers. A null answer has a null value in the map.
*
* @throws IllegalStateException if this is not the context of a workflow node.
*/
Expand Down
26 changes: 16 additions & 10 deletions core/src/main/java/com/google/adk/workflow/BaseNode.java
Original file line number Diff line number Diff line change
Expand Up @@ -119,21 +119,27 @@ public Optional<Schema> stateSchema() {

/**
* Runs the node and emits its events. Each emission of {@link #runNode} that is an {@link Event}
* passes through, and any other value becomes the output of a new event.
* passes through, a {@link RequestInput} becomes the event that asks for it, and any other value
* becomes the output of a new event.
*/
final Flowable<Event> run(Context context, @Nullable Object nodeInput) {
InvocationContext invocationContext = checkNotNull(context).invocationContext();
// The node runner fills in the author, as only it knows the node's place in the graph.
return Flowable.<Object>defer(() -> runNode(context, nodeInput))
.map(
item ->
item instanceof Event event
? event
: Event.builder()
.id(invocationContext.newUuid())
.timestamp(invocationContext.now().toEpochMilli())
.author("")
.output(item)
.build());
item -> {
if (item instanceof Event event) {
return event;
}
if (item instanceof RequestInput request) {
return request.toEvent(invocationContext);
}
return Event.builder()
.id(invocationContext.newUuid())
.timestamp(invocationContext.now().toEpochMilli())
.author("")
.output(item)
.build();
});
}
}
7 changes: 4 additions & 3 deletions core/src/main/java/com/google/adk/workflow/Node.java
Original file line number Diff line number Diff line change
Expand Up @@ -98,9 +98,10 @@ default boolean requiresAllPredecessors() {
/**
* Runs the node on {@code nodeInput}, the output it receives from its predecessors.
*
* <p>An emitted {@link Event} passes through, and any other emitted value becomes the node's
* output (at most one per run); emit nothing for no output, since RxJava forbids null. The
* wildcard lets an implementation return a {@code Flowable} of its own type.
* <p>An emitted {@link Event} passes through, an emitted {@link RequestInput} pauses the workflow
* until the user answers it, and any other emitted value becomes the node's output (at most one
* per run); emit nothing for no output, since RxJava forbids null. The wildcard lets an
* implementation return a {@code Flowable} of its own type.
*/
@Experimental
Flowable<?> runNode(Context context, @Nullable Object nodeInput);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -196,9 +196,9 @@ public ImmutableSet<String> interruptIds() {
}

/**
* Returns the user's answers to the long-running calls that this node or its descendants raised
* in an earlier turn, keyed by interrupt ID. A null answer has a null value in the map. The map
* is empty unless the node reruns on resume.
* Returns the user's answers to the requests for input and long-running calls that this node or
* its descendants raised in an earlier turn, keyed by interrupt ID. A null answer has a null
* value in the map. The map is empty unless the node reruns on resume.
*/
public Map<String, @Nullable Object> resumeInputs() {
return resumeInputs;
Expand Down
127 changes: 127 additions & 0 deletions core/src/main/java/com/google/adk/workflow/RequestInput.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
/*
* Copyright 2026 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package com.google.adk.workflow;

import com.google.adk.JsonBaseModel;
import com.google.adk.agents.InvocationContext;
import com.google.adk.annotations.Experimental;
import com.google.adk.events.Event;
import com.google.auto.value.AutoValue;
import com.google.common.collect.ImmutableSet;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import com.google.genai.types.Content;
import com.google.genai.types.FunctionCall;
import com.google.genai.types.Part;
import com.google.genai.types.Schema;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import org.jspecify.annotations.Nullable;

/**
* A node's request for user input, which pauses the workflow until the user answers it. A node
* emits it from {@link Node#runNode}, and the runtime records it as a long-running {@value
* #FUNCTION_CALL_NAME} function call whose ID is {@link #interruptId}. The user answers with a
* function response that carries the same ID.
*/
@Experimental
@AutoValue
public abstract class RequestInput {

/** The name of the function call that carries a request for input. */
public static final String FUNCTION_CALL_NAME = "adk_request_input";

static final String INTERRUPT_ID_KEY = "interruptId";
static final String MESSAGE_KEY = "message";
static final String PAYLOAD_KEY = "payload";
static final String RESPONSE_SCHEMA_KEY = "response_schema";

/** Returns the ID of the request, which the answering function response carries. */
public abstract String interruptId();

/** Returns the message to show the user, if any. */
public abstract Optional<String> message();

/** Returns additional data for rendering the request, if any. */
public abstract Optional<Object> payload();

/**
* Returns the schema the answer must match, if any. An answer of the form {@code {"result":
* value}} is unwrapped first, and a string value is read as JSON unless the schema accepts a
* string.
*/
public abstract Optional<Schema> responseSchema();

/** Returns a builder whose interrupt ID is a new random UUID. */
public static Builder builder() {
return new AutoValue_RequestInput.Builder().interruptId(UUID.randomUUID().toString());
}

public abstract Builder toBuilder();

/**
* Returns the long-running function-call event that records this request. It has the model role,
* so history rewriting keeps it as the call that the answer responds to.
*
* @throws IllegalArgumentException if the payload has no JSON form
*/
Event toEvent(InvocationContext invocationContext) {
Map<String, @Nullable Object> args = new LinkedHashMap<>();
args.put(INTERRUPT_ID_KEY, interruptId());
args.put(MESSAGE_KEY, message().orElse(null));
// Converted here so a payload with no JSON form fails the asking node, not the session service.
args.put(PAYLOAD_KEY, payload().map(RequestInput::toJsonValue).orElse(null));
args.put(RESPONSE_SCHEMA_KEY, responseSchema().map(RequestInput::toJsonValue).orElse(null));
FunctionCall call =
FunctionCall.builder().name(FUNCTION_CALL_NAME).id(interruptId()).args(args).build();
return Event.builder()
.id(invocationContext.newUuid())
.timestamp(invocationContext.now().toEpochMilli())
.author("")
.content(
Content.builder()
.role("model")
.parts(Part.builder().functionCall(call).build())
.build())
.longRunningToolIds(ImmutableSet.of(interruptId()))
.build();
}

private static @Nullable Object toJsonValue(Object value) {
return JsonBaseModel.getMapper().convertValue(value, Object.class);
}

/** Builder for {@link RequestInput}. */
@AutoValue.Builder
public abstract static class Builder {

@CanIgnoreReturnValue
public abstract Builder interruptId(String interruptId);

@CanIgnoreReturnValue
public abstract Builder message(@Nullable String message);

@CanIgnoreReturnValue
public abstract Builder payload(@Nullable Object payload);

@CanIgnoreReturnValue
public abstract Builder responseSchema(@Nullable Schema responseSchema);

public abstract RequestInput build();
}
}
Loading
Loading