wenjin272 commented on code in PR #1268:
URL: https://github.com/apache/flink-agents/pull/1268#discussion_r4293163914
##########
docs/content/docs/development/multi-agent/subagents.md:
##########
@@ -307,30 +443,854 @@ parent.addResource(
+ "\"required\":[\"diff\"]}")
.build());
```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The framework does not check the schema against the prompt. A field the prompt
does
+not use is ignored, and a placeholder no field fills is left in the prompt as
literal
+text. So keep the two in step when you change either.
+
+### External: write a setup class
+
+An external sub-agent has no plan of its own, and it is not compiled into the
+parent's. The work happens in your own code or in a remote agent service. The
parent
+only submits to it and waits for the outcome. The framework cannot perform
that work
+for you, because it does not know your protocol. So you write the part that
performs
+one invocation, and the framework supplies everything around it. It runs your
code
+under durable execution, and after a failover it replays the recorded result
instead
+of calling your code again.
+
+What you write is a setup class. Its only job is to carry out one invocation,
so it
+declares no actions and no events of its own. It extends a framework base
class and
+implements the methods that class declares.
+
+Registration names the setup class in a descriptor, and both languages
construct the
+setup from the descriptor at runtime. Registering a setup instance instead is
+rejected.
+
+The invocation itself is carried out in one of two modes, and which one you
write
+follows from your remote interface. When one request returns the result, write
a
+*deferred* setup. You own the whole interaction, and the framework only holds
it back
+until the caller asks for the result. When the remote instead hands back a run
that
+you later query for status and result, write an *async (pub/sub)* setup. The
base
+class owns that composition of publishing, polling, and fetching, and you
supply the
+transport calls it is built from.
+
+#### Deferred execution
+
+A deferred setup sends nothing when the caller submits. It only registers the
+invocation. The request goes out when the caller first awaits the result. So a
call
+that is cancelled instead never issues its request.
+
+What you write is the interaction itself, in the `prepare` method of your
setup class.
+`prepare` runs once per invocation and returns the durable call that carries
it. The
+example is one blocking request, and whatever else your call does is yours to
decide,
+because the framework does not know your protocol.
+The framework assigns `session_id` and `call_id` before it calls `prepare`,
and they
+are the only values that a replay reproduces. The durable id you build from
them is
+the key that recovery matches a recorded result against. When the key matches,
your
+call is not run again. So build the key from those two values alone. Anything
else in
+it, a timestamp for instance, makes a replay look up a different key and run
the work
+a second time. Passing the same values to your own remote lets it recognize a
+re-issued request as the same one.
+
+The reconciler is optional. Omitting it leaves one crash window uncovered, and
+[Durability and failover](#durability-and-failover) describes that window.
Supply one
+when the work cannot be repeated.
+
+{{< tabs "Supply a Reconciler" >}}
+
+{{< tab "Python" >}}
```python
-reviewer = ReActAgent(
- chat_model=chat_model_descriptor,
- prompt=Prompt.from_text(
- "Review the following code diff and report any problems:\n\n{diff}"
+def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ return SubagentResult.ok(post_review(self.endpoint, prompt, key))
+
+ # Consulted only during recovery, when this call is revisited and no
result was
+ # recorded for it. Return the recovered outcome, or raise the recovered
failure.
+ # It takes no arguments.
+ def reconcile() -> SubagentResult:
+ # fetch_review_by_key is your own lookup; not part of Flink Agents.
+ found = fetch_review_by_key(self.endpoint, key)
+ if found is None:
+ raise RuntimeError("the review service has no record of this call")
+ return SubagentResult.ok(found)
+
+ return (key, call, reconcile)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+// Inside the DurableCallable that prepare returns, alongside getId() and
call().
+
+// Consulted only during recovery, when this call is revisited and no result
was
+// recorded for it. Return the recovered outcome, or throw the recovered
failure.
+@Override
+public Callable<SubagentResult> reconciler() {
+ return () -> {
+ // fetchReviewByKey is your own lookup; not part of Flink Agents.
+ String found = fetchReviewByKey(endpoint, key);
+ if (found == null) {
+ throw new IllegalStateException("the review service has no record
of this call");
+ }
+ return SubagentResult.ok(found);
+ };
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+{{< tabs "Deferred Sub-agent Setup" >}}
+
+{{< tab "Python" >}}
+```python
+from typing import Any
+
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.deferred_subagent import (
+ DeferredSubagentSetup,
+ PreparedTriple,
+)
+
+
+def post_review(endpoint: str, prompt: Any, key: str) -> str:
+ """Your own call to the review service; not part of Flink Agents.
+
+ Keying the work lets the service treat a re-issued request as the same one.
+ """
+ ...
+
+
+class RemoteReviewSetup(DeferredSubagentSetup):
+ endpoint: str = ""
+
+ def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+ ) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ try:
+ return SubagentResult.ok(post_review(self.endpoint, prompt,
key))
+ except Exception as e:
+ # An expected failure is a value; only a system fault should
escape.
+ return SubagentResult.error(e)
+
+ # The first value is the key recovery matches a recorded result
against, so
+ # it is built from the two ids alone. The third is the reconciler.
+ return (key, call, None)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.runtime.subagent.BaseDeferredSubagentSetup;
+
+/**
+ * Your own call to the review service; not part of Flink Agents. Keying the
work lets
+ * the service treat a re-issued request as the same one.
+ */
+static String postReview(String endpoint, Object prompt, String key) {
+ ...
+}
+
+public class RemoteReviewSetup extends BaseDeferredSubagentSetup {
+
+ private final String endpoint;
+
+ public RemoteReviewSetup(ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ this.endpoint = descriptor.getArgument("endpoint");
+ }
+
+ @Override
+ protected DurableCallable<SubagentResult> prepare(
+ RunnerContext ctx, Object prompt, String sessionId, String callId)
{
+ // Built from the two ids alone, so a replay reproduces it.
+ String key = sessionId + "#" + callId;
+ return new DurableCallable<SubagentResult>() {
+ @Override
+ public String getId() {
+ // The key recovery matches a recorded result against.
+ return key;
+ }
+
+ @Override
+ public Class<SubagentResult> getResultClass() {
+ return SubagentResult.class;
+ }
+
+ @Override
+ public SubagentResult call() {
+ try {
+ return SubagentResult.ok(postReview(endpoint, prompt,
key));
+ } catch (Exception e) {
+ // An expected failure is a value; only a system fault
should escape.
+ return SubagentResult.error(e);
+ }
+ }
+ };
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+Register the setup under the name invocations will use. Every descriptor
argument
+beyond the built-in ones is configuration for the setup to read back. In
Python it
+reads them as model fields, and in Java through the public
+`(ResourceDescriptor, ResourceContext)` constructor. The `endpoint` above is
such
+configuration, not a built-in transport.
+
+{{< tabs "Register an External Sub-agent" >}}
+
+{{< tab "Python" >}}
+```python
+parent.add_resource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteReviewSetup",
+ endpoint="https://example.com/review";,
+ description="Review code using the remote service",
+ input_schema=(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ ),
),
)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+parent.addResource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteReviewSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/review";)
+ .addInitialArgument("description", "Review code using the remote
service")
+ .addInitialArgument(
+ "input_schema",
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .build());
+```
+{{< /tab >}}
+
+{{< tab "YAML" >}}
+```yaml
+agents:
+ - name: orchestrator
+ subagents:
+ - name: remote_reviewer
+ clazz: my_app.agents.RemoteReviewSetup
+ endpoint: https://example.com/review
+ description: Review code using the remote service
+ input_schema:
'{"type":"object","properties":{"diff":{"type":"string"}},"required":["diff"]}'
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The YAML form of that registration is the `subagents` list of an agent entry,
as in
+the YAML tab above. That list can also sit at the shared document level,
outside
+`agents:`, so that several agents in one file reuse it. A chat model setup
declared in
+the same file also takes a `subagents` argument, but it is a list of
registered names
+rather than a list of entries. See
+[delegate from a chat model](#delegate-from-a-chat-model).
+
+Two constraints apply to the entry:
+
+- No class alias is defined for `AGENT` resources, so `clazz` is always a
+ fully-qualified class path.
+- The entry's language must match the loader. A Python YAML file names a Python
+ class and may not set `type: java`. A Java YAML file must set `type: java`.
+
+#### Async (pub/sub) execution
+
+An async sub-agent talks to a service that runs the task for you. The service
accepts
+the work and answers at once, and you query it later for the run's status and
+result.
+
+An async setup publishes the run when the caller submits, through one durable
POST.
+Waiting for the result then polls the run's status until it reaches a terminal
state,
+and fetches it.
+
+That is the difference from the deferred mode, which sends nothing when the
caller
+submits. The mode's name is the framework's own. The POST is the publish, and
the
+polling is the subscribe.
+
+The submit-poll-fetch composition is the same for every such remote, so the
base
+class owns it, along with the durable execution and the crash-window recovery.
You
+supply the three transport calls it is built from:
+
+- `callSubmitRequest` / `call_submit_request` starts the run.
+- `callQueryStatus` / `call_query_status` returns a `RunStatus`, one of
+ `notStarted()`, `running()`, `completed()`, or `failed(error)`. It must be a
+ read-only probe with no side effects. Report an expected failure as a
`FAILED`
+ status.
+- `callFetchResult` / `call_fetch_result` returns the `SubagentResult`. It
must be
+ an idempotent read, because a replay may call it again. The status snapshot
never
+ carries the payload, and a failed run is reported without fetching at all.
+
+{{< tabs "Async Sub-agent Setup" >}}
+
+{{< tab "Python" >}}
+```python
+from typing import Any
+
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.async_subagent import BaseAsyncSubagentSetup,
RunStatus
+
+# client(...) and task_id stand for your own transport client and run-id
+# scheme; neither is part of Flink Agents.
+
+
+class RemoteRunSetup(BaseAsyncSubagentSetup):
+ endpoint: str = ""
+
+ def call_submit_request(self, session_id: str, call_id: str, prompt: Any)
-> None:
+ client(self.endpoint).submit(prompt, task_id=f"{session_id}#{call_id}")
+
+ def call_query_status(self, session_id: str, call_id: str) -> RunStatus:
+ probe = client(self.endpoint).status(task_id=f"{session_id}#{call_id}")
+ if probe is None:
+ return RunStatus.not_started()
+ if probe.succeeded:
+ return RunStatus.completed()
+ if probe.failed:
+ return RunStatus.failed(probe.error)
+ return RunStatus.running()
+
+ def call_fetch_result(self, session_id: str, call_id: str) ->
SubagentResult:
+ return SubagentResult.ok(
+ client(self.endpoint).fetch(task_id=f"{session_id}#{call_id}")
+ )
+
+ # Optional, and a no-op by default, so a cancelled call leaves the remote
run
+ # going. A replay may propagate the same cancellation again, so your
remote has
+ # to accept it twice.
+ def call_cancel_request(self, session_id: str, call_id: str) -> None:
+ client(self.endpoint).cancel(task_id=f"{session_id}#{call_id}")
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.runtime.subagent.BaseAsyncSubagentSetup;
+
+// client() and taskId(...) stand for your own transport client and run-id
+// scheme; neither is part of Flink Agents.
+public class RemoteRunSetup extends BaseAsyncSubagentSetup {
+
+ private final String endpoint;
+
+ public RemoteRunSetup(ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ this.endpoint = descriptor.getArgument("endpoint");
+ }
+ @Override
+ protected void callSubmitRequest(String sessionId, String callId, Object
prompt)
+ throws Exception {
+ client().submit(String.valueOf(prompt), taskId(sessionId, callId));
+ }
+
+ @Override
+ protected RunStatus callQueryStatus(String sessionId, String callId) {
+ // TaskStatus is your own transport type, not a framework type.
+ TaskStatus probe = client().status(taskId(sessionId, callId));
+ if (probe == null) {
+ return RunStatus.notStarted();
+ }
+ if (probe.succeeded()) {
+ return RunStatus.completed();
+ }
+ if (probe.failed()) {
+ return RunStatus.failed(probe.error());
+ }
+ return RunStatus.running();
+ }
+
+ @Override
+ protected SubagentResult callFetchResult(String sessionId, String callId)
throws Exception {
+ return client().fetchResult(taskId(sessionId, callId));
+ }
+
+ // Optional, and a no-op by default, so a cancelled call leaves the remote
run
+ // going. A replay may propagate the same cancellation again, so your
remote has
+ // to accept it twice.
+ @Override
+ protected void callCancelRequest(String sessionId, String callId) {
+ client().cancel(taskId(sessionId, callId));
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+Register the setup like any external sub-agent. The async form adds one
descriptor
+argument, `status_poll_interval_millis`, and it defaults to 500.
+
+{{< tabs "Register an Async Sub-agent" >}}
+
+{{< tab "Python" >}}
+```python
parent.add_resource(
- "reviewer",
+ "remote_runner",
ResourceType.AGENT,
- InternalSubagentResourceDescriptorBuilder()
- .agent(reviewer)
- .description("Reviews a code diff and reports problems.")
- # "diff" matches the {diff} placeholder of the child's prompt.
- .input_schema(
- '{"type":"object","properties":{"diff":{"type":"string"}},'
- '"required":["diff"]}'
- )
- .build(),
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteRunSetup",
+ endpoint="https://example.com/runs";,
+ description="Run a review task on the remote service",
+ input_schema=(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ ),
+ status_poll_interval_millis=500,
+ ),
)
```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+parent.addResource(
+ "remote_runner",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteRunSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/runs";)
+ .addInitialArgument("description", "Run a review task on the remote
service")
+ .addInitialArgument(
+ "input_schema",
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .addInitialArgument("status_poll_interval_millis", 500L)
+ .build());
+```
+{{< /tab >}}
+
+{{< tab "YAML" >}}
+```yaml
+agents:
+ - name: orchestrator
+ subagents:
+ - name: remote_runner
+ clazz: my_app.agents.RemoteRunSetup
+ endpoint: https://example.com/runs
+ description: Run a review task on the remote service
+ input_schema:
'{"type":"object","properties":{"diff":{"type":"string"}},"required":["diff"]}'
+ status_poll_interval_millis: 500
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The base class also recovers the crash window between "POST sent" and "POST
durably
+recorded". On recovery it probes the run's status first, and it re-posts only
when the
+probe reports that the run never started. Override `reconcileSubmitRequest` /
+`reconcile_submit_request` only if your service needs a different repair.
+
+## Invoke a sub-agent
+
+A sub-agent is invoked in one of two ways, model-driven or caller-driven, and
the two
+reuse different existing abstractions.
+
+Model-driven delegation rides the tool abstraction. The sub-agent is offered
to the
+model as a tool, so your action sends a `ChatRequestEvent` and the model
answers with
+a call to the sub-agent. The result then returns to the model as the tool
+observation, until a final `ChatResponseEvent` reaches your action. This is
the same
+cycle any tool call makes.
+
+Caller-driven delegation cannot ride that cycle, because a tool's result goes
back
+into the chat loop rather than to the action that asked for it. An action that
needs
+the delegated outcome therefore calls the sub-agent itself, with no model in
the loop.
+
+Both ways end in the same call. Your action, or the framework on the model's
behalf,
+runs `submit(ctx, prompt)` on the registered name. That call returns a
+`SubagentFuture`, which is the handle for that one call. Awaiting the handle
yields
+the call's `SubagentResult`.
+
+### Delegate from a chat model
+
+List the registered name under `subagents` of the parent's chat model:
+
+{{< tabs "Declare Sub-agents on a Chat Model" >}}
+
+{{< tab "Python" >}}
+```python
+ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP,
+ connection="ollama_connection",
+ model="qwen3:8b",
+ subagents=["reviewer"],
+)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "ollamaConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .addInitialArgument("subagents", List.of("reviewer"))
+ .build();
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The model sees `reviewer` as a tool, described by the registration's
description and
+input schema. A schema is required for model-selected calls. A missing
description
+falls back to a generic delegation description.
+
+The arguments the model sends become the sub-agent's input, and its outputs
return to
+the model as the tool observation. The model can then keep reasoning over
them. For
+an internal sub-agent, the observation is the JSON encoding of the full output
list
+described in [the result](#the-result). That holds even when the child emits a
single
+output, so the shape does not depend on how many outputs the child emits.
+
+An external setup class can state the shape of that observation. It declares
+`result_type()` in Python or `getResultType()` in Java, and the default
reports the
+result as it arrived. Declare one to narrow a wider payload, or to convert a
payload
+that is not already JSON-compatible. An internal sub-agent has no such
declaration, so
+its observation is always the JSON encoding of its output list. This shapes the
+observation only. A caller that awaits the handle gets the result the sub-agent
+produced, as [the result](#the-result) describes.
+
+The external sub-agents `remote_reviewer` and `remote_runner` are offered the
same
+way. Only their registration differs.
+
+### Call from an action
+
+An action can submit input to a registered sub-agent directly and await its
+outcome. This path needs no description and no input declaration, so it also
works
+for a child registered as-is.
+
+{{< tabs "Call a Sub-agent From an Action" >}}
+
+{{< tab "Python" >}}
+```python
+class DelegatingAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ async def delegate(event: Event, ctx: RunnerContext) -> None:
+ reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+ future = await reviewer.submit(ctx, "review the diff")
+ result = await future
+ if result.success:
+ ctx.send_event(OutputEvent(output=result.result[0]))
+ else:
+ ctx.send_event(OutputEvent(output=f"failed:
{result.error_message}"))
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+public class DelegatingAgent extends Agent {
+
+ @Action(EventType.InputEvent)
+ public static void delegate(Event event, RunnerContext ctx) throws
Exception {
+ SubagentSetup reviewer =
+ (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+ SubagentResult result = reviewer.submit(ctx, "review the
diff").await();
+ if (result.isSuccess()) {
+ String answer = String.valueOf(((List<?>)
result.getResult()).get(0));
+ ctx.sendEvent(new OutputEvent(answer));
+ } else {
+ ctx.sendEvent(new OutputEvent("failed: " +
result.getErrorMessage()));
+ }
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+What you pass to `submit` is what the sub-agent receives. An internal child
gets it
+as its `InputEvent` payload, and a model-driven call passes the arguments map
+instead. What comes back is a `SubagentResult`. For the internal sub-agent
used here,
+its `result` is the list of `OutputEvent` payloads the child emitted. That is
why the
+example reads `result.result[0]`. An external sub-agent returns whatever its
setup
+produces.
+
+In Python `submit` is a coroutine, so the call is awaited in two phases and the
+action must be `async`. In Java `submit` returns the handle directly and
`await()`
+yields the result. The same two calls drive `remote_reviewer` and
`remote_runner`
+unchanged. Only the registration differs.
+
+## Work with open handles
+
+A `SubagentFuture` is live from `submit` until it is awaited or cancelled.
+
+It works like the `AsyncFuture` that
+[async execution]({{< ref
"docs/development/create-agent/actions#async-execution" >}})
+returns. You get it from one call, and you await it to get the outcome. It is
its own
+type rather than an `AsyncFuture`, because it always resolves to a
`SubagentResult`,
+it carries the call's `(session_id, call_id)`, and it owns its durable
execution
+inside the setup.
+
+### Fan out to several sub-agents
+
+`combine` batches handles into a `SubagentFutures`, and it works the way
`gather`
+works for ordinary async futures. `gather` does not accept a `SubagentFuture`,
so
+handles batch this way instead. Every handle in the batch is prepared before
any is
+executed, so deferred requests all go out up front and remote runs overlap.
Results
Review Comment:
Could we avoid promising concurrent execution for deferred calls here? In
both Java and Python, `SubagentFutureGroup` prepares all handles, then executes
and waits for each durable call one by one (the implementation explicitly
leaves batching to TODO #926). In the `RemoteReviewSetup` example above, the
POST is inside the callable, so the second request starts only after the first
completes. A minimal reproduction produces `POST a -> complete a -> POST b ->
complete b`.
Please describe `combine` as grouping handles and preserving result order,
without implying that it makes deferred requests overlap. The similar overlap
claim at the end of Restrictions should also distinguish async setups, which
start remote work during `submit`, from deferred/internal calls.
##########
docs/content/docs/development/multi-agent/subagents.md:
##########
@@ -307,30 +443,854 @@ parent.addResource(
+ "\"required\":[\"diff\"]}")
.build());
```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The framework does not check the schema against the prompt. A field the prompt
does
+not use is ignored, and a placeholder no field fills is left in the prompt as
literal
+text. So keep the two in step when you change either.
+
+### External: write a setup class
+
+An external sub-agent has no plan of its own, and it is not compiled into the
+parent's. The work happens in your own code or in a remote agent service. The
parent
+only submits to it and waits for the outcome. The framework cannot perform
that work
+for you, because it does not know your protocol. So you write the part that
performs
+one invocation, and the framework supplies everything around it. It runs your
code
+under durable execution, and after a failover it replays the recorded result
instead
+of calling your code again.
+
+What you write is a setup class. Its only job is to carry out one invocation,
so it
+declares no actions and no events of its own. It extends a framework base
class and
+implements the methods that class declares.
+
+Registration names the setup class in a descriptor, and both languages
construct the
+setup from the descriptor at runtime. Registering a setup instance instead is
+rejected.
+
+The invocation itself is carried out in one of two modes, and which one you
write
+follows from your remote interface. When one request returns the result, write
a
+*deferred* setup. You own the whole interaction, and the framework only holds
it back
+until the caller asks for the result. When the remote instead hands back a run
that
+you later query for status and result, write an *async (pub/sub)* setup. The
base
+class owns that composition of publishing, polling, and fetching, and you
supply the
+transport calls it is built from.
+
+#### Deferred execution
+
+A deferred setup sends nothing when the caller submits. It only registers the
+invocation. The request goes out when the caller first awaits the result. So a
call
+that is cancelled instead never issues its request.
+
+What you write is the interaction itself, in the `prepare` method of your
setup class.
+`prepare` runs once per invocation and returns the durable call that carries
it. The
+example is one blocking request, and whatever else your call does is yours to
decide,
+because the framework does not know your protocol.
+The framework assigns `session_id` and `call_id` before it calls `prepare`,
and they
+are the only values that a replay reproduces. The durable id you build from
them is
+the key that recovery matches a recorded result against. When the key matches,
your
+call is not run again. So build the key from those two values alone. Anything
else in
+it, a timestamp for instance, makes a replay look up a different key and run
the work
+a second time. Passing the same values to your own remote lets it recognize a
+re-issued request as the same one.
+
+The reconciler is optional. Omitting it leaves one crash window uncovered, and
+[Durability and failover](#durability-and-failover) describes that window.
Supply one
+when the work cannot be repeated.
+
+{{< tabs "Supply a Reconciler" >}}
+
+{{< tab "Python" >}}
```python
-reviewer = ReActAgent(
- chat_model=chat_model_descriptor,
- prompt=Prompt.from_text(
- "Review the following code diff and report any problems:\n\n{diff}"
+def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ return SubagentResult.ok(post_review(self.endpoint, prompt, key))
+
+ # Consulted only during recovery, when this call is revisited and no
result was
+ # recorded for it. Return the recovered outcome, or raise the recovered
failure.
+ # It takes no arguments.
+ def reconcile() -> SubagentResult:
+ # fetch_review_by_key is your own lookup; not part of Flink Agents.
+ found = fetch_review_by_key(self.endpoint, key)
+ if found is None:
+ raise RuntimeError("the review service has no record of this call")
+ return SubagentResult.ok(found)
+
+ return (key, call, reconcile)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+// Inside the DurableCallable that prepare returns, alongside getId() and
call().
+
+// Consulted only during recovery, when this call is revisited and no result
was
+// recorded for it. Return the recovered outcome, or throw the recovered
failure.
+@Override
+public Callable<SubagentResult> reconciler() {
+ return () -> {
+ // fetchReviewByKey is your own lookup; not part of Flink Agents.
+ String found = fetchReviewByKey(endpoint, key);
+ if (found == null) {
+ throw new IllegalStateException("the review service has no record
of this call");
+ }
+ return SubagentResult.ok(found);
+ };
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+{{< tabs "Deferred Sub-agent Setup" >}}
+
+{{< tab "Python" >}}
+```python
+from typing import Any
+
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.deferred_subagent import (
+ DeferredSubagentSetup,
+ PreparedTriple,
+)
+
+
+def post_review(endpoint: str, prompt: Any, key: str) -> str:
+ """Your own call to the review service; not part of Flink Agents.
+
+ Keying the work lets the service treat a re-issued request as the same one.
+ """
+ ...
+
+
+class RemoteReviewSetup(DeferredSubagentSetup):
+ endpoint: str = ""
+
+ def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+ ) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ try:
+ return SubagentResult.ok(post_review(self.endpoint, prompt,
key))
+ except Exception as e:
+ # An expected failure is a value; only a system fault should
escape.
+ return SubagentResult.error(e)
+
+ # The first value is the key recovery matches a recorded result
against, so
+ # it is built from the two ids alone. The third is the reconciler.
+ return (key, call, None)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.runtime.subagent.BaseDeferredSubagentSetup;
+
+/**
+ * Your own call to the review service; not part of Flink Agents. Keying the
work lets
+ * the service treat a re-issued request as the same one.
+ */
+static String postReview(String endpoint, Object prompt, String key) {
+ ...
+}
+
+public class RemoteReviewSetup extends BaseDeferredSubagentSetup {
+
+ private final String endpoint;
+
+ public RemoteReviewSetup(ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ this.endpoint = descriptor.getArgument("endpoint");
+ }
+
+ @Override
+ protected DurableCallable<SubagentResult> prepare(
+ RunnerContext ctx, Object prompt, String sessionId, String callId)
{
+ // Built from the two ids alone, so a replay reproduces it.
+ String key = sessionId + "#" + callId;
+ return new DurableCallable<SubagentResult>() {
+ @Override
+ public String getId() {
+ // The key recovery matches a recorded result against.
+ return key;
+ }
+
+ @Override
+ public Class<SubagentResult> getResultClass() {
+ return SubagentResult.class;
+ }
+
+ @Override
+ public SubagentResult call() {
+ try {
+ return SubagentResult.ok(postReview(endpoint, prompt,
key));
+ } catch (Exception e) {
+ // An expected failure is a value; only a system fault
should escape.
+ return SubagentResult.error(e);
+ }
+ }
+ };
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+Register the setup under the name invocations will use. Every descriptor
argument
+beyond the built-in ones is configuration for the setup to read back. In
Python it
+reads them as model fields, and in Java through the public
+`(ResourceDescriptor, ResourceContext)` constructor. The `endpoint` above is
such
+configuration, not a built-in transport.
+
+{{< tabs "Register an External Sub-agent" >}}
+
+{{< tab "Python" >}}
+```python
+parent.add_resource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteReviewSetup",
+ endpoint="https://example.com/review";,
+ description="Review code using the remote service",
+ input_schema=(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ ),
),
)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+parent.addResource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteReviewSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/review";)
+ .addInitialArgument("description", "Review code using the remote
service")
+ .addInitialArgument(
+ "input_schema",
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .build());
+```
+{{< /tab >}}
+
+{{< tab "YAML" >}}
+```yaml
+agents:
+ - name: orchestrator
+ subagents:
+ - name: remote_reviewer
+ clazz: my_app.agents.RemoteReviewSetup
+ endpoint: https://example.com/review
+ description: Review code using the remote service
+ input_schema:
'{"type":"object","properties":{"diff":{"type":"string"}},"required":["diff"]}'
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The YAML form of that registration is the `subagents` list of an agent entry,
as in
+the YAML tab above. That list can also sit at the shared document level,
outside
+`agents:`, so that several agents in one file reuse it. A chat model setup
declared in
+the same file also takes a `subagents` argument, but it is a list of
registered names
+rather than a list of entries. See
+[delegate from a chat model](#delegate-from-a-chat-model).
+
+Two constraints apply to the entry:
+
+- No class alias is defined for `AGENT` resources, so `clazz` is always a
+ fully-qualified class path.
+- The entry's language must match the loader. A Python YAML file names a Python
+ class and may not set `type: java`. A Java YAML file must set `type: java`.
+
+#### Async (pub/sub) execution
+
+An async sub-agent talks to a service that runs the task for you. The service
accepts
+the work and answers at once, and you query it later for the run's status and
+result.
+
+An async setup publishes the run when the caller submits, through one durable
POST.
+Waiting for the result then polls the run's status until it reaches a terminal
state,
+and fetches it.
+
+That is the difference from the deferred mode, which sends nothing when the
caller
Review Comment:
Could we focus these explanations more on how users work with sub-agents and
less on the framework’s design rationale? For example:
- Instead of explaining “the POST is the publish, and the polling is the
subscribe,” describe the observable behavior: submitting starts a remote run,
and awaiting polls its status and retrieves the result.
- In “Work with open handles,” focus on awaiting, combining, cancelling, and
accessing session/call IDs rather than why `SubagentFuture` is a separate type
from `AsyncFuture` or where it owns durable execution.
- In “Invoke a sub-agent,” replace “rides the tool abstraction” with the
practical distinction: model-driven results return to the model, while
caller-driven results return to the action.
The `prepare`, durable ID, and reconciler contracts should stay, since users
implementing external setups need those details.
##########
docs/content/docs/development/multi-agent/subagents.md:
##########
@@ -24,30 +24,170 @@ specific language governing permissions and limitations
under the License.
-->
-A sub-agent handles a task delegated by a parent agent. An internal sub-agent
runs
-an agent workflow in the same job. An external sub-agent calls an agent service
-through a `SubagentSetup` implementation.
+## Overview
-## Declare an internal sub-agent
+A sub-agent is an agent that another agent delegates a task to. The parent
hands it
+input, the sub-agent does the work, and the parent receives the outcome as a
+`SubagentResult`. Use one when the delegated work is multi-step or
long-running and
+should run isolated from the parent.
-Use `InternalSubagentResourceDescriptorBuilder` to attach a description and
input
-schema to a child agent. The builder returns a normal resource descriptor for
`addResource`
-or `add_resource`.
+The work runs either inside the job or outside it. An *internal* sub-agent is a
+child `Agent`, including a `ReActAgent`, compiled into the parent's plan. The
+child's own actions and events run in the same job. An *external* sub-agent
has no
+plan of its own. The work happens in your own code or in a remote agent
service,
+outside the parent's event loop. It takes one of two forms, *deferred* or
*async
+(pub/sub)*.
+Either way, the call is initiated by the model or by you. In *model-driven
+delegation* you list the sub-agent on a chat model setup and the model chooses
when
+to delegate. In *caller-driven delegation* one of your actions submits the task
+itself.
+
+Every sub-agent call runs through
+[durable execution]({{< ref
"docs/development/create-agent/actions#durable-execution" >}}).
+There is no durability toggle and no sub-agent timeout option.
+
+## Declare a sub-agent
+
+A sub-agent enters the parent's plan as a resource registered under
+`ResourceType.AGENT`. The name you register it under is the name that
invocations and
+metrics use. What you register depends on the form. An internal sub-agent
registers
+a child `Agent`. An external sub-agent registers a descriptor that names your
setup
+class, and a YAML file can declare it.
+
+### Internal: compile a child agent
+
+Nothing in a child's own code marks it as a sub-agent. The child is an ordinary
+agent with its own resources, actions, and event loop, and it runs standalone
+exactly as written. A parent registers it as an `AGENT` resource, and that
+registration is the whole of the difference. It gives the child a name in the
+parent's plan. It can also give the child an input schema, so a chat model can
call
+it. And it supplies the caller that submits input and awaits the child's
outputs.
+
+Any `Agent` can serve as the child. The two common kinds are a workflow agent
and a
+`ReActAgent`. Either way, the child's plan is compiled into the parent's when
the
+parent's plan is built.
+
+#### A workflow agent as the child
+
+The example child owns a chat model and answers each review request with its
Review Comment:
Could we streamline the guide and make it easier to scan? A few concrete
suggestions:
- **Reuse existing examples.** “A workflow agent as the child” repeats the
connection/model setup and request/response actions already shown in the Chat
with Models guide. Could we link to that example and keep only the child
registration and input adaptation here—for example, reading `task["code"]` for
model-driven calls versus the submitted string for caller-driven calls?
- **Consolidate repeated explanations.** The introduction to internal
sub-agents already explains that a child is an ordinary Agent made into a
sub-agent through registration; the workflow example immediately repeats this.
Likewise, “always durable, no sub-agent timeout option” appears in both
Overview and Durability and failover. A brief introduction with a link to the
detailed section should be enough.
- **Use tables for comparisons.** Before the external setup examples, a
deferred-vs-async table could show when work starts (first await vs. submit),
what users implement (`prepare` vs. submit/status/fetch hooks), and when to
choose each (a request returning the result vs. a remote run queried later).
This would help readers choose an approach without extracting these differences
from several long paragraphs.
##########
docs/content/docs/development/multi-agent/subagents.md:
##########
@@ -67,156 +207,147 @@ parent.add_resource(
.build(),
)
```
+{{< /tab >}}
-`ReviewAgent` is your child workflow; the Java `ReviewRequest` is a data class
-with fields describing its input. Alternatively, pass a JSON schema string to
-`inputSchema(...)` in Java or `input_schema(...)` in Python. Choose either an
-input type or an explicit schema. The schema describes the submitted arguments,
-which the child receives as its input event payload; it does not convert that
-payload into an instance of the input type.
-
-When a chat model is configured with `subagents=["reviewer"]`, the description
-helps it choose when to delegate, and the schema tells it which arguments to
-send. A schema is required for model-selected calls. A missing description uses
-a generic delegation description.
-
-For calls made explicitly by an action, the existing shorthand remains
available:
-
+{{< tab "Java" >}}
```java
-parent.addResource("reviewer", ResourceType.AGENT, new ReviewAgent());
-```
+import org.apache.flink.agents.api.resource.ResourceType;
+import
org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
-```python
-parent.add_resource("reviewer", ResourceType.AGENT, ReviewAgent())
+parent.addResource(
+ "reviewer",
+ ResourceType.AGENT,
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new ReviewAgent())
+ .description("Review code for correctness and potential defects")
+ .inputType(ReviewRequest.class)
+ .build());
```
+{{< /tab >}}
-This shorthand does not declare an input schema. Use the builder when exposing
-the child to a model. Recursive registration, such as a child registering its
-parent, is rejected when the plan is built.
+{{< /tabs >}}
-Register the resources a child uses, such as its chat model connection, on the
-child itself. In Python, a child resolves resources against its own plan only,
-so a resource registered solely on the parent is not visible to the child.
+`ReviewRequest` describes the arguments the child accepts, and the Java form
is a bean.
+Where no type fits, a JSON schema string declares the same arguments:
-## Declare an external sub-agent
+{{< tabs "Declare the Input as a Schema" >}}
-Implement `SubagentSetup` (or the appropriate runtime base class) and register
a
-resource descriptor naming that implementation. Do not register a setup
instance.
+{{< tab "Python" >}}
+```python
+parent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(ReviewAgent())
+ .description("Review code for correctness and potential defects")
+ .input_schema(
+ '{"type":"object","properties":{"code":{"type":"string"}},'
+ '"required":["code"]}'
+ )
+ .build(),
+)
+```
+{{< /tab >}}
+{{< tab "Java" >}}
```java
parent.addResource(
- "remote_reviewer",
+ "reviewer",
ResourceType.AGENT,
- ResourceDescriptor.Builder.newBuilder(RemoteReviewSetup.class.getName())
- .addInitialArgument("endpoint", "https://example.com/review";)
- .addInitialArgument("description", "Review code using the remote
service")
- .addInitialArgument("input_schema", inputSchema)
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new ReviewAgent())
+ .description("Review code for correctness and potential defects")
+ .inputSchema(
+
"{\"type\":\"object\",\"properties\":{\"code\":{\"type\":\"string\"}},"
+ + "\"required\":[\"code\"]}")
.build());
```
+{{< /tab >}}
-The Java setup exposes a public `(ResourceDescriptor, ResourceContext)`
-constructor and implements the sub-agent invocation contract. `endpoint` above
-is configuration read by the example implementation, not a built-in transport.
+{{< /tabs >}}
-```python
-parent.add_resource(
- "remote_reviewer",
- ResourceType.AGENT,
- ResourceDescriptor(
- clazz="my_app.agents.RemoteReviewSetup",
- endpoint="https://example.com/review";,
- description="Review code using the remote service",
- input_schema=input_schema,
- ),
-)
-```
+Declare either an input type or an input schema, not both. A declaration that
carries
+both is rejected when the descriptor is built. In Python the type must be a
pydantic
+model, and a type that is not is rejected there too.
-The Python setup declares these configuration fields and implements the
-invocation contract. Both languages construct the setup at runtime from the
-descriptor. Existing code that passes a setup instance to `addResource` or
-`add_resource` must instead pass a descriptor with the same configuration.
+An input type is a shorthand for a schema. The framework derives the schema
from the
+type and registers that, so use a type when you already have one and a schema
when
+you do not. Nothing is ever converted into an instance of the type. The
sub-agent
+receives the arguments as a map keyed by the schema's field names.
-## Delegate from a chat model
+For calls an action makes itself, the child registers as-is:
-List the child's registered name under `subagents` of the parent's chat model:
+{{< tabs "Register the Child Agent Directly" >}}
-```java
-ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
- .addInitialArgument("connection", "ollamaConnection")
- .addInitialArgument("model", "qwen3:8b")
- .addInitialArgument("subagents", List.of("reviewer"))
- .build();
+{{< tab "Python" >}}
+```python
+parent.add_resource("reviewer", ResourceType.AGENT, ReviewAgent())
```
+{{< /tab >}}
-```python
-ResourceDescriptor(
- clazz=ResourceName.ChatModel.OLLAMA_SETUP,
- connection="ollama_connection",
- model="qwen3:8b",
- subagents=["reviewer"],
-)
+{{< tab "Java" >}}
+```java
+parent.addResource("reviewer", ResourceType.AGENT, new ReviewAgent());
```
+{{< /tab >}}
-The model sees the child as a tool named `_subagent_reviewer`, described by the
-registration's description and input schema. The arguments the model sends
-become the child's input, and the child's outputs return to the model as the
-tool observation, so it can continue reasoning over them. For an internal
-sub-agent, the observation is the JSON encoding of the full output list
-described in [Call from an action](#call-from-an-action), even when the child
-emits a single output, so its shape does not depend on how many outputs the
-child emits.
+{{< /tabs >}}
-## Call from an action
+This form carries no description and no input declaration, so a chat model
rejects
+the sub-agent at setup time rather than calling it with no arguments. Only the
+descriptor form exposes a child to a model.
-An action can submit input to a child directly and await its outcome. This path
-does not need an input schema.
+There is no third form. An internal sub-agent cannot be declared in YAML,
because
Review Comment:
This statement is now out of date with the target branch: main has added
internal sub-agent references in YAML in commit `762c2c4ae`. For example, a
parent can declare `subagents: [{name: reviewer, agent: child}]`, where `child`
names another Agent in the YAML document. Both Java and Python loaders support
this.
When rebasing, could we replace this restriction with a reference to the
YAML guide's “Declaring sub-agents” section and preserve the navigation link
recently added on main?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
