yunfengzhou-hub commented on code in PR #1268:
URL: https://github.com/apache/flink-agents/pull/1268#discussion_r4300860093
##########
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:
Done, and the async execution ref link went with the type argument. Invoke
now opens on where the result goes. The prepare, durable id, and reconciler
contracts are unchanged.
--
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]
