From 7cf44e2af93d9d9351b759350719db5a8f08eed1 Mon Sep 17 00:00:00 2001 From: Yi Zhang Date: Wed, 30 Sep 2026 16:39:19 +0800 Subject: [PATCH 1/4] [api][java] Accept string and map input in the ReActAgent start action The start action dispatched on ClassUtils.isPrimitiveOrWrapper, which does not cover String, so a String input failed both with and without a user prompt, while the Python start action and the documented input contract (development/react_agent.md lists str under the {input} placeholder) both accept it. A Map input only worked by falling into the Pojo branch's JSON round-trip, and that round-trip read the JSON into the raw Map.class, so any non-String Pojo field or Map value surfaced as a bare ClassCastException in Prompt#format, far from the input dispatch and with no guidance, while Python stringifies every value. Route String through the primitive branch (the {input} placeholder) and give Map an explicit branch that stringifies keys and values, mirroring the Python dict branch. The Pojo branch now stringifies every field the same way instead of assigning the raw read result. Values keep their language-native string form (String.valueOf on Java, str() on Python), including nested ones. The unsupported-input error now lists Map among the accepted kinds. The map-without-prompt rejection is unchanged, as both ends reject it today. Added a ReActAgentTest matrix covering primitive, string, map, Row and Pojo inputs, including a Pojo with a non-String field. Generated-by: Qoder --- .../flink/agents/api/agents/ReActAgent.java | 22 +++- .../agents/api/agents/ReActAgentTest.java | 107 ++++++++++++++++++ 2 files changed, 126 insertions(+), 3 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java index 70a858bb2..c6293a3cd 100644 --- a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java +++ b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java @@ -274,7 +274,10 @@ public static void startAction(Event event, RunnerContext ctx) { } List inputMessages = new ArrayList<>(); - if (ClassUtils.isPrimitiveOrWrapper(input.getClass())) { + // A String is not a primitive wrapper, but the documented input contract covers it like + // one: without a prompt it becomes the user message, with a prompt it fills the {input} + // placeholder. + if (input instanceof String || ClassUtils.isPrimitiveOrWrapper(input.getClass())) { if (userPrompt != null) { inputMessages = userPrompt.formatMessages( @@ -297,14 +300,27 @@ public static void startAction(Event event, RunnerContext ctx) { for (String name : Objects.requireNonNull(userInput.getFieldNames(true))) { fields.put(name, String.valueOf(userInput.getField(name))); } + } else if (input instanceof Map) { + for (Map.Entry entry : ((Map) input).entrySet()) { + fields.put(String.valueOf(entry.getKey()), String.valueOf(entry.getValue())); + } } else { // regard as pojo ObjectMapper objectMapper = new ObjectMapper(); try { - fields = mapper.readValue(objectMapper.writeValueAsString(input), Map.class); + // A read into the raw Map.class ignores the declared String values, so JSON + // numbers and booleans come back as Integer/Boolean. Stringify every value + // like the Map branch above, otherwise Prompt#format fails with a bare + // ClassCastException when it casts a placeholder value to String. + Map pojoFields = + mapper.readValue(objectMapper.writeValueAsString(input), Map.class); + for (Map.Entry entry : pojoFields.entrySet()) { + fields.put( + String.valueOf(entry.getKey()), String.valueOf(entry.getValue())); + } } catch (JsonProcessingException e) { throw new RuntimeException( String.format( - "Input must be primitive type, Row or Pojo, but is %s", + "Input must be primitive type, Row, Map or Pojo, but is %s", input.getClass())); } } diff --git a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java index d4474a5c2..ddcdcafbe 100644 --- a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java @@ -43,6 +43,7 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.types.Row; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; @@ -474,6 +475,112 @@ public void testConstructorSkillsKeepDescriptorLanguage() { assertThat(actual.>getArgument("skills")).containsExactly("a", "b"); } + // --- startAction input dispatch --- + + @Test + @DisplayName("A primitive input without a prompt becomes the user message") + public void testPrimitiveInputWithoutPromptBecomesUserMessage() { + RunnerContext ctx = mock(RunnerContext.class); + + ReActAgent.startAction(new InputEvent(42), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("42"); + } + + @Test + @DisplayName("A string input without a prompt becomes the user message") + public void testStringInputWithoutPromptBecomesUserMessage() { + RunnerContext ctx = mock(RunnerContext.class); + + ReActAgent.startAction(new InputEvent("analyze this review"), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("analyze this review"); + } + + @Test + @DisplayName("A string input with a prompt fills the {input} placeholder") + public void testStringInputWithPromptFillsInputPlaceholder() throws Exception { + RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task: {input}")); + + ReActAgent.startAction(new InputEvent("analyze this review"), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: analyze this review"); + } + + @Test + @DisplayName("A map input without a prompt reports that a prompt is required") + public void testMapInputWithoutPromptReportsPromptRequirement() { + RunnerContext ctx = mock(RunnerContext.class); + + assertThatThrownBy( + () -> + ReActAgent.startAction( + new InputEvent(Map.of("task", "analyze this review")), ctx)) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("not primitive types"); + } + + @Test + @DisplayName("A map input with a prompt uses the keys as placeholders") + public void testMapInputWithPromptUsesKeysAsPlaceholders() throws Exception { + RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task: {task}")); + + ReActAgent.startAction(new InputEvent(Map.of("task", "analyze this review")), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: analyze this review"); + } + + @Test + @DisplayName("A Row input with a prompt uses the field names as placeholders") + public void testRowInputWithPromptUsesFieldNamesAsPlaceholders() throws Exception { + RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task: {task}")); + Row row = Row.withNames(); + row.setField("task", "analyze this review"); + + ReActAgent.startAction(new InputEvent(row), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: analyze this review"); + } + + @Test + @DisplayName("A Pojo with a non-String field fills placeholders with its stringified value") + public void testPojoWithNonStringFieldFillsPlaceholdersWithStringifiedValue() throws Exception { + RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task: {count}")); + WithCount input = new WithCount(); + input.count = 5; + + ReActAgent.startAction(new InputEvent(input), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: 5"); + } + + @Test + @DisplayName("An input of an unsupported kind reports the accepted kinds") + public void testInputOfUnsupportedKindReportsAcceptedKinds() throws Exception { + RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task: {input}")); + + assertThatThrownBy(() -> ReActAgent.startAction(new InputEvent(new FieldLess()), ctx)) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Row, Map or Pojo") + .hasMessageContaining("FieldLess"); + } + + private static RunnerContext contextWithPrompt(Prompt prompt) throws Exception { + RunnerContext ctx = mock(RunnerContext.class); + when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT)).thenReturn(prompt); + return ctx; + } + + private static ChatMessage soleUserMessage(RunnerContext ctx) { + ArgumentCaptor captor = ArgumentCaptor.forClass(Event.class); + verify(ctx).sendEvent(captor.capture()); + ChatRequestEvent request = (ChatRequestEvent) captor.getValue(); + assertThat(request.getMessages()).hasSize(1); + ChatMessage message = request.getMessages().get(0); + assertThat(message.getRole()).isEqualTo(MessageRole.USER); + return message; + } + private static ReActAgent agentWithSchema(Object outputSchema) { return agentWithSchema(outputSchema, null); } From 6a70342c3cfe624c197ebdb186664fc246d97ce9 Mon Sep 17 00:00:00 2001 From: Yi Zhang Date: Tue, 13 Oct 2026 18:51:51 +0800 Subject: [PATCH 2/4] [api][java][python] Let a ReActAgent run as a sub-agent child A model-driven call hands the child the arguments of its input schema as a map. Without a user-supplied prompt, a ReActAgent rejected every map, so a child registered with the single "input" field schema could not be called. A map with exactly one "input" entry now becomes the user message directly, mirroring the prompt-less primitive branch; any other map without a prompt still fails fast. A child scope also falls back to the root agent's resources, so an unconfigured child could pick up its parent ReActAgent's built-in user or schema prompt under the same name. Gate both lookups on the running scope's own action config, and report a declared prompt that cannot be resolved as a broken plan. Generated-by: Qoder --- .../flink/agents/api/agents/ReActAgent.java | 133 ++++++++++++------ .../agents/api/agents/ReActAgentTest.java | 115 ++++++++++++++- python/flink_agents/api/agents/react_agent.py | 77 +++++++--- .../api/agents/tests/test_react_agent.py | 120 ++++++++++++++++ 4 files changed, 380 insertions(+), 65 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java index c6293a3cd..372a80d7c 100644 --- a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java +++ b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java @@ -81,6 +81,15 @@ public class ReActAgent extends Agent { private static final String SKILLS_SOURCES_ARGUMENT = "skills_sources"; private static final String SKILLS_RESOURCE_NAME = "_react_agent_skills"; + /** + * Action config flag recording that this agent registered {@link #DEFAULT_USER_PROMPT}, as + * {@code output_schema} does for {@link #DEFAULT_SCHEMA_PROMPT}. A sub-agent's resource lookup + * falls back to the root agent's resources, so {@link #startAction} looks up a built-in prompt + * only when the running scope's own action config declares it; otherwise an unconfigured child + * would pick up a parent ReActAgent's prompt under the same name. + */ + private static final String HAS_USER_PROMPT = "has_user_prompt"; + private static final ObjectMapper mapper = new ObjectMapper(); public ReActAgent( @@ -171,6 +180,7 @@ public ReActAgent( if (prompt != null) { this.addResource(DEFAULT_USER_PROMPT, ResourceType.PROMPT, prompt); + actionConfig.put(HAS_USER_PROMPT, true); } try { @@ -260,18 +270,27 @@ private static ResourceDescriptor withSkills( descriptor.getLanguage(), descriptor.getModule(), descriptor.getClazz(), arguments); } + /** + * Whether {@code input} is a map holding a single "input" entry - the one structured shape + * {@link #startAction} converts without a user-supplied prompt, so a sub-agent declared with a + * single "input" field needs no prompt. Keyed by equals, so a non-String key never matches. + */ + private static boolean isSingleInputMap(Object input) { + return input instanceof Map + && ((Map) input).size() == 1 + && ((Map) input).containsKey("input"); + } + public static void startAction(Event event, RunnerContext ctx) { // Built-in events reach an action as their concrete subclass (restored at the JSON // boundary), so the dispatched InputEvent is cast directly. InputEvent inputEvent = (InputEvent) event; Object input = inputEvent.getInput(); - Prompt userPrompt; - try { - userPrompt = (Prompt) ctx.getResource(DEFAULT_USER_PROMPT, ResourceType.PROMPT); - } catch (Exception e) { - userPrompt = null; - } + Prompt userPrompt = + Boolean.TRUE.equals(ctx.getActionConfigValue(HAS_USER_PROMPT)) + ? declaredPrompt(ctx, DEFAULT_USER_PROMPT) + : null; List inputMessages = new ArrayList<>(); // A String is not a primitive wrapper, but the documented input contract covers it like @@ -287,51 +306,65 @@ public static void startAction(Event event, RunnerContext ctx) { } } else { if (userPrompt == null) { - throw new RuntimeException( - String.format( - "The input type is %s, which is not primitive types," - + " user should provide prompt to help convert it to ChatMessage", - input.getClass())); - } - - Map fields = new HashMap<>(); - if (input instanceof Row) { - Row userInput = (Row) input; - for (String name : Objects.requireNonNull(userInput.getFieldNames(true))) { - fields.put(name, String.valueOf(userInput.getField(name))); - } - } else if (input instanceof Map) { - for (Map.Entry entry : ((Map) input).entrySet()) { - fields.put(String.valueOf(entry.getKey()), String.valueOf(entry.getValue())); + if (isSingleInputMap(input)) { + // The one structured shape that works without a user-supplied prompt: a + // single "input" entry, the arguments of a model-driven call against an + // input schema declaring only that field. The value becomes the user message + // directly, mirroring the prompt-less primitive branch. + inputMessages.add( + new ChatMessage( + MessageRole.USER, + String.valueOf(((Map) input).get("input")))); + } else { + throw new RuntimeException( + String.format( + "The input type is %s, which is not primitive types," + + " user should provide prompt to help convert it to" + + " ChatMessage", + input.getClass())); } - } else { // regard as pojo - ObjectMapper objectMapper = new ObjectMapper(); - try { - // A read into the raw Map.class ignores the declared String values, so JSON - // numbers and booleans come back as Integer/Boolean. Stringify every value - // like the Map branch above, otherwise Prompt#format fails with a bare - // ClassCastException when it casts a placeholder value to String. - Map pojoFields = - mapper.readValue(objectMapper.writeValueAsString(input), Map.class); - for (Map.Entry entry : pojoFields.entrySet()) { + } else { + Map fields = new HashMap<>(); + if (input instanceof Row) { + Row userInput = (Row) input; + for (String name : Objects.requireNonNull(userInput.getFieldNames(true))) { + fields.put(name, String.valueOf(userInput.getField(name))); + } + } else if (input instanceof Map) { + for (Map.Entry entry : ((Map) input).entrySet()) { fields.put( String.valueOf(entry.getKey()), String.valueOf(entry.getValue())); } - } catch (JsonProcessingException e) { - throw new RuntimeException( - String.format( - "Input must be primitive type, Row, Map or Pojo, but is %s", - input.getClass())); + } else { // regard as pojo + ObjectMapper objectMapper = new ObjectMapper(); + try { + // A read into the raw Map.class ignores the declared String values, so JSON + // numbers and booleans come back as Integer/Boolean. Stringify every value + // like the Map branch above, otherwise Prompt#format fails with a bare + // ClassCastException when it casts a placeholder value to String. + Map pojoFields = + mapper.readValue(objectMapper.writeValueAsString(input), Map.class); + for (Map.Entry entry : pojoFields.entrySet()) { + fields.put( + String.valueOf(entry.getKey()), + String.valueOf(entry.getValue())); + } + } catch (JsonProcessingException e) { + throw new RuntimeException( + String.format( + "Input must be primitive type, Row, Map or Pojo, but is %s", + input.getClass())); + } } - } - inputMessages = userPrompt.formatMessages(MessageRole.USER, fields); + inputMessages = userPrompt.formatMessages(MessageRole.USER, fields); + } } Object outputSchema = ctx.getActionConfigValue("output_schema"); - Prompt schemaPrompt = promptOrNull(ctx, DEFAULT_SCHEMA_PROMPT); - if (schemaPrompt != null) { + if (outputSchema != null) { + Prompt schemaPrompt = declaredPrompt(ctx, DEFAULT_SCHEMA_PROMPT); // A plan built before the native prompt existed has none, and then adds no // instruction on the native path. Prompt instruction = @@ -383,6 +416,24 @@ private static boolean appliesSchemaNatively(RunnerContext ctx, Object outputSch } } + /** + * Resolves a built-in prompt the running scope's action config declares. The constructor + * registers the prompt together with its declaration, so a failed lookup is a broken plan and + * is reported rather than silently dropping the prompt. + */ + private static Prompt declaredPrompt(RunnerContext ctx, String name) { + try { + return (Prompt) ctx.getResource(name, ResourceType.PROMPT); + } catch (Exception e) { + throw new IllegalStateException( + String.format( + "Prompt %s is declared by this agent's action config but cannot be" + + " resolved.", + name), + e); + } + } + @Action(EventType.ChatResponseEvent) public static void stopAction(Event event, RunnerContext ctx) { ChatResponseEvent chatResponse = (ChatResponseEvent) event; diff --git a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java index ddcdcafbe..b45613984 100644 --- a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java @@ -63,6 +63,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.tuple; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -313,6 +314,7 @@ private static RunnerContext contextFor(ReActAgent agent, Resource chatModel) th when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT)) .thenReturn((Resource) prompts.get("_default_user_prompt")); when(ctx.getActionConfigValue("output_schema")).thenReturn(config.get("output_schema")); + when(ctx.getActionConfigValue("has_user_prompt")).thenReturn(config.get("has_user_prompt")); return ctx; } @@ -565,16 +567,125 @@ public void testInputOfUnsupportedKindReportsAcceptedKinds() throws Exception { .hasMessageContaining("FieldLess"); } + // --- built-in prompt scoping --- + + @Test + @DisplayName("A configured agent declares its built-in prompts in the start action config") + public void testConfiguredAgentDeclaresBuiltInPromptsInActionConfig() { + ReActAgent agent = + new ReActAgent( + chatModelDescriptor(), Prompt.fromText("Task: {input}"), WithCount.class); + + Map config = agent.getActions().get("startAction").f2; + assertThat(config).containsEntry("has_user_prompt", true); + assertThat(config).containsEntry("output_schema", WithCount.class); + } + + @Test + @DisplayName("A bare agent declares no built-in prompt in the start action config") + public void testBareAgentDeclaresNoBuiltInPrompt() { + Map config = bareAgent().getActions().get("startAction").f2; + + assertThat(config).doesNotContainKeys("has_user_prompt", "output_schema"); + } + + @Test + @DisplayName("A prompt reachable but not declared by this agent is not looked up") + public void testUndeclaredPromptsAreNotLookedUp() throws Exception { + // As a sub-agent, a lookup missing locally falls back to the root agent's resources, so + // a parent ReActAgent's prompts are reachable under the same names. + RunnerContext ctx = mock(RunnerContext.class); + when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT)) + .thenReturn(Prompt.fromText("PARENT ONLY: {input}")); + when(ctx.getResource("_default_schema_prompt", ResourceType.PROMPT)) + .thenReturn(Prompt.fromText("PARENT SCHEMA")); + + ReActAgent.startAction(new InputEvent("review the diff"), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("review the diff"); + verify(ctx, never()).getResource(anyString(), any()); + } + + @Test + @DisplayName("A declared output schema inserts the schema instruction before the user message") + public void testDeclaredOutputSchemaInsertsSchemaInstruction() throws Exception { + RunnerContext ctx = mock(RunnerContext.class); + when(ctx.getActionConfigValue("output_schema")).thenReturn(WithCount.class); + when(ctx.getResource("_default_schema_prompt", ResourceType.PROMPT)) + .thenReturn(Prompt.fromText("match the schema")); + + ReActAgent.startAction(new InputEvent("review the diff"), ctx); + + ChatRequestEvent request = soleRequest(ctx); + assertThat(request.getMessages()).hasSize(2); + assertThat(request.getMessages().get(0).getRole()).isEqualTo(MessageRole.SYSTEM); + assertThat(request.getMessages().get(0).getText()).isEqualTo("match the schema"); + assertThat(request.getMessages().get(1).getText()).isEqualTo("review the diff"); + assertThat(request.getOutputSchema()).isEqualTo(WithCount.class); + } + + @Test + @DisplayName("A declared prompt that cannot be resolved is reported, not dropped") + public void testDeclaredPromptThatCannotBeResolvedIsReported() throws Exception { + RunnerContext ctx = mock(RunnerContext.class); + when(ctx.getActionConfigValue("has_user_prompt")).thenReturn(true); + when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT)) + .thenThrow(new IllegalArgumentException("Resource not found")); + + assertThatThrownBy(() -> ReActAgent.startAction(new InputEvent("review the diff"), ctx)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("_default_user_prompt") + .hasCauseInstanceOf(IllegalArgumentException.class); + } + + // --- single-input map --- + + @Test + @DisplayName("A single-input map without a prompt passes the value through") + public void testSingleInputMapWithoutPromptBecomesUserMessage() { + RunnerContext ctx = mock(RunnerContext.class); + + ReActAgent.startAction(new InputEvent(Map.of("input", "analyze this review")), ctx); + + assertThat(soleUserMessage(ctx).getText()).isEqualTo("analyze this review"); + } + + @Test + @DisplayName("A multi-key map without a prompt still reports that a prompt is required") + public void testMultiKeyMapWithoutPromptStillReportsPromptRequirement() { + RunnerContext ctx = mock(RunnerContext.class); + + assertThatThrownBy( + () -> + ReActAgent.startAction( + new InputEvent(Map.of("input", "x", "other", "y")), ctx)) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("not primitive types"); + } + + private static ReActAgent bareAgent() { + return new ReActAgent(chatModelDescriptor(), null, null); + } + + private static ResourceDescriptor chatModelDescriptor() { + return ResourceDescriptor.Builder.newBuilder("com.example.ChatModel").build(); + } + private static RunnerContext contextWithPrompt(Prompt prompt) throws Exception { RunnerContext ctx = mock(RunnerContext.class); + when(ctx.getActionConfigValue("has_user_prompt")).thenReturn(true); when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT)).thenReturn(prompt); return ctx; } - private static ChatMessage soleUserMessage(RunnerContext ctx) { + private static ChatRequestEvent soleRequest(RunnerContext ctx) { ArgumentCaptor captor = ArgumentCaptor.forClass(Event.class); verify(ctx).sendEvent(captor.capture()); - ChatRequestEvent request = (ChatRequestEvent) captor.getValue(); + return (ChatRequestEvent) captor.getValue(); + } + + private static ChatMessage soleUserMessage(RunnerContext ctx) { + ChatRequestEvent request = soleRequest(ctx); assertThat(request.getMessages()).hasSize(1); ChatMessage message = request.getMessages().get(0); assertThat(message.getRole()).isEqualTo(MessageRole.USER); diff --git a/python/flink_agents/api/agents/react_agent.py b/python/flink_agents/api/agents/react_agent.py index d1661a873..c7052a949 100644 --- a/python/flink_agents/api/agents/react_agent.py +++ b/python/flink_agents/api/agents/react_agent.py @@ -52,6 +52,12 @@ _DEFAULT_SCHEMA_PROMPT = "_default_schema_prompt" _DEFAULT_NATIVE_SCHEMA_PROMPT = "_default_native_schema_prompt" _DEFAULT_USER_PROMPT = "_default_user_prompt" +# Action config flag recording that the agent registered _DEFAULT_USER_PROMPT, +# as output_schema does for _DEFAULT_SCHEMA_PROMPT. start_action looks up a +# built-in prompt only when the running scope's own action config declares it, +# keeping a sub-agent's prompts its own on a runtime whose child scope falls +# back to the root agent's resources. +_HAS_USER_PROMPT = "has_user_prompt" _OUTPUT_SCHEMA = "_output_schema" _SKILLS_ARGUMENT = "skills" _SKILLS_SOURCES_ARGUMENT = "skills_sources" @@ -252,6 +258,7 @@ def __init__( output_schema=OutputSchema(output_schema=output_schema) if output_schema else None, + has_user_prompt=bool(prompt), ) @staticmethod @@ -261,12 +268,11 @@ def start_action(event: Event, ctx: RunnerContext) -> None: # at the JSON boundary), so the dispatched InputEvent is used directly. usr_input = cast("InputEvent", event).input - try: - prompt = cast( - "Prompt", ctx.get_resource(_DEFAULT_USER_PROMPT, ResourceType.PROMPT) - ) - except KeyError: - prompt = None + prompt = ( + _declared_prompt(ctx, _DEFAULT_USER_PROMPT) + if ctx.get_action_config_value(key=_HAS_USER_PROMPT) + else None + ) if isinstance(usr_input, bool | str | int | float | type(None)): usr_input = str(usr_input) @@ -278,26 +284,36 @@ def start_action(event: Event, ctx: RunnerContext) -> None: usr_msgs = [ChatMessage.user(usr_input)] else: if not prompt: - err_msg = ( - f"Input type is {usr_input.__class__}, which is not primitive types. " - f"User should provide prompt to help convert it to ChatMessage." + if isinstance(usr_input, dict) and set(usr_input.keys()) == {"input"}: + # The one structured shape that works without a user-supplied + # prompt: a single "input" entry, the arguments of a model-driven + # call against an input schema declaring only that field. The + # value becomes the user message directly, mirroring the + # prompt-less primitive branch. + usr_msgs = [ChatMessage.user(str(usr_input["input"]))] + else: + err_msg = ( + f"Input type is {usr_input.__class__}, which is not primitive types. " + f"User should provide prompt to help convert it to ChatMessage." + ) + raise RuntimeError(err_msg) + else: + if isinstance(usr_input, Row): + usr_input = usr_input.as_dict(recursive=True) + elif isinstance(usr_input, dict): + pass + else: # regard as pojo + usr_input = usr_input.__dict__ + # Convert Any values to str to match format_messages signature + str_usr_input = {k: str(v) for k, v in usr_input.items()} + usr_msgs = prompt.format_messages( + role=MessageRole.USER, **str_usr_input ) - raise RuntimeError(err_msg) - if isinstance(usr_input, Row): - usr_input = usr_input.as_dict(recursive=True) - elif isinstance(usr_input, dict): - pass - else: # regard as pojo - usr_input = usr_input.__dict__ - # Convert Any values to str to match format_messages signature - str_usr_input = {k: str(v) for k, v in usr_input.items()} - usr_msgs = prompt.format_messages(role=MessageRole.USER, **str_usr_input) - - schema_prompt = ReActAgent._prompt_or_none(ctx, _DEFAULT_SCHEMA_PROMPT) output_schema = ctx.get_action_config_value(key="output_schema") - if schema_prompt is not None: + if output_schema is not None: + schema_prompt = _declared_prompt(ctx, _DEFAULT_SCHEMA_PROMPT) # A plan built before the native prompt existed has none, and then adds # no instruction on the native path. instruction = ( @@ -364,3 +380,20 @@ def stop_action(event: Event, ctx: RunnerContext) -> None: output = response.text ctx.send_event(OutputEvent(output=output)) + + +def _declared_prompt(ctx: RunnerContext, name: str) -> Prompt: + """Resolve a built-in prompt the running scope's action config declares. + + The constructor registers the prompt together with its declaration, so a + failed lookup is a broken plan and is reported rather than silently dropping + the prompt. + """ + try: + return cast("Prompt", ctx.get_resource(name, ResourceType.PROMPT)) + except KeyError as e: + err_msg = ( + f"Prompt {name} is declared by this agent's action config but cannot " + f"be resolved." + ) + raise RuntimeError(err_msg) from e diff --git a/python/flink_agents/api/agents/tests/test_react_agent.py b/python/flink_agents/api/agents/tests/test_react_agent.py index 3435a47ba..0fc71a444 100644 --- a/python/flink_agents/api/agents/tests/test_react_agent.py +++ b/python/flink_agents/api/agents/tests/test_react_agent.py @@ -16,6 +16,7 @@ # limitations under the License. ################################################################################# from typing import Any, Callable +from unittest.mock import MagicMock import pytest from pydantic import BaseModel, Field @@ -27,6 +28,9 @@ _DEFAULT_SCHEMA_PROMPT, ReActAgent, ) +from flink_agents.api.chat_message import MessageRole +from flink_agents.api.events.event import InputEvent +from flink_agents.api.prompts.prompt import Prompt from flink_agents.api.resource import ( JavaResourceDescriptor, ResourceDescriptor, @@ -94,6 +98,122 @@ def _expected_prompt(rendered: Any) -> str: return f"The final response should be json format, and match the schema {rendered}." +def _bare_agent() -> ReActAgent: + """A ReActAgent with neither a prompt nor an output schema.""" + return ReActAgent( + chat_model=ResourceDescriptor(clazz=_CHAT_MODEL_CLASS, model="qwen3:8b") + ) + + +def _context_without_prompt() -> tuple[MagicMock, list]: + """A mock context owning no prompt resource, collecting sent events.""" + ctx = MagicMock() + ctx.get_resource = MagicMock(side_effect=KeyError("no such resource")) + ctx.get_action_config_value = MagicMock(return_value=None) + sent_events: list = [] + ctx.send_event = MagicMock(side_effect=lambda e: sent_events.append(e)) + return ctx, sent_events + + +def _sole_user_text(sent_events: list) -> str: + """The text of the single user message of the emitted chat request.""" + assert len(sent_events) == 1 + messages = sent_events[0].messages + assert len(messages) == 1 + assert messages[0].role == MessageRole.USER + return messages[0].text + + +def _start_action_config(agent: ReActAgent) -> dict: + """The action config the agent registered for its start action.""" + return agent._actions["start_action"][2] + + +# --- built-in prompt scoping --- + + +def test_configured_agent_declares_built_in_prompts_in_action_config() -> None: + """A configured agent declares its built-in prompts in the start action config.""" + agent = ReActAgent( + chat_model=ResourceDescriptor(clazz=_CHAT_MODEL_CLASS, model="qwen3:8b"), + prompt=Prompt.from_text("Task: {input}"), + output_schema=Person, + ) + + config = _start_action_config(agent) + assert config["has_user_prompt"] is True + assert config["output_schema"] is not None + + +def test_bare_agent_declares_no_built_in_prompt() -> None: + """A bare agent declares no built-in prompt in the start action config.""" + config = _start_action_config(_bare_agent()) + + assert config["has_user_prompt"] is False + assert config["output_schema"] is None + + +def test_undeclared_prompts_are_not_looked_up() -> None: + """A prompt reachable but not declared by this agent is not looked up.""" + ctx, sent_events = _context_without_prompt() + # A runtime whose child scope falls back to the root agent's resources makes a + # parent ReActAgent's prompts reachable under the same names. + ctx.get_resource = MagicMock(return_value=Prompt.from_text("PARENT ONLY: {input}")) + + ReActAgent.start_action(InputEvent(input="review the diff"), ctx) + + assert _sole_user_text(sent_events) == "review the diff" + ctx.get_resource.assert_not_called() + + +def test_declared_output_schema_inserts_schema_instruction() -> None: + """A declared output schema inserts the schema instruction before the user message.""" + ctx, sent_events = _context_without_prompt() + output_schema = _start_action_config(_agent(Person))["output_schema"] + ctx.get_action_config_value = MagicMock( + side_effect=lambda key: output_schema if key == "output_schema" else None + ) + ctx.get_resource = MagicMock(return_value=Prompt.from_text("match the schema")) + + ReActAgent.start_action(InputEvent(input="review the diff"), ctx) + + messages = sent_events[0].messages + assert [m.text for m in messages] == ["match the schema", "review the diff"] + assert messages[1].role == MessageRole.USER + ctx.get_resource.assert_any_call(_DEFAULT_SCHEMA_PROMPT, ResourceType.PROMPT) + + +def test_declared_prompt_that_cannot_be_resolved_is_reported() -> None: + """A declared prompt that cannot be resolved is reported, not dropped.""" + ctx, _ = _context_without_prompt() + ctx.get_action_config_value = MagicMock( + side_effect=lambda key: key == "has_user_prompt" or None + ) + + with pytest.raises(RuntimeError, match="_default_user_prompt"): + ReActAgent.start_action(InputEvent(input="review the diff"), ctx) + + +# --- start_action input dispatch --- + + +def test_single_input_map_without_prompt_becomes_user_message() -> None: + """A single-input map without a prompt passes the value through.""" + ctx, sent_events = _context_without_prompt() + + ReActAgent.start_action(InputEvent(input={"input": "analyze this review"}), ctx) + + assert _sole_user_text(sent_events) == "analyze this review" + + +def test_multi_key_map_without_prompt_still_reports_prompt_requirement() -> None: + """A multi-key map without a prompt still reports that a prompt is required.""" + ctx, _ = _context_without_prompt() + + with pytest.raises(RuntimeError, match="not primitive types"): + ReActAgent.start_action(InputEvent(input={"input": "x", "other": "y"}), ctx) + + def test_unrenderable_output_schema_raises_naming_the_model() -> None: """A schema that cannot be rendered fails at construction, not at the provider.""" with pytest.raises(TypeError, match="Unrenderable cannot be rendered"): From 96d74849df757c64903b6aa1e65f72550706b5cd Mon Sep 17 00:00:00 2001 From: Yi Zhang Date: Tue, 13 Oct 2026 18:51:52 +0800 Subject: [PATCH 3/4] [runtime][python] Add e2e coverage for ReActAgent sub-agents The internal sub-agent e2e suites only exercised hand-written echo agents as children. Cover the built-in ReActAgent itself so its start action, the framework's chat model action and its stop action all run inside the child scope, driven by scripted connections instead of a live model. Both delegation paths are covered on each side: a caller-driven path where a root action submits a string prompt and awaits the future, and a model-driven path where the parent model issues a _subagent_child tool call whose {"input": ...} arguments flow through the prompt-free map path; the child's reply returns as the tool observation that closes the loop. The caller-driven Python scenario also drives a child tool loop, and the Java harness covers the parent prompt leaks. The new Python caller-driven test also caught a latent scope bug: FlinkRunnerContext resolved action_config and get_action_config_value against the root plan even inside a child scope, so the child's ReActAgent.start_action crashed reading its own output_schema config. Mirror the Java context's scope-aware currentPlan() by resolving both against the active scope's plan. Generated-by: Qoder --- .../react_agent_subagent_test.py | 344 +++++++++++++ .../runtime/flink_runner_context.py | 18 +- .../subagent/ReActAgentSubagentTest.java | 458 ++++++++++++++++++ 3 files changed, 818 insertions(+), 2 deletions(-) create mode 100644 python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py create mode 100644 runtime/src/test/java/org/apache/flink/agents/runtime/subagent/ReActAgentSubagentTest.java diff --git a/python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py b/python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py new file mode 100644 index 000000000..c7e8681ed --- /dev/null +++ b/python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py @@ -0,0 +1,344 @@ +################################################################################ +# Licensed to the Apache Software Foundation (ASF) under one +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to you under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +################################################################################ +"""E2E tests for a real ReActAgent running as an internal sub-agent. + +The children here are the built-in agent itself, not hand-written echo agents: +its start action, the framework's chat model action and its stop action must +all run inside the child scope, driven by scripted connections instead of a +live model. The model-driven test additionally runs the whole delegation loop: +the parent model issues a ``_subagent_child`` tool call, the framework routes +it to the child, and the child's answer flows back as the tool observation +that closes the loop. + +The delegated child is registered with a single-field ``input`` schema, so +that test also proves the declared schema survives compilation and the +child's prompt-free ``{input}`` map path accepts the model's arguments as-is. + +The third test lets the child declare a tool of its own and scripts two model +rounds for it: the first reply asks for the tool, the tool executes inside +the child scope, and the second round sees the tool result, closing the +reasoning/tool loop in isolation. +""" + +import os +import sysconfig +from pathlib import Path +from typing import Any + +from pyflink.common import Encoder +from pyflink.common.typeinfo import Types +from pyflink.datastream import KeySelector, StreamExecutionEnvironment +from pyflink.datastream.connectors.file_system import StreamingFileSink + +from flink_agents.api.agents.agent import Agent +from flink_agents.api.agents.react_agent import ReActAgent +from flink_agents.api.chat_message import ( + ChatMessage, + MessageRole, + TextBlock, + ToolCallBlock, + ToolResultBlock, +) +from flink_agents.api.chat_models.chat_model import ( + BaseChatModelConnection, + BaseChatModelSetup, +) +from flink_agents.api.chat_result import ChatResult +from flink_agents.api.decorators import action +from flink_agents.api.events.event import Event, InputEvent, OutputEvent +from flink_agents.api.execution_environment import AgentsExecutionEnvironment +from flink_agents.api.resource import ResourceDescriptor, ResourceType +from flink_agents.api.runner_context import RunnerContext +from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder +from flink_agents.api.tools.tool import Tool + +os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"] + +CHILD_SCOPE = "child" + + +class InputKeySelector(KeySelector): + """Keys every element by itself.""" + + def get_key(self, value: Any) -> Any: + """Return the element itself as key.""" + return value + + +class ScriptedModelSetup(BaseChatModelSetup): + """A setup whose behavior lives entirely in its scripted connection.""" + + @property + def model_kwargs(self) -> dict[str, Any]: + """Return chat model settings.""" + return {} + + +class ParentScriptedConnection(BaseChatModelConnection): + """Delegates once, then answers with the observed child output.""" + + def chat( + self, + messages: Any, + tools: list | None = None, + output_schema: Any = None, + **kwargs: Any, + ) -> ChatResult: + """Issue the delegation tool call, then echo the observation back.""" + self._reject_unsupported_output_schema(output_schema) + tool_messages = [m for m in messages if m.role == MessageRole.TOOL] + if not tool_messages: + # The child is declared with a single "input" field, so the model + # builds the one argument the agent's prompt-free path accepts. + tool_call = ToolCallBlock( + call_id="call-1", + name="_subagent_child", + input={"input": "review the diff"}, + ) + return ChatResult( + message=ChatMessage.assistant( + [TextBlock(text="delegating"), tool_call] + ) + ) + result = tool_messages[0].blocks[0] + assert isinstance(result, ToolResultBlock) + return ChatResult( + message=ChatMessage.assistant(f"observed: {result.text}") + ) + + +class ChildScriptedConnection(BaseChatModelConnection): + """Echoes the last user message, revealing what the child agent received.""" + + def chat( + self, + messages: Any, + tools: list | None = None, + output_schema: Any = None, + **kwargs: Any, + ) -> ChatResult: + """Answer with the child's own user message.""" + self._reject_unsupported_output_schema(output_schema) + return ChatResult( + message=ChatMessage.assistant(f"child saw: {messages[-1].text}") + ) + + +def shout(text: str) -> str: + """Return the text in upper case, marked as tool-produced.""" + return f"{text.upper()}!" + + +class ToolCallingChildConnection(BaseChatModelConnection): + """Asks for the child's tool first, then answers from the tool result.""" + + def chat( + self, + messages: Any, + tools: list | None = None, + output_schema: Any = None, + **kwargs: Any, + ) -> ChatResult: + """Emit the tool call on the first round, consume the result on the second.""" + self._reject_unsupported_output_schema(output_schema) + tool_messages = [m for m in messages if m.role == MessageRole.TOOL] + if not tool_messages: + tool_call = ToolCallBlock( + call_id="call-1", name="shout", input={"text": "hello world"} + ) + return ChatResult( + message=ChatMessage.assistant( + [TextBlock(text="let me check"), tool_call] + ) + ) + result = tool_messages[0].blocks[0] + assert isinstance(result, ToolResultBlock) + return ChatResult( + message=ChatMessage.assistant(f"tool says: {result.text}") + ) + + +def _child_react_agent() -> ReActAgent: + """A ReActAgent with neither a prompt nor an output schema. + + The connection is registered on the child itself, keeping the child plan + self-contained: a child scope resolves resources against the child plan + only, so the connection must live there. + """ + child = ReActAgent( + chat_model=ResourceDescriptor( + clazz=f"{ScriptedModelSetup.__module__}.{ScriptedModelSetup.__name__}", + connection="child_connection", + model="mock-model", + ), + ) + child.add_resource( + "child_connection", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor( + clazz=f"{ChildScriptedConnection.__module__}.{ChildScriptedConnection.__name__}", + ), + ) + return child + + +def _tool_calling_child_react_agent() -> ReActAgent: + """A ReActAgent whose model asks for the child's own tool across two rounds.""" + child = ReActAgent( + chat_model=ResourceDescriptor( + clazz=f"{ScriptedModelSetup.__module__}.{ScriptedModelSetup.__name__}", + connection="tool_child_connection", + model="mock-model", + tools=["shout"], + ), + ) + child.add_resource( + "tool_child_connection", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor( + clazz=f"{ToolCallingChildConnection.__module__}.{ToolCallingChildConnection.__name__}", + ), + ) + child.add_resource("shout", ResourceType.TOOL, Tool.from_callable(shout)) + return child + + +class CallerRootAgent(Agent): + """Caller agent awaiting the ReAct child from an async action.""" + + @action(InputEvent.EVENT_TYPE) + @staticmethod + async def orchestrate(event: Event, ctx: RunnerContext) -> None: + """Await the child and emit its outcome.""" + child = ctx.get_resource(CHILD_SCOPE, 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: + msg = f"failed:{result.error_message}" + ctx.send_event(OutputEvent(output=msg)) + + +def _delegating_parent() -> ReActAgent: + """A bare ReActAgent parent that declares the child to its model. + + Sub-classing ReActAgent would trip the plan compiler, which rejects + inherited ``@action`` methods on a concrete agent, so the parent is a + plain instance carrying instance-level resources instead. + """ + parent = ReActAgent( + chat_model=ResourceDescriptor( + clazz=f"{ScriptedModelSetup.__module__}.{ScriptedModelSetup.__name__}", + connection="parent_connection", + model="mock-model", + subagents=[CHILD_SCOPE], + ), + ) + parent.add_resource( + "parent_connection", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor( + clazz=f"{ParentScriptedConnection.__module__}.{ParentScriptedConnection.__name__}", + ), + ) + # Only a declared input schema lets the parent model call the child. + parent.add_resource( + CHILD_SCOPE, + ResourceType.AGENT, + InternalSubagentResourceDescriptorBuilder() + .agent(_child_react_agent()) + .description("Reviews a code diff.") + .input_schema( + '{"type":"object","properties":{"input":{"type":"string"}},' + '"required":["input"]}' + ) + .build(), + ) + return parent + + +def _run_and_collect(root: Agent, tmp_path: Path) -> str: + """Run the root agent over one element and collect what the job emitted.""" + env = StreamExecutionEnvironment.get_execution_environment() + env.set_parallelism(1) + input_stream = env.from_collection(["hello"]) + + agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env) + output_datastream = ( + agents_env.from_datastream(input=input_stream, key_selector=InputKeySelector()) + .apply(root) + .to_datastream() + ) + + result_dir = tmp_path / "results" + result_dir.mkdir(parents=True, exist_ok=True) + output_datastream.map(str, Types.STRING()).add_sink( + StreamingFileSink.for_row_format( + base_path=str(result_dir.absolute()), + encoder=Encoder.simple_string_encoder(), + ).build() + ) + agents_env.execute() + return "".join(p.read_text() for p in result_dir.rglob("*") if p.is_file()) + + +def test_caller_driven_react_subagent(tmp_path: Path) -> None: + """A submitted string runs the child's ReAct loop and returns its answer.""" + root = CallerRootAgent() + root.add_resource(CHILD_SCOPE, ResourceType.AGENT, _child_react_agent()) + + contents = _run_and_collect(root, tmp_path) + + # The child connection saw the submitted string as its user message, which is + # the prompt-free primitive path, and the child's reply is the call result. + assert "child saw: review the diff" in contents, ( + f"missing child output; collected: {contents!r}" + ) + + +def test_model_driven_react_subagent_delegation(tmp_path: Path) -> None: + """A model-issued delegation reaches the declared child and closes the loop.""" + contents = _run_and_collect(_delegating_parent(), tmp_path) + + # The final answer echoes the tool observation, proving the whole loop: the + # model's {"input": ...} arguments became the child's user message through + # the prompt-free {input} map path, and the child's reply came back as the + # observation. A child's own output never reaches the sink directly -- only + # this parent answer does -- so both fragments here mean the round-trip + # completed; the assertion stays split because the observation carries the + # result in the framework's string form, not verbatim. + assert "observed:" in contents, f"no observation echoed; collected: {contents!r}" + assert "child saw: review the diff" in contents, ( + f"child output missing from the loop; collected: {contents!r}" + ) + + +def test_child_tool_call_loop_runs_inside_the_child_scope(tmp_path: Path) -> None: + """The child's own tool executes inside the child scope across two rounds.""" + root = CallerRootAgent() + root.add_resource( + CHILD_SCOPE, ResourceType.AGENT, _tool_calling_child_react_agent() + ) + + contents = _run_and_collect(root, tmp_path) + + # The tool result flowed back into the child's second model round, which + # answered from it; the child's tool never touches the parent's resources. + assert "tool says: HELLO WORLD!" in contents, ( + f"tool loop output missing; collected: {contents!r}" + ) diff --git a/python/flink_agents/runtime/flink_runner_context.py b/python/flink_agents/runtime/flink_runner_context.py index 23e0654c8..21485641d 100644 --- a/python/flink_agents/runtime/flink_runner_context.py +++ b/python/flink_agents/runtime/flink_runner_context.py @@ -767,6 +767,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() + if plan_json is None: + return self.__agent_plan + return self.__scoped_plan_and_cache(plan_json)[0] + def __observe_subagent_setup(self, resource: Any) -> None: """Wire a lazily materialized sub-agent handle into the task lifecycle. @@ -902,14 +916,14 @@ def notify_record_finished(self, key: Any) -> None: @override def action_config(self) -> Dict[str, Any]: """Get config of the action.""" - return self.__agent_plan.get_action_config( + return self.__active_plan().get_action_config( self._j_runner_context.getActionName() ) @override def get_action_config_value(self, key: str) -> Any: """Get config of the action.""" - return self.__agent_plan.get_action_config_value( + return self.__active_plan().get_action_config_value( action_name=self._j_runner_context.getActionName(), key=key ) diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/ReActAgentSubagentTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/ReActAgentSubagentTest.java new file mode 100644 index 000000000..f27e0680f --- /dev/null +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/ReActAgentSubagentTest.java @@ -0,0 +1,458 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.agents.runtime.subagent; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.InputEvent; +import org.apache.flink.agents.api.OutputEvent; +import org.apache.flink.agents.api.agents.Agent; +import org.apache.flink.agents.api.agents.ReActAgent; +import org.apache.flink.agents.api.annotation.ToolParam; +import org.apache.flink.agents.api.chat.messages.ChatMessage; +import org.apache.flink.agents.api.chat.messages.ChatResult; +import org.apache.flink.agents.api.chat.messages.MessageRole; +import org.apache.flink.agents.api.chat.messages.TextBlock; +import org.apache.flink.agents.api.chat.messages.ToolCallBlock; +import org.apache.flink.agents.api.chat.messages.ToolResultBlock; +import org.apache.flink.agents.api.chat.model.BaseChatModelConnection; +import org.apache.flink.agents.api.chat.model.BaseChatModelSetup; +import org.apache.flink.agents.api.context.RunnerContext; +import org.apache.flink.agents.api.prompt.Prompt; +import org.apache.flink.agents.api.resource.ResourceContext; +import org.apache.flink.agents.api.resource.ResourceDescriptor; +import org.apache.flink.agents.api.resource.ResourceType; +import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder; +import org.apache.flink.agents.api.subagent.SubagentResult; +import org.apache.flink.agents.api.subagent.SubagentSetup; +import org.apache.flink.agents.api.tools.Tool; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.runtime.operator.ActionExecutionOperator; +import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; + +import javax.annotation.Nullable; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * End-to-end coverage for a real {@link ReActAgent} running as an internal sub-agent, over the + * operator harness. Unlike {@link InternalSubagentCallTest}, whose children are hand-written echo + * agents, the child here is the built-in agent itself: its start action, the framework's chat model + * action and its stop action must all run inside the child scope, driven by a scripted connection + * instead of a live model. + * + *

The second test drives the model path: the parent model issues a {@code _subagent_child} tool + * call, the framework routes it to the child, and the child's answer flows back as the tool + * observation that closes the loop. The child is registered with a single-field {@code input} + * schema, so this also proves the declared schema survives compilation and the child's prompt-free + * {@code {input}} map path accepts the model's arguments as-is. + * + *

The third test lets the child declare a tool of its own and scripts two model rounds for it: + * the first reply asks for the tool, the tool executes inside the child scope, and the second round + * sees the tool result, closing the reasoning/tool loop in isolation. + * + *

The last two tests configure the parent ReActAgent with a prompt and an output schema and + * leave the child bare. A child scope's resource lookup falls back to the root agent's resources, + * so they pin that the child's built-in prompts stay its own rather than resolving to the parent's + * under the same names. + * + *

Requires JDK 21+ for the same mailbox-suspension reason as {@link InternalSubagentCallTest}. + */ +@EnabledForJreRange(min = JRE.JAVA_21) +public class ReActAgentSubagentTest { + + @BeforeEach + void resetScripts() { + ParentScriptedConnection.reset(); + ChildScriptedConnection.reset(); + } + + // --- scripted chat model --- + + /** Materialized from a descriptor; the connection carries the script. */ + public static class ScriptedModelSetup extends BaseChatModelSetup { + + public ScriptedModelSetup(ResourceDescriptor descriptor, ResourceContext resourceContext) { + super(descriptor, resourceContext); + } + + @Override + public Map getParameters() { + return new HashMap<>(); + } + } + + /** Serves the parent agent; replies come from a static script. */ + public static class ParentScriptedConnection extends BaseChatModelConnection { + + static final Deque REPLIES = new ArrayDeque<>(); + static final List> CONVERSATIONS = new ArrayList<>(); + + public ParentScriptedConnection( + ResourceDescriptor descriptor, ResourceContext resourceContext) { + super(descriptor, resourceContext); + } + + static void reset() { + REPLIES.clear(); + CONVERSATIONS.clear(); + } + + @Override + public ChatResult chat( + List messages, List tools, Map modelParams) { + CONVERSATIONS.add(new ArrayList<>(messages)); + return new ChatResult( + REPLIES.isEmpty() ? ChatMessage.assistant("no script left") : REPLIES.poll()); + } + } + + /** Serves the child agent; replies come from a static script. */ + public static class ChildScriptedConnection extends BaseChatModelConnection { + + static final Deque REPLIES = new ArrayDeque<>(); + static final List> CONVERSATIONS = new ArrayList<>(); + static final List> OFFERED_TOOLS = new ArrayList<>(); + + public ChildScriptedConnection( + ResourceDescriptor descriptor, ResourceContext resourceContext) { + super(descriptor, resourceContext); + } + + static void reset() { + REPLIES.clear(); + CONVERSATIONS.clear(); + OFFERED_TOOLS.clear(); + } + + @Override + public ChatResult chat( + List messages, List tools, Map modelParams) { + CONVERSATIONS.add(new ArrayList<>(messages)); + OFFERED_TOOLS.add( + tools == null + ? List.of() + : tools.stream().map(Tool::getName).collect(Collectors.toList())); + return new ChatResult( + REPLIES.isEmpty() ? ChatMessage.assistant("no script left") : REPLIES.poll()); + } + } + + // --- caller actions --- + + @SuppressWarnings("unused") + public static void callReActChild(Event event, RunnerContext ctx) throws Exception { + SubagentSetup child = (SubagentSetup) ctx.getResource("child", ResourceType.AGENT); + SubagentResult result = child.submit(ctx, "review the diff").await(); + ctx.sendEvent( + new OutputEvent( + result.isSuccess() + ? firstOutput(result) + : "unexpected: " + result.getErrorMessage())); + } + + // --- tests --- + + @Test + @Timeout(60) + void callerDrivenCallRunsTheReActLoopInsideTheChildScope() throws Exception { + ChildScriptedConnection.REPLIES.add(new ChatMessage(MessageRole.ASSISTANT, "reviewed: ok")); + + Agent parent = new Agent(); + registerConnections(parent); + parent.addResource("child", ResourceType.AGENT, bareReactAgent("child_conn")); + parent.addAction( + new String[] {InputEvent.EVENT_TYPE}, + ReActAgentSubagentTest.class.getMethod( + "callReActChild", Event.class, RunnerContext.class)); + + List output = run(parent, 7L); + + // The child's stop action output is the caller-visible result, and the submitted string + // became the child's user message through the prompt-free primitive path. + assertThat(output).containsExactly("reviewed: ok"); + assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1); + assertThat(latestUserText(ChildScriptedConnection.CONVERSATIONS.get(0))) + .isEqualTo("review the diff"); + } + + @Test + @Timeout(60) + void modelDrivenDelegationRoutesToTheDeclaredChildAndClosesTheLoop() throws Exception { + ToolCallBlock delegation = delegation(); + ParentScriptedConnection.REPLIES.add(assistantWithToolCall("delegating", delegation)); + ParentScriptedConnection.REPLIES.add( + new ChatMessage(MessageRole.ASSISTANT, "delegated and reviewed")); + ChildScriptedConnection.REPLIES.add(new ChatMessage(MessageRole.ASSISTANT, "reviewed: ok")); + + // The registration declares the child's schema, which carries it through open(). + ReActAgent parent = + new ReActAgent( + ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName()) + .addInitialArgument("connection", "parent_conn") + .addInitialArgument("subagents", List.of("child")) + .build(), + null, + null); + registerConnections(parent); + parent.addResource( + "child", ResourceType.AGENT, declaredChild(bareReactAgent("child_conn"))); + + List output = run(parent, 7L); + + assertThat(output).containsExactly("delegated and reviewed"); + // The model's arguments reached the child as its user message through the prompt-free + // {input} map path, and the child's answer came back as the closing observation. + assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1); + assertThat(latestUserText(ChildScriptedConnection.CONVERSATIONS.get(0))) + .isEqualTo("review the diff"); + assertThat(ParentScriptedConnection.CONVERSATIONS).hasSize(2); + assertThat(allTexts(ParentScriptedConnection.CONVERSATIONS.get(1))) + .anySatisfy(text -> assertThat(text).contains("reviewed: ok")); + } + + @Test + @Timeout(60) + void toolCallLoopRunsInsideTheChildScope() throws Exception { + ToolCallBlock lookup = new ToolCallBlock("call-1", "shout", Map.of("text", "hello world")); + ChildScriptedConnection.REPLIES.add(assistantWithToolCall("let me check", lookup)); + ChildScriptedConnection.REPLIES.add( + new ChatMessage(MessageRole.ASSISTANT, "tool says: HELLO WORLD!")); + + Agent parent = new Agent(); + registerConnections(parent); + parent.addResource("child", ResourceType.AGENT, toolCallingReactAgent()); + parent.addAction( + new String[] {InputEvent.EVENT_TYPE}, + ReActAgentSubagentTest.class.getMethod( + "callReActChild", Event.class, RunnerContext.class)); + + List output = run(parent, 7L); + + // The tool schema was offered on the first round, the tool executed inside the child + // scope, and the second round saw the tool result before answering. + assertThat(output).containsExactly("tool says: HELLO WORLD!"); + assertThat(ChildScriptedConnection.OFFERED_TOOLS.get(0)).containsExactly("shout"); + assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(2); + assertThat(ChildScriptedConnection.CONVERSATIONS.get(1)) + .anySatisfy( + message -> { + assertThat(message.getRole()).isEqualTo(MessageRole.TOOL); + ToolResultBlock result = (ToolResultBlock) message.getBlocks().get(0); + assertThat(result.getText()).isEqualTo("HELLO WORLD!"); + }); + } + + @Test + @Timeout(60) + void bareChildDoesNotInheritTheParentPrompt() throws Exception { + ParentScriptedConnection.REPLIES.add(assistantWithToolCall("delegating", delegation())); + ParentScriptedConnection.REPLIES.add( + new ChatMessage(MessageRole.ASSISTANT, "delegated and reviewed")); + ChildScriptedConnection.REPLIES.add(new ChatMessage(MessageRole.ASSISTANT, "reviewed: ok")); + + ReActAgent parent = + delegatingParent(Prompt.fromText("PARENT ONLY: {input}"), /* outputSchema= */ null); + + List output = run(parent, 7L); + + assertThat(output).containsExactly("delegated and reviewed"); + // The parent prompt applies to the parent's own input only; the bare child takes the + // delegated value through its prompt-free {input} map path. + assertThat(latestUserText(ParentScriptedConnection.CONVERSATIONS.get(0))) + .isEqualTo("PARENT ONLY: 7"); + assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1); + assertThat(latestUserText(ChildScriptedConnection.CONVERSATIONS.get(0))) + .isEqualTo("review the diff"); + } + + @Test + @Timeout(60) + void bareChildDoesNotInheritTheParentOutputSchema() throws Exception { + ParentScriptedConnection.REPLIES.add(assistantWithToolCall("delegating", delegation())); + ParentScriptedConnection.REPLIES.add( + new ChatMessage(MessageRole.ASSISTANT, "{\"verdict\":\"ok\"}")); + ChildScriptedConnection.REPLIES.add(new ChatMessage(MessageRole.ASSISTANT, "reviewed: ok")); + + ReActAgent parent = delegatingParent(/* prompt= */ null, Verdict.class); + + run(parent, 7L); + + // The parent's own request carries its schema instruction, which proves the instruction + // is reachable by name; the bare child's request must not. + assertThat(allTexts(ParentScriptedConnection.CONVERSATIONS.get(0))) + .anySatisfy(text -> assertThat(text).contains(SCHEMA_INSTRUCTION)); + assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1); + List childConversation = ChildScriptedConnection.CONVERSATIONS.get(0); + assertThat(allTexts(childConversation)) + .noneSatisfy(text -> assertThat(text).contains(SCHEMA_INSTRUCTION)); + assertThat(childConversation) + .extracting(ChatMessage::getRole) + .containsExactly(MessageRole.USER); + } + + // --- helpers --- + + private static final String SCHEMA_INSTRUCTION = "The final response should be json format"; + + /** The parent's structured answer, giving it an output schema the bare child must not see. */ + public static class Verdict { + public String verdict; + } + + private static ToolCallBlock delegation() { + return new ToolCallBlock("call-1", "_subagent_child", Map.of("input", "review the diff")); + } + + private static ChatMessage assistantWithToolCall(String text, ToolCallBlock toolCall) { + return ChatMessage.assistant(List.of(new TextBlock(text), toolCall)); + } + + /** A ReActAgent parent configured as given, declaring a bare ReActAgent child to its model. */ + private static ReActAgent delegatingParent( + @Nullable Prompt prompt, @Nullable Class outputSchema) { + ReActAgent parent = + new ReActAgent( + ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName()) + .addInitialArgument("connection", "parent_conn") + .addInitialArgument("subagents", List.of("child")) + .build(), + prompt, + outputSchema); + registerConnections(parent); + parent.addResource( + "child", ResourceType.AGENT, declaredChild(bareReactAgent("child_conn"))); + return parent; + } + + /** + * Declares {@code child} with a single {@code input} field, the arguments shape the prompt-free + * map path of a ReActAgent passes through as the user message. + */ + private static ResourceDescriptor declaredChild(Agent child) { + return new InternalSubagentResourceDescriptorBuilder() + .agent(child) + .description("Reviews a code diff.") + .inputSchema( + "{\"type\":\"object\",\"properties\":{\"input\":{\"type\":\"string\"}}," + + "\"required\":[\"input\"]}") + .build(); + } + + private static ReActAgent toolCallingReactAgent() throws Exception { + ReActAgent child = + new ReActAgent( + ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName()) + .addInitialArgument("connection", "child_conn") + .addInitialArgument("tools", List.of("shout")) + .build(), + null, + null); + child.addResource( + "shout", + ResourceType.TOOL, + Tool.fromMethod(ReActAgentSubagentTest.class.getMethod("shout", String.class))); + return child; + } + + @SuppressWarnings("unused") + public static String shout(@ToolParam(name = "text") String text) { + return text.toUpperCase(Locale.ROOT) + "!"; + } + + private static ReActAgent bareReactAgent(String connectionName) { + return new ReActAgent( + ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName()) + .addInitialArgument("connection", connectionName) + .build(), + null, + null); + } + + private static String firstOutput(SubagentResult result) { + return String.valueOf(((List) result.getResult()).get(0)); + } + + private static String latestUserText(List messages) { + for (int i = messages.size() - 1; i >= 0; i--) { + ChatMessage message = messages.get(i); + if (message.getRole() == MessageRole.USER) { + return message.getText(); + } + } + throw new AssertionError("no user message in " + messages); + } + + private static List allTexts(List messages) { + return messages.stream() + .flatMap(message -> message.getBlocks().stream()) + .filter(block -> block instanceof TextBlock || block instanceof ToolResultBlock) + .map( + block -> + block instanceof TextBlock + ? ((TextBlock) block).getText() + : ((ToolResultBlock) block).getText()) + .collect(Collectors.toList()); + } + + private static void registerConnections(Agent agent) { + agent.addResource( + "parent_conn", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor.Builder.newBuilder(ParentScriptedConnection.class.getName()) + .build()); + agent.addResource( + "child_conn", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor.Builder.newBuilder(ChildScriptedConnection.class.getName()) + .build()); + } + + @SuppressWarnings("unchecked") + private static List run(Agent root, long input) throws Exception { + try (KeyedOneInputStreamOperatorTestHarness harness = + new KeyedOneInputStreamOperatorTestHarness<>( + new ActionExecutionOperatorFactory<>(new AgentPlan(root), true), + (KeySelector) value -> value, + TypeInformation.of(Long.class))) { + harness.open(); + harness.processElement(new StreamRecord<>(input)); + ((ActionExecutionOperator) harness.getOperator()) + .waitInFlightEventsFinished(); + return ((List>) harness.getRecordOutput()) + .stream().map(StreamRecord::getValue).collect(Collectors.toList()); + } + } +} From f42dda0cacd8b576b99404558c8b490eaf1cbd66 Mon Sep 17 00:00:00 2001 From: Yi Zhang Date: Tue, 13 Oct 2026 18:51:52 +0800 Subject: [PATCH 4/4] [docs] Document delegating to sub-agents and running a ReActAgent as one The sub-agents page covered declaring a child but not calling it, and the ReAct agent page had no recipe for running a ReActAgent as a child. Extend the sub-agents page with delegating from a chat model, calling from an action and the SubagentResult both paths return, and a section on using a ReActAgent as the child: registering it through InternalSubagentResourceDescriptorBuilder and matching the input schema to the child's prompt. The ReAct agent page links there. Generated-by: Qoder --- .../docs/development/multi-agent/subagents.md | 203 ++++++++++++++++++ docs/content/docs/development/react_agent.md | 4 + 2 files changed, 207 insertions(+) diff --git a/docs/content/docs/development/multi-agent/subagents.md b/docs/content/docs/development/multi-agent/subagents.md index 86e610897..0871c6799 100644 --- a/docs/content/docs/development/multi-agent/subagents.md +++ b/docs/content/docs/development/multi-agent/subagents.md @@ -94,6 +94,10 @@ This shorthand does not declare an input schema. Use the builder when exposing the child to a model. Recursive registration, such as a child registering its parent, is rejected when the plan is built. +Register the resources a child uses, such as its chat model connection, on the +child itself. In Python, a child resolves resources against its own plan only, +so a resource registered solely on the parent is not visible to the child. + ## Declare an external sub-agent Implement `SubagentSetup` (or the appropriate runtime base class) and register a @@ -131,3 +135,202 @@ The Python setup declares these configuration fields and implements the invocation contract. Both languages construct the setup at runtime from the descriptor. Existing code that passes a setup instance to `addResource` or `add_resource` must instead pass a descriptor with the same configuration. + +## Delegate from a chat model + +List the child's registered name under `subagents` of the parent's chat model: + +```java +ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP) + .addInitialArgument("connection", "ollamaConnection") + .addInitialArgument("model", "qwen3:8b") + .addInitialArgument("subagents", List.of("reviewer")) + .build(); +``` + +```python +ResourceDescriptor( + clazz=ResourceName.ChatModel.OLLAMA_SETUP, + connection="ollama_connection", + model="qwen3:8b", + subagents=["reviewer"], +) +``` + +The model sees the child as a tool named `_subagent_reviewer`, described by the +registration's description and input schema. The arguments the model sends +become the child's input, and the child's outputs return to the model as the +tool observation, so it can continue reasoning over them. For an internal +sub-agent, the observation is the JSON encoding of the full output list +described in [Call from an action](#call-from-an-action), even when the child +emits a single output, so its shape does not depend on how many outputs the +child emits. + +## Call from an action + +An action can submit input to a child directly and await its outcome. This path +does not need an input schema. + +```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())); + } + } +} +``` + +```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}")) +``` + +The awaited value is a `SubagentResult`. On success, its `result` field +(`getResult()` in Java) carries the child's result. For an internal sub-agent, +such as `reviewer` here, that is the list of outputs the child emitted, which is +why the example takes the first element. An external sub-agent returns whatever +its `SubagentSetup` implementation produces, such as a string or an object. On +failure, `success` is `False` (`isSuccess()` is `false` in Java), and +`error_message` (`getErrorMessage()`) carries the reason. + +## Use a ReActAgent as the child + +A ReActAgent runs as a child like any other agent, keeping its own event loop, +resources and conversation. It uses only its own prompt and output schema, even +when the parent is also a ReActAgent. + +```java +ReActAgent reviewer = + new ReActAgent( + ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP) + .addInitialArgument("connection", "reviewerConnection") + .addInitialArgument("model", "qwen3:8b") + .build(), + null, + null); +reviewer.addResource( + "reviewerConnection", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_CONNECTION) + .build()); + +parent.addResource( + "reviewer", + ResourceType.AGENT, + new InternalSubagentResourceDescriptorBuilder() + .agent(reviewer) + .description("Reviews a code diff and reports problems.") + .inputSchema( + "{\"type\":\"object\",\"properties\":{\"input\":{\"type\":\"string\"}}," + + "\"required\":[\"input\"]}") + .build()); +``` + +```python +reviewer = ReActAgent( + chat_model=ResourceDescriptor( + clazz=ResourceName.ChatModel.OLLAMA_SETUP, + connection="reviewer_connection", + model="qwen3:8b", + ), +) +reviewer.add_resource( + "reviewer_connection", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor(clazz=ResourceName.ChatModel.OLLAMA_CONNECTION), +) + +parent.add_resource( + "reviewer", + ResourceType.AGENT, + InternalSubagentResourceDescriptorBuilder() + .agent(reviewer) + .description("Reviews a code diff and reports problems.") + .input_schema( + '{"type":"object","properties":{"input":{"type":"string"}},' + '"required":["input"]}' + ) + .build(), +) +``` + +Called from an action, the child takes a string input as its user message, or +uses it to fill the `{input}` placeholder of its prompt. Delegated to by a +model, the child receives the arguments as a map keyed by the input schema's +field names and turns that map into its user message, so the schema and the +child's prompt form one contract: + +- **With a prompt**, the schema's field names must match the prompt's + placeholders: each argument fills the placeholder of the same name, as + described in [Prompt]({{< ref "docs/development/react_agent#prompt" >}}). +- **Without a prompt**, declare a single `input` field, as in the example above. + A map holding only `input` passes its value through as the user message; any + other map is rejected because there is no prompt to format it. + +A child with a prompt declares the fields its placeholders use: + +```java +ReActAgent reviewer = + new ReActAgent( + chatModelDescriptor, + Prompt.fromText( + "Review the following code diff and report any problems:\n\n{diff}"), + null); + +parent.addResource( + "reviewer", + ResourceType.AGENT, + new InternalSubagentResourceDescriptorBuilder() + .agent(reviewer) + .description("Reviews a code diff and reports problems.") + // "diff" matches the {diff} placeholder of the child's prompt. + .inputSchema( + "{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}}," + + "\"required\":[\"diff\"]}") + .build()); +``` + +```python +reviewer = ReActAgent( + chat_model=chat_model_descriptor, + prompt=Prompt.from_text( + "Review the following code diff and report any problems:\n\n{diff}" + ), +) + +parent.add_resource( + "reviewer", + 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(), +) +``` + +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 changing either. diff --git a/docs/content/docs/development/react_agent.md b/docs/content/docs/development/react_agent.md index 32ae5518d..ba7b68cad 100644 --- a/docs/content/docs/development/react_agent.md +++ b/docs/content/docs/development/react_agent.md @@ -399,3 +399,7 @@ The constructor registers those sources on the agent, so they do not also need ` **Which declaration wins.** Sources reach the agent from three levels. `@Skills`/`@skills` methods are merged first, then the execution environment's `SKILLS` resources, then the agent's own. A declaration from a later level wins when two of them resolve to the same skill name. +## As a Sub-Agent + +A ReActAgent can run as a sub-agent of another agent, delegated to by the parent's chat model or called from the parent's actions. See [Use a ReActAgent as the child]({{< ref "docs/development/multi-agent/subagents#use-a-reactagent-as-the-child" >}}). +