GitHub user pltbkd created a discussion: [Feature] Sub-agent Resource for Flink
Agents - Framework part
# [Feature] Sub-agent Resource for Flink Agents
## Motivation
As discussed in #660, there are scenarios where a Flink Agent needs to call
external agents — for example, delegating a subtask to a remote LLM agent
service or a third-party agent platform. Currently, users can only do this by
manually writing remote calls inside actions, which means they miss out on
durable execution, checkpoint recovery, and tool call integration.
Beyond external agent calls, several related scenarios have emerged:
- **Parallel exploration**: Multiple agents working in different directions
simultaneously, each with its own reasoning loop and tools.
- **Independent review**: An agent providing an independent perspective — e.g.,
a reviewer agent that evaluates code without being influenced by the main
agent's context.
- **Reusing agents as sub-links**: Composing an agent defined by another team
as a building block in a larger agent pipeline.
These scenarios also create demand for **internal sub-agents** — the ability to
register a Flink Agents `Agent` as a sub-agent of another `Agent`, with proper
context isolation and execution semantics.
This proposal introduces `AGENT` as a first-class resource type in Flink
Agents, supporting both internal and external sub-agents with a unified API.
## Goals
- **Introduce `AGENT` resource type**: Sub-agents can be registered via
`addResource(name, AGENT, ...)` and accessed via `ctx.getResource(name,
AGENT)`, usable by both Actions and ChatModel.
- **Support for external, internal, and Remote Agents**: Supports custom
external agents, supports qualified internal agents, and will later implement
remote parallel execution of internal agents based on RpcOperator.
- **Context isolation, DurableExecute, and failover recovery**: Both internal
and external sub-agents benefit from durable execution with automatic failover
recovery. Internal agents introduce scope-level isolation.
The proposed changes have been initially validated with a POC (mainly on java
api). The overall design still has room for discussion — feedback and
suggestions are welcome.
---
## Proposed Interface
### 1. Resource Type
```java
public enum ResourceType {
CHAT_MODEL, CHAT_MODEL_CONNECTION, TOOL, EMBEDDING_MODEL,
VECTOR_STORE, PROMPT, MCP_SERVER, SKILLS,
AGENT // Sub-agent resource
}
```
### 2. Subagent Interface
`Subagent` is the common interface for all sub-agents. Both external
(`SubagentSetup`) and internal sub-agents implement it.
```java
public class Result {
boolean success; // success or failure
Object result; // type depends on agent implementation; built-in
agents return List<Object>
Exception exception; // present when success=false
}
public interface Subagent {
default Result call(RunnerContext ctx, Object prompt) {
return call(ctx, prompt, ctx.nextSessionId());
}
default Result call(RunnerContext ctx, Object prompt, String sessionId) {
return ctx.durableExecuteAsync(asAsyncCallable(ctx, prompt, sessionId));
}
default DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object
prompt) {
return asAsyncCallable(ctx, prompt, ctx.nextSessionId());
}
DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object prompt,
String sessionId);
}
```
```python
class Result:
success: bool
result: Any
exception: Optional[Exception]
class Subagent(Protocol):
def call(self, ctx: RunnerContext, prompt: Any, session_id: Optional[str] =
None) -> Result:
if session_id is None:
session_id = ctx.next_session_id()
return ctx.durable_execute_async(
self.as_async_callable(ctx, prompt, session_id))
def as_async_callable(self, ctx: RunnerContext, prompt: Any, session_id:
Optional[str] = None) -> DurableCallable[Result]:
if session_id is None:
session_id = ctx.next_session_id()
...
```
`SubagentSetup` is the abstract base for external sub-agents, implementing
`Subagent`:
```java
public abstract class SubagentSetup extends SerializableResource implements
Subagent {
@Override
@JsonIgnore
public ResourceType getResourceType() { return ResourceType.AGENT; }
}
```
```python
class SubagentSetup(SerializableResource, Subagent):
def get_resource_type(self) -> ResourceType:
return ResourceType.AGENT
```
### 3. Registration
Internal and external sub-agents are registered identically, only via
`addResource`, not via annotation:
```java
// Internal sub-agent
addResource("reviewer", ResourceType.AGENT, new ReviewerAgent());
// External sub-agent
addResource("summarizer", ResourceType.AGENT, new HttpSubagentSetup(endpoint));
```
```python
# Internal
self.add_resource("reviewer", ResourceType.AGENT, ReviewerAgent())
# External
self.add_resource("summarizer", ResourceType.AGENT, HttpSubagentSetup(endpoint))
```
```yaml
agents:
- name: reviewer-agent
module: com.example
clazz: ReviewerAgent
- name: root-agent
module: com.example
clazz: RootAgent
subagents:
- name: reviewer
agent_ref: reviewer-agent # internal: reference to another
agent definition
- name: summarizer
target_module: com.example # external: reference to
SubagentSetup subclass
target_clazz: HttpSubagentSetup
arguments:
endpoint: "http://summarizer:8080";
```
### 4. Calling
```java
// Synchronous
Subagent subagent = ctx.getResource("reviewer", ResourceType.AGENT);
Result result = subagent.call(ctx, prompt);
// Parallel
DurableCallable<Result> c1 = subagent1.asAsyncCallable(ctx, prompt1);
DurableCallable<Result> c2 = subagent2.asAsyncCallable(ctx, prompt2);
List<Result> results = ctx.durableExecuteAllAsync(c1, c2);
```
```python
# Synchronous
subagent: Subagent = ctx.get_resource("reviewer", ResourceType.AGENT)
result: Result = subagent.call(ctx, prompt)
# Parallel
c1 = subagent1.as_async_callable(ctx, prompt1)
c2 = subagent2.as_async_callable(ctx, prompt2)
results: list[Result] = await ctx.durable_execute_all_async(c1, c2)
```
### 5. Implementing External Sub-agent
```java
public class HttpSubagentSetup extends SubagentSetup {
private final String endpointUrl;
public HttpSubagentSetup(String endpointUrl) {
this.endpointUrl = endpointUrl;
}
@Override
public DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object
prompt, String sessionId) {
return () -> {
List<Object> output = httpClient.post(endpointUrl,
Map.of("prompt", prompt, "session", sessionId));
return new Result(true, output, null);
};
}
}
```
```python
class HttpSubagentSetup(SubagentSetup):
def __init__(self, endpoint: str):
self._endpoint = endpoint
def as_async_callable(self, ctx: RunnerContext, prompt: Any, session_id:
str) -> DurableCallable[Result]:
def _call() -> Result:
output = httpx.post(self._endpoint, json={"prompt": prompt,
"session": session_id})
return Result(success=True, result=output, exception=None)
return DurableCallable(_call)
```
---
## Execution Process
### 1. Compilation
Both internal and external sub-agents compile to `SerializableResourceProvider`
in the plan JSON, ensuring cross-language interoperability:
```
AGENT resource processing
│
├── SubagentSetup (external)
│ → SerializableResourceProvider
│
└── Agent (internal)
→ Compiled into child AgentPlan
→ InternalSubagentSetup → SerializableResourceProvider
```
Both paths produce `SerializableResourceProvider`, resulting in a uniform plan
JSON representation with no runtime difference.
Additionally, the compiler registers built-in sub-agents: if `CHAT_MODEL` is
available, `ReActAgent` is automatically registered as `"general-purpose"`.
This registration runs after all user-declared resources are extracted,
ensuring decorator-declared `CHAT_MODEL` is ready.
### 2. Execution Model
The default `call()` implementation executes the `DurableCallable` returned by
`asAsyncCallable` via `durableExecuteAsync`, leveraging the durable execution
mechanism for state recording and recovery.
**Parallel calls**: users manually split — call `asAsyncCallable` for each
sub-agent, then pass all `DurableCallable`s to `durableExecuteAllAsync` (to be
introduced in an upcoming discussion).
`call()` returns `Result`; sub-agent implementations should intercept internal
exceptions and populate Result, without directly exposing them to the caller.
`tool_call_action` constructs success/error `ToolResponseEvent` based on
Result, without try/catch.
### 3. ChatModel Integration
Sub-agents are exposed to the LLM as tools (Subagent as Tool pattern).
- **Prompt injection**: Available AGENT resources are listed in the system
prompt alongside tool descriptions, letting the LLM know which sub-agents it
can delegate to.
- **Dispatch**: `tool_call_action` checks resource type — TOOL resources go to
`tool.call()`, AGENT resources go to `subagent.call(ctx, prompt)`.
- **Result**: Returned via existing `ToolResponseEvent`.
```
LLM response → tool_calls → ToolRequestEvent → tool_call_action
│
├── name matches TOOL resource → tool.call(**kwargs)
└── name matches AGENT resource → subagent.call(ctx, prompt)
│
└── ToolResponseEvent → back to LLM
```
LLM APIs (OpenAI, Ollama, etc.) only have tool/function call concepts — there
is no "agent call". Sub-agents are exposed as tools with a single `prompt`
parameter. The `tool_call_action` already holds `ctx` as an action parameter,
which is passed directly to `subagent.call(ctx, prompt)`.
---
## Internal Sub-agent Design Preview
Internal sub-agent compilation — registering `Agent` instances as sub-agents —
will be discussed in a future proposal. The resource model and calling API are
already designed to accommodate it.
### 1. Compilation
The sub-agent's `Agent` is compiled into an independent `AgentPlan`
(childPlan), wrapped as `InternalSubagentSetup(childPlan, scope)`. The child
plan inherits the parent plan's `AgentConfiguration`. Circular references are
detected at compile time. `scope` is the sub-agent's resource name, used for
event routing.
### 2. Invocation
Calling a sub-agent emits a `SubagentCallEvent`. During async execution, the
coroutine yields, and the AEO processes the event in a subsequent mailbox
cycle, scheduling the sub-agent's internal actions. Only the event sending
timing is modified for `SubagentCallEvent` to support this execution model —
`asAsyncCallable` sends `SubagentCallEvent` on the mailbox thread, and
`durableExecuteAsync` blocks on the worker thread waiting for the result.
### 3. Scope-based Execution
`SubagentCallEvent` is converted to `InputEvent` within the sub-agent's
scheduling scope. Intermediate events are dispatched only within that scope —
they do not leak to the parent agent or other sub-agents. The result
`OutputEvent` is intercepted and returned to the caller. When all pending
events are processed and no actions are running (quiesce), the call completes
automatically.
### 4. Isolation
`SubagentRunnerContext` provides isolated context:
- **Actions and events**: Sub-agent uses its own `AgentPlan`'s definitions,
fully independent from parent.
- **Memory**: Read-only access to parent memory; Sensory memory is allowed but
not written back to parent scope memory.
- **Resources**: Uses own registered resources first, falls back to parent
(parent resource cache inheritance).
- **Config**: Inherits parent plan's `AgentConfiguration`.
Nested sub-agents are supported recursively via parent resource cache fallback.
### 5. Fault Tolerance
- **Completed calls**: Intercepted directly by DurableExecute mechanism;
failover hits cached result.
- **Incomplete calls**: Supports ActionState; after failover, deterministic
SessionId associates the new sub-agent call with recovered Actions, enabling
intermediate state recovery rather than full re-execution.
---
## Design Discussion
### 1. Internal vs External Capability Parity
External and internal sub-agents provide identical user-facing API
(registration, `call()`, `asAsyncCallable()`). However, the underlying
implementations differ significantly — external sub-agents are black-box calls
(HTTP/gRPC), while internal sub-agents are framework-internal plan nesting
execution.
| Capability | External Sub-agent | Internal Sub-agent |
|---|---|---|
| **Context isolation** | Managed by external system via deterministic
SessionId | Framework creates isolated `SubagentRunnerContext` |
| **Memory** | Maintained by external system | Read-only parent access,
write-isolated, not persisted |
| **State recovery** | DurableExecute persists call records | Same as external;
additionally, deterministic SessionId links call chain to sub-agent internal
action ActionState, supporting intermediate state recovery |
| **Parallel execution** | Multiple agents in parallel via
`durableExecuteAllAsync` | One action at a time per ActionExecutionOperator,
but coroutine-based async partial parallelism |
| **Backpressure** | Via `durableExecuteAsync` worker pool | Same as external;
needs to consider subagent nested call deadlock issues |
### 2. Introducing `SubagentCompatible` Marker
When an `Agent` is registered as a sub-agent, it runs in an isolated
`SubagentRunnerContext`, meaning certain capabilities are restricted, such as
the inability to write to memory. Not all Agents can correctly accommodate
these constraints.
To ensure registration correctness, we propose a `SubagentCompatible` marker
(interface / annotation / yaml field) that an Agent declares to indicate it is
safe to run as a sub-agent. The framework validates this marker at compile
time: when an unmarked `Agent` is registered as a sub-agent, compilation fails
fast.
This provides an explicit contract: Agent authors must evaluate and declare
compatibility with sub-agent constraints (e.g., no dependency on writing to
parent memory, no reliance on cross-scope events), avoiding hard-to-debug
behavioral anomalies at runtime.
We need to evaluate what capability restrictions sub-agents have, to confirm
whether this marker is necessary.
---
## To Be Refined
- **Timeout / cancellation semantics**: `call()` currently does not provide
timeout or cancellation. Further investigation of requirements is needed to
refine the design.
- **Sub-agent prompt injection and dispatch chain for ChatModel**: The prompt
injection format for exposing sub-agents as tools to the LLM, and the
`tool_call_action` dispatch chain, need further clarification.
- **External agent support framework**: Consider providing `SubagentSetup`
implementation helpers for gRPC, A2A (Agent-to-Agent) protocol, etc., to reduce
the cost of integrating external agents.
- **RpcOperator-based RemoteAgent design**: The remote execution approach via
RpcOperator (FLIP-577) on a separate ActionExecutionOperator is yet to be
designed.
GitHub link: https://github.com/apache/flink-agents/discussions/909
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
