yunfengzhou-hub commented on code in PR #1268:
URL: https://github.com/apache/flink-agents/pull/1268#discussion_r4309928701
##########
docs/content/docs/development/multi-agent/subagents.md:
##########
@@ -312,30 +412,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 {#declare-an-external-sub-agent}
+
+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, *deferred* and *async
+(pub/sub)*. Which one you write follows from your remote interface.
+
+| | Deferred | Async (pub/sub) |
+|---|---|---|
+| Work starts | When the caller first awaits the result | When the caller
submits |
+| You implement | `prepare` | `call_submit_request`, `call_query_status`,
`call_fetch_result` |
+| Choose it when | One request returns the result | The remote hands back a
run you query later |
+
+#### 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. It takes no arguments. Return the recovered outcome,
because
+ # an exception here is a system-level failure that fails the action
instead of
+ # reaching the caller.
+ 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:
+ return SubagentResult.error("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, because an exception here is
a
+// system-level failure that fails the action instead of reaching the caller.
+@Override
+public Callable<SubagentResult> reconciler() {
+ return () -> {
+ // fetchReviewByKey is your own lookup; not part of Flink Agents.
+ String found = fetchReviewByKey(endpoint, key);
+ if (found == null) {
+ return SubagentResult.error("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 starts the remote 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.
+
+The base class owns that submit-poll-fetch composition, 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. The
+difference you see is where the result goes.
+
+In model-driven delegation the result returns to the model. 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 goes back to the model as the
tool
+observation, and the model reasons over it until a final `ChatResponseEvent`
reaches
+your action.
+
+In caller-driven delegation the result returns to your action. An action that
needs
+the delegated outcome 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
Review Comment:
Done. Manage sub-agent calls, and Results and failures with Durability and
failover and Metrics and Trace Log promoted to their own sections.
--
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]
