GitHub user alnzng added a comment to the discussion: [Feature] Sub-agent
Resource for Flink Agents - Framework part
>What do you think about the interface design?
- For the first version I think your assumption is reasonable: the caller (e.g.
a supervisor) should expect a full result back from the sub-agents it manages.
Request/response is the right default.
- For the future, we might want a "talker / thinker"-style model: the
supervisor (thinker) reacts to a continuous stream of output from its
sub-agents (talkers) rather than blocking for one final answer per call. That's
the similar idea to the pub/sub interface you mentioned. So I'd keep
request/response for v1 and just keep the door open for a streaming result
later.
>Therefore, I think the re-post issue you mentioned can be avoided with the
>existing mechanisms.
Agreed, I think you're right. It looks like reconciler() is exactly the
submit/probe split, I didn't realize that `DurableCallable` already offers this
capability. We can use the codes like below:
```java
return new BaseSubagentCallable(sessionId, callId) {
@Override
protected Object callInternal() throws Exception { // normal
path: submit + consume
String threadId = sessionId;
httpPost(base + "/threads/" + threadId + "/messages",
Map.of("message", Map.of("content", prompt),
"idempotency_key", sessionId + "#" + callId));
return drainSSE(base + "/threads/" + threadId + "/stream");
}
@Override
public Callable<Result> reconciler() { // recovery
path: probe, don't re-submit
return () -> {
var st = httpGet(base + "/threads/" + sessionId);
switch (st.status) {
case "completed": return Result.ok(st.savedOutput);
case "running": return Result.ok(drainSSE(...)); //
re-attach
case "failed": throw new RuntimeException(st.err);
case "not_started": return Result.ok(callInternal()); // safe
iff submit is idempotent
}
};
}
};
```
So the mechanism is already there, that part is clear to me now.
----------
But one thing I am still not sure: should we provide some built-in abstraction
for this pattern, instead of let every integration write it by hand?
My worry is, a big group of external agents (LangGraph, OpenAI Assistants runs,
A2A long-running task, etc.) are actually the same shape: an async job with
three steps — submit → observe → status. If we don't give an abstraction, then
every author needs to write the same callInternal + reconciler split again and
again. More important, they also need to understand why there are two methods,
when reconciler will be called, and how to handle the crash window. This is
framework recovery knowledge, but it is not really their domain. They only want
to describe their remote agent.
So maybe we can add an additive specialization, for example
`AsyncSubagentSetup`, where the author only provides the primitives they
already understand:
- submit(ctx, id, callId, prompt) — start the job;
- awaitResult(ctx, id) — block / stream until the final result (happy path);
- status(ctx, id) — a read-only probe, return {not_started, running,
completed, failed} (and the result if it is inline);
Then the framework composes `callInternal` and `reconciler` from these
primitives, propagates `sessionId#callId` as the remote idempotency key
automatically, and owns the crash-window switch. The author never writes
reconciler, and does not need to reason about recovery.
WDYT, is this the right place for that framework layer?
GitHub link:
https://github.com/apache/flink-agents/discussions/909#discussioncomment-17886293
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
