yunfengzhou-hub commented on code in PR #1206:
URL: https://github.com/apache/flink-agents/pull/1206#discussion_r4245007497
##########
python/flink_agents/api/agents/react_agent.py:
##########
@@ -105,6 +112,8 @@ def __init__(
chat_model: ResourceDescriptor,
prompt: Prompt | None = None,
output_schema: type[BaseModel] | RowTypeInfo | None = None,
+ subagent_description: str | None = None,
+ subagent_input_schema: str | None = None,
Review Comment:
**Should-fix:** overriding `subagent_description` (or supplying a prompt)
without
also overriding `subagent_input_schema` can silently produce a wrong prompt.
The default schema declares a single `input` string field, so the parent
model
always calls the child with `{"input": ...}`. If the child is constructed
with a
prompt whose placeholders are not `input` (e.g. `{diff}`), `start_action`
takes the
prompt branch and calls `prompt.format_messages(input=...)`. Because
`SafeFormatter`
returns the literal `{diff}` for a missing key instead of raising, the model
receives
a prompt with an unreplaced `{diff}` — no error, just a broken delegation.
Could we fail fast (or warn) at construction time when a prompt is provided
whose
placeholders don't match the declared/default input schema? Or derive the
default
schema from the prompt's placeholders. At minimum, the
`subagent_input_schema`
docstring should call out this coupling explicitly.
##########
plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonSerializer.java:
##########
@@ -128,6 +128,12 @@ private void serializeInternalSubagentProvider(
gen.writeStringField("name", provider.getName());
gen.writeStringField("type", provider.getType().getValue());
gen.writeStringField("scope", provider.getScope());
+ if (provider.getDescription() != null) {
+ gen.writeStringField("description", provider.getDescription());
+ }
+ if (provider.getInputSchema() != null) {
+ gen.writeStringField("inputSchema", provider.getInputSchema());
Review Comment:
**Should-fix (cross-language, relevant to #1141):** the metadata field name
and the
Java→Python conversion path are asymmetric.
1. Field name: Java serializes `inputSchema` (camelCase, here), while the
Python
compiler writes `input_schema` (snake_case, `agent_plan.py:532`). A plan
JSON
produced by one side is not read back with the same key by the other.
2. The Python `__custom_deserialize` branch that converts a Java-emitted
`InternalSubagentProvider` into a `PythonSerializableResourceProvider`
(`agent_plan.py:152-168`) only carries `child_plan` and `scope` — it drops
`description`/`inputSchema` entirely. So a Java-compiled sub-agent that
reaches
Python loses the metadata this PR adds.
Neither bites a single-language job, but both will break model-driven
delegation for
a cross-language sub-agent. Worth aligning the key name and carrying the two
fields
through the conversion before #1141 lands on top of this.
##########
docs/content/docs/development/react_agent.md:
##########
@@ -332,3 +344,187 @@ RowTypeInfo myRowTypeInfo =
{{< /tabs >}}
+## As a General-Purpose Sub-Agent
+
+A ReActAgent can also run as a sub-agent of a parent agent. Register it as an
`AGENT` resource of the parent and it runs as an internal sub-agent — an
isolated, durable agent inside the same job — that keeps its own event loop,
resources and conversation. The parent reaches it through two paths: the parent
model delegating to it as a tool, or a parent action submitting a prompt
directly.
+
+### Register the Child Agent
+
+Register the child under a name, like any other resource. The parent can be
any agent, including another ReActAgent.
+
+{{< tabs "Register Sub-Agent Child" >}}
+
+{{< tab "Python" >}}
+```python
+reviewer = ReActAgent(
+ chat_model=ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP,
+ connection="reviewer_connection",
+ model="qwen3:8b",
+ ),
+)
+# The connection is registered on the child itself, keeping it self-contained.
+reviewer.add_resource(
+ "reviewer_connection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+ ResourceDescriptor(clazz=ResourceName.ChatModel.OLLAMA_CONNECTION),
+)
+
+parent = ReActAgent(chat_model=parent_chat_model_descriptor)
+parent.add_resource("reviewer", ResourceType.AGENT, reviewer)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+ReActAgent reviewer =
+ new ReActAgent(
+
ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "reviewerConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .build(),
+ null,
+ null);
+// The connection is registered on the child itself, keeping it self-contained.
+reviewer.addResource(
+ "reviewerConnection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+
ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_CONNECTION)
+ .build());
+
+ReActAgent parent = new ReActAgent(parentChatModelDescriptor, null, null);
+parent.addResource("reviewer", ResourceType.AGENT, reviewer);
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+{{< hint warning >}}
+Keep the child self-contained and register the chat model connection it uses
on the child itself, as above. This is required in Python, where a child
resolves resources against its own plan only — a connection registered solely
on the parent is not visible to the child.
+{{< /hint >}}
+
+### Model-Driven Delegation
+
+Declare the child's name under `subagents` of the parent's chat model, and the
parent model can delegate to it: the child is presented as a `_subagent_<name>`
tool whose description and input schema come from the child agent.
+
+{{< tabs "Model-Driven Delegation" >}}
+
+{{< tab "Python" >}}
+```python
+parent = ReActAgent(
+ chat_model=ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP,
+ connection="my_ollama_connection",
+ model="qwen3:8b",
+ subagents=["reviewer"],
+ ),
+)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+ReActAgent parent =
+ new ReActAgent(
+
ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "myOllamaConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .addInitialArgument("subagents", List.of("reviewer"))
+ .build(),
+ null,
+ null);
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+A ReActAgent child works with the defaults out of the box. The default input
schema declares a single `input` string field, and the framework routes that
field to the child as its user message — directly when the child declares no
prompt, or through the child's `{input}` placeholder when it declares one. The
child's answer flows back to the parent model as the tool observation, so the
parent can continue reasoning over it.
+
+### Caller-Driven Delegation
+
+A parent action can submit a prompt to the child directly and await its
outcome.
+
+{{< tabs "Caller-Driven Delegation" >}}
+
+{{< tab "Python" >}}
+```python
+class DelegatingAgent(Agent):
+ @action(InputEvent.EVENT_TYPE)
+ @staticmethod
+ async def delegate(event: Event, ctx: RunnerContext) -> None:
+ child = ctx.get_resource("reviewer", ResourceType.AGENT)
+ future = await child.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 child =
+ (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+ SubagentResult result = child.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 >}}
+
+The awaited value is a `SubagentResult`. On success, its `result` field
(`getResult()` in Java) is the list of outputs the child emitted — a ReActAgent
child emits a single final answer, so the first element is what you usually
want. On failure, `success` is `False` / `isSuccess()` is `false`, and
`error_message` / `getErrorMessage()` carries the reason.
Review Comment:
**Question:** for model-driven delegation the child's answer reaches the
parent model
as a list-wrapped observation (e.g. `["reviewed -> HELLO WORLD!"]`), because
`SubagentResult.result` is the list of outputs the child emitted and
`to_chat_message_content` JSON-encodes it as-is.
The docs here correctly tell a *caller-driven* user to take
`result.result[0]`, but in
the model-driven path the model sees the raw JSON list. For a ReActAgent
child that
always emits exactly one final answer, is the list wrapper intentional in
the tool
observation, or should the single-output case be flattened so the model
observes the
plain string? A one-line note either way would help readers.
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -745,6 +745,20 @@ def __scoped_plan_and_cache(self, plan_json: str) -> tuple:
self.__scoped_resource_caches[plan_json] = entry
return entry
+ def __active_plan(self) -> Any:
+ """The plan in effect: the child plan's while inside a sub-agent call,
else
+ the root plan's.
+
+ Mirrors the Java context's ``currentPlan()``: action-config lookups
must
+ resolve against the plan whose action is executing, so a child agent's
+ action reads its own config instead of failing on the root plan, which
+ does not carry that action.
+ """
+ plan_json = self._j_runner_context.getActiveScopePlanJson()
Review Comment:
**Question (perf):** `__active_plan()` crosses the py4j boundary via
`getActiveScopePlanJson()` on every `action_config` /
`get_action_config_value` read.
The parsed plan is cached in `__scoped_resource_caches`, and the Java getter
is cheap
(returns a cached string, no re-serialization), but the getter call itself
is not
memoized, so config-heavy actions pay a JNI round-trip per lookup.
Is that acceptable on the hot path, or is it worth caching the active plan
JSON
(per scope) on the Python side the first time it's seen?
##########
plan/src/test/java/org/apache/flink/agents/plan/serializer/ResourceProviderDeserializerTest.java:
##########
@@ -120,4 +125,68 @@ public void
testDeserializePythonSerializableResourceProvider() throws Exception
assertEquals("FunctionTool",
pythonSerializableResourceProvider.getClazz());
assertEquals(serialized,
pythonSerializableResourceProvider.getSerialized());
}
+
+ @Test
+ public void testInternalSubagentProviderRoundTrip() throws Exception {
Review Comment:
**Nit (commit message):** commit 3e81f73 says "Also fixes a latent
serializer bug the
new round-trip tests exposed: AgentPlanJsonSerializer never closed the
object it
opens." But `AgentPlanJsonSerializer.java` is not modified in this PR — it's
byte-identical to main, which already has the closing `writeEndObject()`
(and the
explanatory comment) from #1138; the generator is balanced (6 start / 6 end).
The round-trip tests added here are great, but the "fixes a latent
serializer bug"
paragraph describes a change that isn't in this diff. Could we drop/reword
that
paragraph so the commit message matches what's actually changed?
--
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]
