From 18a4621652ccbda345d6ba75164b6f7769a49702 Mon Sep 17 00:00:00 2001 From: WenjinXie Date: Fri, 16 Oct 2026 11:51:25 +0800 Subject: [PATCH 1/3] [api][java][python] Honor empty text when loading YAML prompts Use message templates when text is empty, matching PromptSpec validation. Cover omitted, null and empty text with message role, order and formatting assertions. Generated-by: Codex CLI 0.153.4 (GPT-6) Co-authored-by: Codex --- .../flink/agents/api/yaml/YamlLoader.java | 2 +- .../api/yaml/YamlLoaderBuildersTest.java | 29 +++++++++++++++---- python/flink_agents/api/yaml/loader.py | 2 +- .../api/yaml/tests/test_loader.py | 27 +++++++++++++++++ 4 files changed, 52 insertions(+), 8 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java index a9958f7c18..bbb1242e38 100644 --- a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java @@ -159,7 +159,7 @@ public static FunctionTool buildTool(ToolSpec spec) { /** Build a {@link Prompt} from a parsed {@link PromptSpec}. */ public static Prompt buildPrompt(PromptSpec spec) { - if (spec.getText() != null) { + if (spec.getText() != null && !spec.getText().isEmpty()) { return Prompt.fromText(spec.getText()); } List messages = diff --git a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java index 709cd1d0da..860727bee5 100644 --- a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderBuildersTest.java @@ -20,6 +20,8 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; +import org.apache.flink.agents.api.chat.messages.ChatMessage; +import org.apache.flink.agents.api.chat.messages.MessageRole; import org.apache.flink.agents.api.function.JavaFunction; import org.apache.flink.agents.api.function.PythonFunction; import org.apache.flink.agents.api.prompt.Prompt; @@ -35,7 +37,10 @@ import org.apache.flink.agents.api.yaml.spec.SkillsSpec; import org.apache.flink.agents.api.yaml.spec.ToolSpec; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.util.List; import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; @@ -178,15 +183,27 @@ void buildPromptText() throws Exception { assertThat(prompt.formatString(Map.of())).isEqualTo("hi"); } - @Test - void buildPromptMessages() throws Exception { + @ParameterizedTest + @ValueSource(strings = {"", "text: null\n", "text: \"\"\n"}) + void buildPromptMessagesWithUnsetText(String textField) throws Exception { PromptSpec spec = M.readValue( - "name: p\nmessages:\n - {role: system, content: hi}\n", PromptSpec.class); + "name: p\n" + + textField + + "messages:\n" + + " - {role: system, content: 'Hello {name}'}\n" + + " - {role: user, content: '{question}'}\n", + PromptSpec.class); Prompt prompt = YamlLoader.buildPrompt(spec); - // Both message-templated and text-templated prompts produce LocalPrompt instances; - // confirm the constructed prompt formats without error. - assertThat(prompt).isNotNull(); + List messages = + prompt.formatMessages( + MessageRole.SYSTEM, Map.of("name", "Alice", "question", "How are you?")); + assertThat(messages) + .extracting(ChatMessage::getRole) + .containsExactly(MessageRole.SYSTEM, MessageRole.USER); + assertThat(messages) + .extracting(ChatMessage::getText) + .containsExactly("Hello Alice", "How are you?"); } @Test diff --git a/python/flink_agents/api/yaml/loader.py b/python/flink_agents/api/yaml/loader.py index dec3c52664..77ba571193 100644 --- a/python/flink_agents/api/yaml/loader.py +++ b/python/flink_agents/api/yaml/loader.py @@ -196,7 +196,7 @@ def _build_tool(spec: ToolSpec) -> FunctionTool: def _build_prompt(spec: PromptSpec) -> Prompt: - if spec.text is not None: + if spec.text: return Prompt.from_text(spec.text) messages = [ ChatMessage.of(MessageRole(m.role.value), m.content) diff --git a/python/flink_agents/api/yaml/tests/test_loader.py b/python/flink_agents/api/yaml/tests/test_loader.py index 6abb8564ad..21ee9ceffe 100644 --- a/python/flink_agents/api/yaml/tests/test_loader.py +++ b/python/flink_agents/api/yaml/tests/test_loader.py @@ -222,6 +222,33 @@ def test_build_agents_loads_tools_and_prompts() -> None: assert msg_prompt.template[1].text == "{q}" +@pytest.mark.parametrize( + "text_field", ["", " text: null\n", ' text: ""\n'] +) +def test_build_agents_preserves_prompt_messages_with_unset_text( + tmp_path: Path, text_field: str +) -> None: + path = tmp_path / "prompt.yaml" + path.write_text( + "agents:\n" + " - name: chat_agent\n" + " prompts:\n" + " - name: greeting\n" + f"{text_field}" + " messages:\n" + " - {role: system, content: 'Hello {name}'}\n" + " - {role: user, content: '{question}'}\n" + ) + agents, _, _ = build_agents(path) + prompt = agents["chat_agent"].resources[ResourceType.PROMPT]["greeting"] + messages = prompt.format_messages(name="Alice", question="How are you?") + assert [message.role for message in messages] == [ + MessageRole.SYSTEM, + MessageRole.USER, + ] + assert [message.text for message in messages] == ["Hello Alice", "How are you?"] + + def test_build_agents_handles_shared_resources_and_actions() -> None: agents, shared_resources, shared_actions = build_agents( _FIXTURES / "with_shared.yaml" From 68515724bf4a3b3d87280fb4887b7fa9237613a2 Mon Sep 17 00:00:00 2001 From: WenjinXie Date: Fri, 16 Oct 2026 13:05:59 +0800 Subject: [PATCH 2/3] [api][java][python] Declare internal subagents in YAML Resolve agent-local subagent references against all Agents in the current file and Agents already registered in the same execution environment. Load files in order and reject duplicate Agent names before binding. Reuse the existing descriptor builder, plan cycle validation, serialization and runtime invocation path. Keep top-level subagents external and preserve external setup configuration arguments. Cover forward and cross-file references, reused and nested children, invalid declarations, cycles, duplicate names, failure state, shared child actions and schema parity. Verify parent and child application and a serialized-plan invocation on the Java operator. Update both published schema copies and the bundled contract hash. Generated-by: Codex CLI 0.153.4 (GPT-6) Co-authored-by: Codex --- .../flink/agents/api/yaml/YamlLoader.java | 69 +++- .../flink/agents/api/yaml/spec/AgentSpec.java | 6 +- .../api/yaml/spec/InternalSubagentSpec.java | 64 ++++ .../agents/api/yaml/spec/SubagentSpec.java | 78 +++++ .../api/yaml/YamlLoaderLoadYamlTest.java | 78 +++++ .../agents/api/yaml/YamlSubagentTest.java | 188 +++++++++++ .../api/yaml/spec/SchemaParityTest.java | 1 + .../assets/yaml-contracts.yaml | 2 +- .../flink-agents-dev/assets/yaml-schema.json | 55 +++- docs/yaml-schema.json | 55 +++- .../agents/plan/YamlSubagentPlanTest.java | 97 ++++++ python/flink_agents/api/yaml/loader.py | 68 +++- python/flink_agents/api/yaml/specs.py | 27 +- .../api/yaml/tests/test_subagents.py | 301 ++++++++++++++++++ .../plan/tests/test_yaml_subagents.py | 148 +++++++++ .../subagent/InternalSubagentCallTest.java | 56 ++++ 16 files changed, 1263 insertions(+), 30 deletions(-) create mode 100644 api/src/main/java/org/apache/flink/agents/api/yaml/spec/InternalSubagentSpec.java create mode 100644 api/src/main/java/org/apache/flink/agents/api/yaml/spec/SubagentSpec.java create mode 100644 api/src/test/java/org/apache/flink/agents/api/yaml/YamlSubagentTest.java create mode 100644 plan/src/test/java/org/apache/flink/agents/plan/YamlSubagentPlanTest.java create mode 100644 python/flink_agents/api/yaml/tests/test_subagents.py create mode 100644 python/flink_agents/plan/tests/test_yaml_subagents.py diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java index bbb1242e38..3b807bece3 100644 --- a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java @@ -32,14 +32,17 @@ import org.apache.flink.agents.api.resource.ResourceType; import org.apache.flink.agents.api.skills.SkillSourceSpec; import org.apache.flink.agents.api.skills.Skills; +import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder; import org.apache.flink.agents.api.tools.FunctionTool; import org.apache.flink.agents.api.yaml.spec.ActionSpec; import org.apache.flink.agents.api.yaml.spec.AgentActionRef; import org.apache.flink.agents.api.yaml.spec.AgentSpec; import org.apache.flink.agents.api.yaml.spec.DescriptorSpec; +import org.apache.flink.agents.api.yaml.spec.InternalSubagentSpec; import org.apache.flink.agents.api.yaml.spec.PackageSkillSpec; import org.apache.flink.agents.api.yaml.spec.PromptSpec; import org.apache.flink.agents.api.yaml.spec.SkillsSpec; +import org.apache.flink.agents.api.yaml.spec.SubagentSpec; import org.apache.flink.agents.api.yaml.spec.ToolSpec; import org.apache.flink.agents.api.yaml.spec.UrlSkillSpec; import org.apache.flink.agents.api.yaml.spec.YamlAgentsDocument; @@ -169,6 +172,37 @@ public static Prompt buildPrompt(PromptSpec spec) { return Prompt.fromMessages(messages); } + private static ResourceDescriptor buildSubagent(SubagentSpec spec, Map agents) { + if (spec.getExternal() != null) { + return buildDescriptor(spec.getExternal(), ResourceType.AGENT); + } + InternalSubagentSpec internal = spec.getInternal(); + Agent child = agents.get(internal.getAgent()); + if (child == null) { + throw new IllegalArgumentException( + "Sub-agent '" + + internal.getName() + + "' references unknown agent '" + + internal.getAgent() + + "'; the child must be declared in the same YAML file or already" + + " registered in this environment."); + } + InternalSubagentResourceDescriptorBuilder builder = + new InternalSubagentResourceDescriptorBuilder() + .agent(child) + .description(internal.getDescription()); + if (internal.getInputSchema() != null) { + try { + builder.inputSchema( + new ObjectMapper().writeValueAsString(internal.getInputSchema())); + } catch (IOException e) { + throw new IllegalArgumentException( + "Invalid input schema for sub-agent '" + internal.getName() + "'", e); + } + } + return builder.build(); + } + /** Build a {@link Skills} resource from a parsed {@link SkillsSpec}. */ public static Skills buildSkills(SkillsSpec spec) { List sources = new ArrayList<>(); @@ -250,6 +284,10 @@ Map getAgentSpecs() { /** Parse one YAML file and build the agents it declares. */ public static LoadedFile buildAgents(Path path) { + return buildAgents(path, Collections.emptyMap()); + } + + private static LoadedFile buildAgents(Path path, Map registeredAgents) { YamlAgentsDocument doc; try { doc = YAML_MAPPER.readValue(Files.readString(path), YamlAgentsDocument.class); @@ -267,10 +305,27 @@ public static LoadedFile buildAgents(Path path) { throw new IllegalArgumentException( "Duplicate agent name '" + spec.getName() + "' in " + path); } + if (registeredAgents.containsKey(spec.getName())) { + throw new IllegalArgumentException( + "Duplicate agent name '" + spec.getName() + "' (loading " + path + ")"); + } agentSpecs.put(spec.getName(), spec); agents.put(spec.getName(), buildAgent(spec)); } + Map availableAgents = new LinkedHashMap<>(registeredAgents); + availableAgents.putAll(agents); + // Bind after all in-file Agents exist, including references to previously loaded Agents. + for (AgentSpec spec : doc.getAgents()) { + for (SubagentSpec subagent : spec.getSubagents()) { + agents.get(spec.getName()) + .addResource( + subagent.getName(), + ResourceType.AGENT, + buildSubagent(subagent, availableAgents)); + } + } + Map> sharedResources = new EnumMap<>(ResourceType.class); for (ResourceType t : ResourceType.values()) { sharedResources.put(t, new LinkedHashMap<>()); @@ -334,11 +389,12 @@ public static void loadYaml(AgentsExecutionEnvironment env, Path path) { /** * Load multiple YAML files. Multiple calls accumulate. Duplicate names — both within a single * file (caught by {@link #buildAgents(Path)}) and across the current environment — raise {@link - * IllegalArgumentException}. + * IllegalArgumentException}. Files are loaded in order; child references may resolve Agents + * registered by earlier files or calls. */ public static void loadYaml(AgentsExecutionEnvironment env, List paths) { for (Path path : paths) { - LoadedFile loaded = buildAgents(path); + LoadedFile loaded = buildAgents(path, env.getAgents()); for (Map.Entry entry : loaded.getAgents().entrySet()) { AgentSpec spec = loaded.getAgentSpecs().get(entry.getKey()); @@ -368,14 +424,6 @@ public static void loadYaml(AgentsExecutionEnvironment env, List paths) { } } - // Cross-environment agent name uniqueness check. - for (String name : loaded.getAgents().keySet()) { - if (env.getAgents().containsKey(name)) { - throw new IllegalArgumentException( - "Duplicate agent name '" + name + "' (loading " + path + ")"); - } - } - // Commit shared resources — env.addResource enforces dedup with its own message. for (Map.Entry> e : loaded.getSharedResources().entrySet()) { @@ -413,7 +461,6 @@ private static Agent buildAgent(AgentSpec spec) { addAgentDescriptors(agent, ResourceType.EMBEDDING_MODEL, spec.getEmbeddingModelSetups()); addAgentDescriptors(agent, ResourceType.VECTOR_STORE, spec.getVectorStores()); addAgentDescriptors(agent, ResourceType.MCP_SERVER, spec.getMcpServers()); - addAgentDescriptors(agent, ResourceType.AGENT, spec.getSubagents()); for (ToolSpec t : spec.getTools()) { agent.addResource(t.getName(), ResourceType.TOOL, buildTool(t)); diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java index 1826a61a00..6f84d8ad34 100644 --- a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java @@ -40,7 +40,7 @@ public final class AgentSpec { private final List embeddingModelSetups; private final List vectorStores; private final List mcpServers; - private final List subagents; + private final List subagents; @JsonCreator public AgentSpec( @@ -57,7 +57,7 @@ public AgentSpec( @JsonProperty("embedding_model_setups") List embeddingModelSetups, @JsonProperty("vector_stores") List vectorStores, @JsonProperty("mcp_servers") List mcpServers, - @JsonProperty("subagents") List subagents) { + @JsonProperty("subagents") List subagents) { this.name = name; this.description = description; this.prompts = orEmpty(prompts); @@ -125,7 +125,7 @@ public List getMcpServers() { return mcpServers; } - public List getSubagents() { + public List getSubagents() { return subagents; } } diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/InternalSubagentSpec.java b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/InternalSubagentSpec.java new file mode 100644 index 0000000000..9698bf59bc --- /dev/null +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/InternalSubagentSpec.java @@ -0,0 +1,64 @@ +/* + * 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.api.yaml.spec; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.Map; + +/** A reference to a child Agent in this file or already registered in the environment. */ +public final class InternalSubagentSpec { + private final String name; + private final String agent; + private final String description; + private final Map inputSchema; + + @JsonCreator + public InternalSubagentSpec( + @JsonProperty(value = "name", required = true) String name, + @JsonProperty(value = "agent", required = true) String agent, + @JsonProperty("description") String description, + @JsonProperty("input_schema") Map inputSchema) { + if (agent == null || agent.isBlank()) { + throw new IllegalArgumentException( + "Internal sub-agent 'agent' must be a non-empty name"); + } + this.name = name; + this.agent = agent; + this.description = description == null ? "" : description; + this.inputSchema = inputSchema; + } + + public String getName() { + return name; + } + + public String getAgent() { + return agent; + } + + public String getDescription() { + return description; + } + + public Map getInputSchema() { + return inputSchema; + } +} diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/SubagentSpec.java b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/SubagentSpec.java new file mode 100644 index 0000000000..0516bc9463 --- /dev/null +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/SubagentSpec.java @@ -0,0 +1,78 @@ +/* + * 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.api.yaml.spec; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.DeserializationContext; +import com.fasterxml.jackson.databind.JsonDeserializer; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; + +import java.io.IOException; + +/** Either an internal Agent reference or an external sub-agent descriptor. */ +@JsonDeserialize(using = SubagentSpec.Deserializer.class) +public final class SubagentSpec { + private final InternalSubagentSpec internal; + private final DescriptorSpec external; + + private SubagentSpec(InternalSubagentSpec internal, DescriptorSpec external) { + this.internal = internal; + this.external = external; + } + + public String getName() { + return internal == null ? external.getName() : internal.getName(); + } + + public InternalSubagentSpec getInternal() { + return internal; + } + + public DescriptorSpec getExternal() { + return external; + } + + static final class Deserializer extends JsonDeserializer { + @Override + public SubagentSpec deserialize(JsonParser parser, DeserializationContext context) + throws IOException { + JsonNode node = parser.getCodec().readTree(parser); + if (!node.has("clazz")) { + for (String field : new String[] {"name", "agent"}) { + if (!node.path(field).isTextual()) { + context.reportInputMismatch( + InternalSubagentSpec.class, + "Internal sub-agent '%s' must be a string", + field); + } + } + if (node.hasNonNull("description") && !node.get("description").isTextual()) { + context.reportInputMismatch( + InternalSubagentSpec.class, + "Internal sub-agent 'description' must be a string or null"); + } + return new SubagentSpec( + parser.getCodec().treeToValue(node, InternalSubagentSpec.class), null); + } + return new SubagentSpec( + null, parser.getCodec().treeToValue(node, DescriptorSpec.class)); + } + } +} diff --git a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java index 13a5155ba6..00e616946a 100644 --- a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java @@ -23,6 +23,7 @@ import org.apache.flink.agents.api.agents.Agent; import org.apache.flink.agents.api.configuration.Configuration; import org.apache.flink.agents.api.function.JavaFunction; +import org.apache.flink.agents.api.resource.ResourceDescriptor; import org.apache.flink.agents.api.resource.ResourceType; import org.apache.flink.agents.api.skills.SkillSourceSpec; import org.apache.flink.agents.api.skills.Skills; @@ -31,6 +32,8 @@ import org.apache.flink.table.api.Table; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.nio.file.Files; import java.nio.file.Path; @@ -116,6 +119,81 @@ void sharedSkillsRegistered() { new SkillSourceSpec("local", Map.of("path", "./more"))); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void resolvesPreviouslyLoadedChild(boolean separateCalls, @TempDir Path tmp) throws Exception { + Path childPath = Files.writeString(tmp.resolve("child.yaml"), "agents: [{name: child}]"); + Path parentPath = + Files.writeString( + tmp.resolve("parent.yaml"), + "agents: [{name: parent, subagents: [{name: reviewer, agent: child}]}]"); + AgentsExecutionEnvironment env = new TestEnv(); + if (separateCalls) { + env.loadYaml(childPath); + env.loadYaml(parentPath); + } else { + env.loadYaml(List.of(childPath, parentPath)); + } + ResourceDescriptor descriptor = + (ResourceDescriptor) + env.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("reviewer"); + assertThat(descriptor.getInitialArguments().get("agent")) + .isSameAs(env.getAgents().get("child")); + assertThat(env.getAgents()).containsOnlyKeys("parent", "child"); + assertThat(env.getAgents().get("child").getResources().get(ResourceType.AGENT)).isEmpty(); + assertThatThrownBy(() -> new TestEnv().loadYaml(parentPath)) + .hasMessageContaining("unknown agent 'child'"); + } + + @Test + void childMustBeLoadedFirstAndFailurePreservesState(@TempDir Path tmp) throws Exception { + AgentsExecutionEnvironment env = new TestEnv(); + env.loadYaml( + Files.writeString( + tmp.resolve("existing.yaml"), + "agents: [{name: existing}]\nprompts: [{name: existing_prompt, text: keep}]")); + Agent original = env.getAgents().get("existing"); + Object prompt = env.getResources().get(ResourceType.PROMPT).get("existing_prompt"); + Path childPath = Files.writeString(tmp.resolve("child.yaml"), "agents: [{name: child}]"); + Path parentPath = + Files.writeString( + tmp.resolve("parent.yaml"), + "agents: [{name: parent, subagents: [{name: reviewer, agent: child}]}]\n" + + "prompts: [{name: new_prompt, text: 'not committed'}]"); + assertThatThrownBy(() -> env.loadYaml(List.of(parentPath, childPath))) + .hasMessageContaining("reviewer") + .hasMessageContaining("unknown agent 'child'") + .hasMessageContaining("already registered"); + assertThat(env.getAgents()).containsExactlyEntriesOf(Map.of("existing", original)); + assertThat(env.getResources().get(ResourceType.PROMPT)) + .containsExactlyEntriesOf(Map.of("existing_prompt", prompt)); + env.loadYaml(List.of(childPath, parentPath)); + assertThat(env.getAgents()).containsOnlyKeys("existing", "child", "parent"); + } + + @Test + void currentFileCannotReplaceRegisteredChild(@TempDir Path tmp) throws Exception { + AgentsExecutionEnvironment env = new TestEnv(); + env.loadYaml(Files.writeString(tmp.resolve("child.yaml"), "agents: [{name: child}]")); + Agent original = env.getAgents().get("child"); + Path duplicate = + Files.writeString( + tmp.resolve("duplicate.yaml"), + "agents: [{name: parent, subagents: [{name: reviewer, agent: child}]}, " + + "{name: child, prompts: [{name: replacement, text: bad}]}]\n" + + "prompts: [{name: new_prompt, text: 'not committed'}]"); + assertThatThrownBy(() -> env.loadYaml(duplicate)) + .hasMessageContaining("Duplicate agent name 'child'"); + assertThat(env.getAgents()).containsExactlyEntriesOf(Map.of("child", original)); + assertThat(original.getResources().get(ResourceType.PROMPT)).isEmpty(); + assertThat(env.getResources().values()) + .allSatisfy(resources -> assertThat(resources).isEmpty()); + } + /** Minimal env stub — we can't instantiate LocalExecutionEnvironment from the api module. */ private static final class TestEnv extends AgentsExecutionEnvironment { @Override diff --git a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlSubagentTest.java b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlSubagentTest.java new file mode 100644 index 0000000000..bf3c08ef3d --- /dev/null +++ b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlSubagentTest.java @@ -0,0 +1,188 @@ +/* + * 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.api.yaml; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; +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.yaml.spec.YamlAgentsDocument; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class YamlSubagentTest { + @TempDir Path directory; + + private YamlLoader.LoadedFile load(String yaml) throws Exception { + Path path = directory.resolve("agents.yaml"); + Files.writeString(path, yaml); + return YamlLoader.buildAgents(path); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void resolvesChildInEitherOrder(boolean childFirst) throws Exception { + String parent = + " - name: parent\n subagents:\n" + + " - name: reviewer\n agent: child\n" + + " description: Review code\n" + + " input_schema: {type: object, properties: {code: {type: string}}}\n"; + String child = + " - name: child\n prompts:\n - {name: child_prompt, text: 'Review {code}'}\n"; + YamlLoader.LoadedFile result = + load("agents:\n" + (childFirst ? child + parent : parent + child)); + ResourceDescriptor descriptor = + (ResourceDescriptor) + result.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("reviewer"); + assertThat(descriptor.getClazz()) + .isEqualTo(InternalSubagentResourceDescriptorBuilder.SETUP_CLASS); + assertThat(descriptor.getInitialArguments().get("agent")) + .isSameAs(result.getAgents().get("child")); + assertThat(descriptor.getInitialArguments()).containsEntry("description", "Review code"); + assertThat( + new ObjectMapper() + .readTree( + (String) + descriptor + .getInitialArguments() + .get("input_schema")) + .at("/properties/code/type") + .asText()) + .isEqualTo("string"); + assertThat(result.getAgents().get("child").getResources().get(ResourceType.PROMPT)) + .containsKey("child_prompt"); + } + + @Test + void externalDeclarationsPreserveProviderArguments() throws Exception { + String external = + "{name: remote, clazz: example.RemoteSetup, type: java, agent: remote-id}"; + YamlLoader.LoadedFile result = + load( + "agents: [{name: parent, subagents: [" + + external + + "]}]\nsubagents: [" + + external + + "]"); + ResourceDescriptor shared = + (ResourceDescriptor) + result.getSharedResources().get(ResourceType.AGENT).get("remote"); + ResourceDescriptor local = + (ResourceDescriptor) + result.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("remote"); + assertThat(shared.getInitialArguments()).containsEntry("agent", "remote-id"); + assertThat(local.getInitialArguments()).containsEntry("agent", "remote-id"); + } + + @ParameterizedTest + @ValueSource(strings = {"", ", description: null", ", description: ''"}) + void optionalMetadata(String metadata) throws Exception { + YamlLoader.LoadedFile result = + load( + "agents: [{name: parent, subagents: [{name: reviewer, agent: child" + + metadata + + "}]}, {name: child}]"); + ResourceDescriptor descriptor = + (ResourceDescriptor) + result.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("reviewer"); + assertThat(descriptor.getInitialArguments()) + .containsEntry("description", "") + .doesNotContainKey("input_schema"); + } + + @Test + void sharedInternalReferenceIsRejected() { + assertThatThrownBy( + () -> + load( + "agents: [{name: child}]\nsubagents: [{name: reviewer, agent: child}]")) + .hasMessageContaining("Failed to read YAML") + .hasStackTraceContaining("clazz"); + } + + @ParameterizedTest + @ValueSource( + strings = {"agents: [{name: parent, subagents: [{name: reviewer, agent: missing}]}]"}) + void unknownReferenceNamesTheMissingChild(String yaml) { + assertThatThrownBy(() -> load(yaml)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("reviewer") + .hasMessageContaining("missing") + .hasMessageContaining("same YAML file"); + } + + @ParameterizedTest + @ValueSource( + strings = { + "{name: reviewer, agent: ''}", + "{name: reviewer, agent: ' '}", + "{name: reviewer, agent: null}", + "{name: reviewer, agent: 123}", + "{name: reviewer, agent: true}", + "{name: null, agent: child}", + "{name: 123, agent: child}", + "{name: reviewer, agent: child, description: 123}", + "{name: reviewer, agent: child, type: java}", + "{name: reviewer, agent: child, input_schema: '{}'}", + "{name: reviewer, agent: child, unexpected: true}", + "{name: reviewer}" + }) + void rejectsInvalidDeclarations(String entry) { + assertThatThrownBy( + () -> + new ObjectMapper(new YAMLFactory()) + .readValue( + "agents: [{name: parent, subagents: [" + + entry + + "]}]", + YamlAgentsDocument.class)) + .isInstanceOf(Exception.class); + } + + @ParameterizedTest + @ValueSource( + strings = { + "agents: [{name: child}]\nsubagents: [{name: reviewer, clazz: example.Remote, type: java}, {name: reviewer, clazz: example.Remote, type: java}]", + "agents: [{name: child}, {name: parent, subagents: [{name: reviewer, agent: child}, {name: reviewer, agent: child}]}]" + }) + void duplicateResourceNamesAreRejected(String yaml) { + assertThatThrownBy(() -> load(yaml)).isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java b/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java index 519d9ef204..41abf55f18 100644 --- a/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java @@ -73,6 +73,7 @@ class SchemaParityTest { m.put("ActionSpec", ActionSpec.class); m.put("AgentSpec", AgentSpec.class); m.put("DescriptorSpec", DescriptorSpec.class); + m.put("InternalSubagentSpec", InternalSubagentSpec.class); m.put("PromptMessage", PromptMessage.class); m.put("PromptSpec", PromptSpec.class); m.put("SkillsSpec", SkillsSpec.class); diff --git a/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml b/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml index e1a66669bc..e18af319e7 100644 --- a/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml +++ b/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml @@ -13,7 +13,7 @@ contracts: repository: apache/flink-agents ref: main path: docs/yaml-schema.json - blob_sha: ed9800536d5f801e58b40db61b2488c2d5d5daf4 + blob_sha: 1b8270af990c9b80942e5cdf6319b18ff9b678e3 versions_without_yaml_api: - "0.2.1" diff --git a/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json b/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json index ed9800536d..1b8270af99 100644 --- a/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json +++ b/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json @@ -151,7 +151,14 @@ }, "subagents": { "items": { - "$ref": "#/$defs/DescriptorSpec" + "anyOf": [ + { + "$ref": "#/$defs/InternalSubagentSpec" + }, + { + "$ref": "#/$defs/DescriptorSpec" + } + ] }, "title": "Subagents", "type": "array" @@ -236,6 +243,52 @@ "title": "InjectedArg", "type": "object" }, + "InternalSubagentSpec": { + "additionalProperties": false, + "description": "Reference a child Agent in this file or already registered in the environment.", + "properties": { + "agent": { + "pattern": "\\S", + "title": "Agent", + "type": "string" + }, + "description": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Description" + }, + "input_schema": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Input Schema" + }, + "name": { + "title": "Name", + "type": "string" + } + }, + "required": [ + "name", + "agent" + ], + "title": "InternalSubagentSpec", + "type": "object" + }, "MessageRole": { "description": "Role of a message in a chat conversation.", "enum": [ diff --git a/docs/yaml-schema.json b/docs/yaml-schema.json index ed9800536d..1b8270af99 100644 --- a/docs/yaml-schema.json +++ b/docs/yaml-schema.json @@ -151,7 +151,14 @@ }, "subagents": { "items": { - "$ref": "#/$defs/DescriptorSpec" + "anyOf": [ + { + "$ref": "#/$defs/InternalSubagentSpec" + }, + { + "$ref": "#/$defs/DescriptorSpec" + } + ] }, "title": "Subagents", "type": "array" @@ -236,6 +243,52 @@ "title": "InjectedArg", "type": "object" }, + "InternalSubagentSpec": { + "additionalProperties": false, + "description": "Reference a child Agent in this file or already registered in the environment.", + "properties": { + "agent": { + "pattern": "\\S", + "title": "Agent", + "type": "string" + }, + "description": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Description" + }, + "input_schema": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Input Schema" + }, + "name": { + "title": "Name", + "type": "string" + } + }, + "required": [ + "name", + "agent" + ], + "title": "InternalSubagentSpec", + "type": "object" + }, "MessageRole": { "description": "Role of a message in a chat conversation.", "enum": [ diff --git a/plan/src/test/java/org/apache/flink/agents/plan/YamlSubagentPlanTest.java b/plan/src/test/java/org/apache/flink/agents/plan/YamlSubagentPlanTest.java new file mode 100644 index 0000000000..b42e349bdb --- /dev/null +++ b/plan/src/test/java/org/apache/flink/agents/plan/YamlSubagentPlanTest.java @@ -0,0 +1,97 @@ +/* + * 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.plan; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.flink.agents.api.resource.ResourceType; +import org.apache.flink.agents.api.yaml.YamlLoader; +import org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider; +import org.apache.flink.agents.plan.subagent.InternalSubagentPlans; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class YamlSubagentPlanTest { + @TempDir Path directory; + + private AgentPlan compile(String yaml) throws Exception { + Path path = directory.resolve("agents.yaml"); + Files.writeString(path, yaml); + return new AgentPlan(YamlLoader.buildAgents(path).getAgents().get("parent")); + } + + @Test + void nestedAndReusedChildrenSurvivePlanSerialization() throws Exception { + AgentPlan original = + compile( + "agents:\n" + + " - name: parent\n prompts: [{name: parent_only, text: parent}]\n" + + " subagents:\n" + + " - {name: first, agent: child, description: Review, input_schema: {type: object}}\n" + + " - {name: second, agent: child}\n" + + " - name: child\n prompts: [{name: child_only, text: child}]\n" + + " subagents: [{name: leaf, agent: leaf}]\n" + + " - name: leaf\n"); + ObjectMapper mapper = new ObjectMapper(); + AgentPlan plan = mapper.readValue(mapper.writeValueAsString(original), AgentPlan.class); + JavaResourceProvider first = + (JavaResourceProvider) + plan.getResourceProviders().get(ResourceType.AGENT).get("first"); + JavaResourceProvider second = + (JavaResourceProvider) + plan.getResourceProviders().get(ResourceType.AGENT).get("second"); + assertThat(first.getDescriptor().getInitialArguments()) + .containsEntry("scope", "first") + .containsEntry("description", "Review"); + assertThat(first.getDescriptor().getInitialArguments().get("plan")) + .isEqualTo(second.getDescriptor().getInitialArguments().get("plan")); + assertThat( + mapper.readTree( + (String) + first.getDescriptor() + .getInitialArguments() + .get("input_schema")) + .get("type") + .asText()) + .isEqualTo("object"); + AgentPlan child = InternalSubagentPlans.decode(first.getDescriptor()); + assertThat(child.getResourceProviders().get(ResourceType.AGENT)).containsKey("leaf"); + assertThat(child.getResourceProviders().get(ResourceType.PROMPT)) + .containsKey("child_only") + .doesNotContainKey("parent_only"); + } + + @ParameterizedTest + @ValueSource( + strings = { + "agents: [{name: parent, subagents: [{name: self, agent: parent}]}]", + "agents: [{name: parent, subagents: [{name: child, agent: child}]}, {name: child, subagents: [{name: parent, agent: parent}]}]", + "agents: [{name: parent, subagents: [{name: a, agent: a}]}, {name: a, subagents: [{name: b, agent: b}]}, {name: b, subagents: [{name: a, agent: a}]}]" + }) + void recursiveReferencesFailPlanConstruction(String yaml) { + assertThatThrownBy(() -> compile(yaml)).hasMessageContaining("Cyclic sub-agent"); + } +} diff --git a/python/flink_agents/api/yaml/loader.py b/python/flink_agents/api/yaml/loader.py index 77ba571193..16aac3c245 100644 --- a/python/flink_agents/api/yaml/loader.py +++ b/python/flink_agents/api/yaml/loader.py @@ -19,6 +19,7 @@ environment. """ +import json from pathlib import Path from typing import TYPE_CHECKING, Any, Dict, List, Tuple @@ -42,6 +43,7 @@ ResourceType, ) from flink_agents.api.skills import Skills, SkillSourceSpec +from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder from flink_agents.api.tools.function_tool import FunctionTool from flink_agents.api.yaml.aliases import ( resolve_clazz, @@ -54,6 +56,7 @@ Language, PromptSpec, SkillsSpec, + SubagentSpec, ToolSpec, YamlAgentsDocument, ) @@ -65,7 +68,6 @@ "embedding_model_setups": ResourceType.EMBEDDING_MODEL, "vector_stores": ResourceType.VECTOR_STORE, "mcp_servers": ResourceType.MCP_SERVER, - "subagents": ResourceType.AGENT, } @@ -152,6 +154,27 @@ def _build_descriptor( return ResourceDescriptor(clazz=python_fqn, **kwargs) +def _build_subagent(spec: SubagentSpec, agents: Dict[str, Agent]) -> ResourceDescriptor: + if isinstance(spec, DescriptorSpec): + return _build_descriptor(spec, ResourceType.AGENT) + child = agents.get(spec.agent) + if child is None: + msg = ( + f"Sub-agent {spec.name!r} references unknown agent {spec.agent!r}; " + "the child must be declared in the same YAML file or already " + "registered in this environment." + ) + raise ValueError(msg) + builder = ( + InternalSubagentResourceDescriptorBuilder() + .agent(child) + .description(spec.description or "") + ) + if spec.input_schema is not None: + builder.input_schema(json.dumps(spec.input_schema)) + return builder.build() + + def _add_descriptors_to_agent( agent: Agent, attr_name: str, descriptors: list[DescriptorSpec] ) -> None: @@ -266,6 +289,7 @@ def _build_agent(agent_spec: AgentSpec) -> Agent: def _build_in_file_state( path: Path | str, + registered_agents: Dict[str, Agent] | None = None, ) -> Tuple[ Dict[str, Agent], Dict[ResourceType, Dict[str, Any]], @@ -273,8 +297,10 @@ def _build_in_file_state( Dict[str, AgentSpec], YamlAgentsDocument, ]: - """Parse one YAML file, perform in-file duplicate detection, and build - the in-memory state without touching any execution environment. + """Parse one YAML file and build its state without modifying the environment. + + Registered Agents are available for child references; their names must not + collide with declarations in this file. Returns: agents: name -> Agent @@ -288,16 +314,38 @@ def _build_in_file_state( exactly one place. """ doc = _load_document(path) + registered_agents = registered_agents or {} agent_specs: Dict[str, AgentSpec] = {} agents: Dict[str, Agent] = {} for spec in doc.agents: if spec.name in agents: msg = f"Duplicate agent name {spec.name!r} in {path}" raise ValueError(msg) + if spec.name in registered_agents: + msg = f"Duplicate agent name {spec.name!r} (loading {path})" + raise ValueError(msg) agent_specs[spec.name] = spec agents[spec.name] = _build_agent(spec) + # Include previously loaded Agents without modifying the environment's registry. + available_agents = {**registered_agents, **agents} + # Create every in-file Agent before binding, allowing forward references. + for spec in doc.agents: + for subagent in spec.subagents: + agents[spec.name].add_resource( + subagent.name, + ResourceType.AGENT, + _build_subagent(subagent, available_agents), + ) + shared_resources: Dict[ResourceType, Dict[str, Any]] = {t: {} for t in ResourceType} + for spec in doc.subagents: + if spec.name in shared_resources[ResourceType.AGENT]: + msg = f"Duplicate shared resource name {spec.name!r} in {path}" + raise ValueError(msg) + shared_resources[ResourceType.AGENT][spec.name] = _build_descriptor( + spec, ResourceType.AGENT + ) for attr, resource_type in _DESCRIPTOR_TYPES.items(): for spec in getattr(doc, attr): if spec.name in shared_resources[resource_type]: @@ -386,8 +434,9 @@ def load_yaml( resources on the environment. Multiple calls accumulate. Duplicate names — both within a single file - and across the current environment — raise ``ValueError``. In-file - duplicate detection is delegated to :func:`_build_in_file_state` so + and across the current environment — raise ``ValueError``. Files are loaded + in order, so child references can use Agents from earlier files or calls. + Agent name checks are delegated to :func:`_build_in_file_state` so that ``load_yaml`` and :func:`build_agents` share the same rules. """ if isinstance(paths, str | Path): @@ -395,15 +444,10 @@ def load_yaml( for path in paths: agents, shared_resources, shared_actions, agent_specs, _ = _build_in_file_state( - path + path, env._agents ) - # Cross-environment duplicate checks. In-file duplicates were - # already caught inside ``_build_in_file_state``. - for name in agents: - if name in env._agents: - msg = f"Duplicate agent name {name!r} (loading {path})" - raise ValueError(msg) + # Agent name conflicts were already caught inside ``_build_in_file_state``. for resource_type, name_to_resource in shared_resources.items(): for name in name_to_resource: if name in env.resources[resource_type]: diff --git a/python/flink_agents/api/yaml/specs.py b/python/flink_agents/api/yaml/specs.py index 063c30f1d8..6f46fb3ed6 100644 --- a/python/flink_agents/api/yaml/specs.py +++ b/python/flink_agents/api/yaml/specs.py @@ -29,6 +29,7 @@ from pydantic import ( BaseModel, + BeforeValidator, ConfigDict, Field, StrictBool, @@ -62,6 +63,30 @@ class DescriptorSpec(BaseModel): type: Language | None = None +class InternalSubagentSpec(BaseModel): + """Reference a child Agent in this file or already registered in the environment.""" + + model_config = ConfigDict(extra="forbid") + + name: str + agent: Annotated[str, Field(pattern=r"\S")] + description: str | None = None + input_schema: Dict[str, Any] | None = None + + +def _parse_subagent(value: Any) -> Any: + if isinstance(value, dict): + spec = DescriptorSpec if "clazz" in value else InternalSubagentSpec + return spec.model_validate(value) + return value + + +SubagentSpec = Annotated[ + InternalSubagentSpec | DescriptorSpec, + BeforeValidator(_parse_subagent), +] + + class MessageRole(str, Enum): """Role of a message in a chat conversation.""" @@ -258,7 +283,7 @@ class AgentSpec(BaseModel): embedding_model_setups: List[DescriptorSpec] = Field(default_factory=list) vector_stores: List[DescriptorSpec] = Field(default_factory=list) mcp_servers: List[DescriptorSpec] = Field(default_factory=list) - subagents: List[DescriptorSpec] = Field(default_factory=list) + subagents: List[SubagentSpec] = Field(default_factory=list) class YamlAgentsDocument(BaseModel): diff --git a/python/flink_agents/api/yaml/tests/test_subagents.py b/python/flink_agents/api/yaml/tests/test_subagents.py new file mode 100644 index 0000000000..477a18e54e --- /dev/null +++ b/python/flink_agents/api/yaml/tests/test_subagents.py @@ -0,0 +1,301 @@ +################################################################################ +# 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. +################################################################################# +"""YAML declarations of internal and external sub-agents.""" + +import json +from pathlib import Path + +import pytest +import yaml +from pydantic import ValidationError + +from flink_agents.api.resource import ResourceType +from flink_agents.api.subagent import INTERNAL_SUBAGENT_SETUP_CLASS +from flink_agents.api.yaml.loader import build_agents, load_yaml +from flink_agents.api.yaml.specs import YamlAgentsDocument +from flink_agents.api.yaml.tests.test_loader import _registration_env + + +def _write(tmp_path: Path, document: dict, filename: str = "agents.yaml") -> Path: + path = tmp_path / filename + path.write_text(yaml.safe_dump(document)) + return path + + +@pytest.mark.parametrize("child_first", [False, True]) +def test_internal_reference_preserves_child_and_metadata( + tmp_path: Path, child_first: bool +) -> None: + schema = {"type": "object", "properties": {"code": {"type": "string"}}} + parent = { + "name": "parent", + "subagents": [ + { + "name": "reviewer", + "agent": "child", + "description": "Review code", + "input_schema": schema, + } + ], + } + child = { + "name": "child", + "prompts": [{"name": "child_prompt", "text": "Review {code}"}], + } + agents, _, _ = build_agents( + _write( + tmp_path, {"agents": [child, parent] if child_first else [parent, child]} + ) + ) + descriptor = agents["parent"].resources[ResourceType.AGENT]["reviewer"] + assert descriptor.target_clazz == INTERNAL_SUBAGENT_SETUP_CLASS + assert descriptor.arguments["agent"] is agents["child"] + assert descriptor.arguments["description"] == "Review code" + assert json.loads(descriptor.arguments["input_schema"]) == schema + assert "child_prompt" in agents["child"].resources[ResourceType.PROMPT] + + +def test_external_subagents_preserve_provider_arguments(tmp_path: Path) -> None: + external = {"name": "remote", "clazz": "my_app.RemoteSetup", "agent": "remote-id"} + agents, resources, _ = build_agents( + _write( + tmp_path, + { + "agents": [{"name": "parent", "subagents": [external]}], + "subagents": [external], + }, + ) + ) + assert resources[ResourceType.AGENT]["remote"].arguments == {"agent": "remote-id"} + assert agents["parent"].resources[ResourceType.AGENT]["remote"].arguments == { + "agent": "remote-id" + } + + +@pytest.mark.parametrize("metadata", [{}, {"description": None}, {"description": ""}]) +def test_optional_internal_metadata(tmp_path: Path, metadata: dict) -> None: + agents, _, _ = build_agents( + _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "child", "agent": "child", **metadata}], + }, + {"name": "child"}, + ] + }, + ) + ) + arguments = agents["parent"].resources[ResourceType.AGENT]["child"].arguments + assert arguments["description"] == "" + assert "input_schema" not in arguments + + +def test_shared_internal_reference_rejected_before_registration(tmp_path: Path) -> None: + env = _registration_env() + with pytest.raises(ValidationError, match="clazz"): + load_yaml( + env, + _write( + tmp_path, + { + "agents": [{"name": "child"}], + "subagents": [{"name": "reviewer", "agent": "child"}], + }, + ), + ) + assert not env._agents + assert all(not entries for entries in env.resources.values()) + + +def test_two_parents_share_child_definition(tmp_path: Path) -> None: + agents, _, _ = build_agents( + _write( + tmp_path, + { + "agents": [ + { + "name": name, + "subagents": [{"name": "reviewer", "agent": "child"}], + } + for name in ["one", "two"] + ] + + [{"name": "child"}] + }, + ) + ) + for name in ["one", "two"]: + assert ( + agents[name].resources[ResourceType.AGENT]["reviewer"].arguments["agent"] + is agents["child"] + ) + + +def test_unknown_child_does_not_register_partial_state(tmp_path: Path) -> None: + document = {"agents": [{"name": "parent"}]} + document["agents"][0]["subagents"] = [{"name": "reviewer", "agent": "missing"}] + env = _registration_env() + with pytest.raises( + ValueError, match="reviewer.*unknown agent.*missing.*same YAML file" + ): + load_yaml(env, _write(tmp_path, document)) + assert not env._agents + assert all(not entries for entries in env.resources.values()) + + +@pytest.mark.parametrize( + "entry", + [ + {"name": "reviewer", "agent": ""}, + {"name": "reviewer", "agent": " "}, + {"name": "reviewer", "agent": None}, + {"name": "reviewer", "agent": 123}, + {"name": "reviewer", "agent": True}, + {"name": None, "agent": "child"}, + {"name": 123, "agent": "child"}, + {"name": "reviewer", "agent": "child", "description": 123}, + {"name": "reviewer", "agent": "child", "type": "python"}, + {"name": "reviewer", "agent": "child", "input_schema": "{}"}, + {"name": "reviewer", "agent": "child", "unexpected": True}, + {"name": "reviewer"}, + ], +) +def test_rejects_invalid_subagent_declarations(entry: dict) -> None: + with pytest.raises(ValidationError): + YamlAgentsDocument.model_validate( + {"agents": [{"name": "parent", "subagents": [entry]}]} + ) + + +@pytest.mark.parametrize("shared", [False, True]) +def test_rejects_duplicate_subagent_resource_names( + tmp_path: Path, shared: bool +) -> None: + document = {"agents": [{"name": "parent"}, {"name": "child"}]} + target = document if shared else document["agents"][0] + entry = ( + {"name": "reviewer", "clazz": "my_app.RemoteSetup"} + if shared + else {"name": "reviewer", "agent": "child"} + ) + target["subagents"] = [entry] * 2 + with pytest.raises(ValueError, match="[Dd]uplicate|already"): + build_agents(_write(tmp_path, document)) + + +@pytest.mark.parametrize("separate_calls", [False, True]) +def test_child_in_previously_loaded_file_is_resolved( + tmp_path: Path, separate_calls: bool +) -> None: + child_path = _write(tmp_path, {"agents": [{"name": "child"}]}, "child.yaml") + parent_path = _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "reviewer", "agent": "child"}], + } + ] + }, + "parent.yaml", + ) + env = _registration_env() + if separate_calls: + load_yaml(env, child_path) + load_yaml(env, parent_path) + else: + load_yaml(env, [child_path, parent_path]) + descriptor = env._agents["parent"].resources[ResourceType.AGENT]["reviewer"] + assert descriptor.arguments["agent"] is env._agents["child"] + assert set(env._agents) == {"parent", "child"} + assert not env._agents["child"].resources[ResourceType.AGENT] + # Registrations belong to one environment, not to a process-global registry. + with pytest.raises(ValueError, match="unknown agent.*child"): + load_yaml(_registration_env(), parent_path) + + +def test_child_must_be_loaded_before_parent_and_failure_preserves_state( + tmp_path: Path, +) -> None: + env = _registration_env() + load_yaml( + env, + _write( + tmp_path, + { + "agents": [{"name": "existing"}], + "prompts": [{"name": "existing_prompt", "text": "keep"}], + }, + ), + ) + original = env._agents["existing"] + prompt = env.resources[ResourceType.PROMPT]["existing_prompt"] + child_path = _write(tmp_path, {"agents": [{"name": "child"}]}, "child.yaml") + parent_path = _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "reviewer", "agent": "child"}], + } + ], + "prompts": [{"name": "new_prompt", "text": "not committed"}], + }, + "parent.yaml", + ) + with pytest.raises( + ValueError, match="reviewer.*unknown agent.*child.*already registered" + ): + load_yaml(env, [parent_path, child_path]) + assert env._agents == {"existing": original} + assert env.resources[ResourceType.PROMPT] == {"existing_prompt": prompt} + load_yaml(env, [child_path, parent_path]) + assert set(env._agents) == {"existing", "child", "parent"} + + +def test_current_file_cannot_replace_registered_child(tmp_path: Path) -> None: + env = _registration_env() + load_yaml(env, _write(tmp_path, {"agents": [{"name": "child"}]})) + original = env._agents["child"] + with pytest.raises(ValueError, match="Duplicate agent name.*child"): + load_yaml( + env, + _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "reviewer", "agent": "child"}], + }, + { + "name": "child", + "prompts": [{"name": "replacement", "text": "bad"}], + }, + ], + "prompts": [{"name": "new_prompt", "text": "not committed"}], + }, + ), + ) + assert env._agents == {"child": original} + assert not original.resources[ResourceType.PROMPT] + assert all(not entries for entries in env.resources.values()) diff --git a/python/flink_agents/plan/tests/test_yaml_subagents.py b/python/flink_agents/plan/tests/test_yaml_subagents.py new file mode 100644 index 0000000000..8b22d01cab --- /dev/null +++ b/python/flink_agents/plan/tests/test_yaml_subagents.py @@ -0,0 +1,148 @@ +################################################################################ +# 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. +################################################################################ +"""Compile YAML parent/child declarations through the production plan path.""" + +import json +from pathlib import Path + +import pytest +import yaml + +from flink_agents.api.resource import ResourceType +from flink_agents.api.yaml.loader import build_agents, load_yaml +from flink_agents.api.yaml.tests.test_loader import _registration_env +from flink_agents.plan.agent_plan import AgentPlan +from flink_agents.plan.configuration import AgentConfiguration +from flink_agents.plan.subagent import decode_plan + + +def _write(tmp_path: Path, agents: list[dict], **shared: object) -> Path: + path = tmp_path / "agents.yaml" + path.write_text(yaml.safe_dump({"agents": agents, **shared})) + return path + + +@pytest.mark.parametrize("separate_files", [False, True]) +def test_yaml_child_plan_preserves_actions_metadata_and_local_resources( + tmp_path: Path, + separate_files: bool, +) -> None: + path = _write( + tmp_path, + [ + { + "name": "parent", + "prompts": [{"name": "parent_only", "text": "parent"}], + "subagents": [ + { + "name": "reviewer", + "agent": "child", + "description": "Review code", + "input_schema": { + "type": "object", + "properties": {"code": {"type": "string"}}, + }, + } + ], + }, + { + "name": "child", + "actions": ["increment"], + "prompts": [{"name": "child_only", "text": "child"}], + }, + ], + actions=[ + { + "name": "increment", + "function": "flink_agents.api.yaml.tests.fixtures.loader_targets:increment", + "trigger_conditions": ["input"], + } + ], + ) + env = _registration_env() + if separate_files: + document = yaml.safe_load(path.read_text()) + parent = document["agents"].pop(0) + child_path = tmp_path / "child.yaml" + child_path.write_text(yaml.safe_dump(document)) + path.write_text(yaml.safe_dump({"agents": [parent]})) + load_yaml(env, [child_path, path]) + else: + load_yaml(env, path) + original = env._agents["parent"].resources[ResourceType.AGENT]["reviewer"] + plan = AgentPlan.from_agent(env._agents["parent"], AgentConfiguration()) + restored = AgentPlan.model_validate_json( + plan.model_dump_json(serialize_as_any=True) + ) + descriptor = restored.resource_providers[ResourceType.AGENT]["reviewer"].descriptor + assert descriptor.arguments["scope"] == "reviewer" + assert descriptor.arguments["description"] == "Review code" + assert "code" in json.loads(descriptor.arguments["input_schema"])["properties"] + child = decode_plan(descriptor) + assert "increment" in child.actions + assert "child_only" in child.resource_providers[ResourceType.PROMPT] + assert "parent_only" not in child.resource_providers[ResourceType.PROMPT] + assert original.arguments["agent"] is env._agents["child"] + assert "plan" not in original.arguments + + +@pytest.mark.parametrize( + "edges", + [ + {"parent": "parent"}, + {"parent": "child", "child": "parent"}, + {"parent": "a", "a": "b", "b": "a"}, + ], +) +def test_yaml_cycles_use_existing_plan_validation(tmp_path: Path, edges: dict) -> None: + path = _write( + tmp_path, + [ + {"name": name, "subagents": [{"name": child, "agent": child}]} + for name, child in edges.items() + ], + ) + agents, _, _ = build_agents(path) + with pytest.raises(ValueError, match="Cyclic sub-agent definition"): + AgentPlan.from_agent(agents["parent"], AgentConfiguration()) + + +def test_yaml_nested_children_and_reused_child_compile(tmp_path: Path) -> None: + path = _write( + tmp_path, + [ + { + "name": "parent", + "subagents": [ + {"name": "first", "agent": "child"}, + {"name": "second", "agent": "child"}, + ], + }, + {"name": "child", "subagents": [{"name": "leaf", "agent": "leaf"}]}, + {"name": "leaf"}, + ], + ) + agents, _, _ = build_agents(path) + plan = AgentPlan.from_agent(agents["parent"], AgentConfiguration()) + resources = plan.resource_providers[ResourceType.AGENT] + assert ( + resources["first"].descriptor.arguments["plan"] + == resources["second"].descriptor.arguments["plan"] + ) + child = decode_plan(resources["first"].descriptor) + assert "leaf" in child.resource_providers[ResourceType.AGENT] diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java index a932b283ca..619fc2f452 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java @@ -34,10 +34,12 @@ import org.apache.flink.agents.api.subagent.SubagentResult; import org.apache.flink.agents.api.subagent.SubagentSetup; import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.runtime.env.RemoteExecutionEnvironment; 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.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; import org.junit.jupiter.api.BeforeEach; @@ -45,7 +47,12 @@ import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.condition.EnabledForJreRange; import org.junit.jupiter.api.condition.JRE; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -244,6 +251,55 @@ public static void writeMemoryThenCall(Event event, RunnerContext ctx) throws Ex // --- tests --- + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @Timeout(60) + void yamlParentInvokesChildAfterPlanSerialization( + boolean separateFiles, @TempDir Path directory) throws Exception { + Path path = directory.resolve("agents.yaml"); + String targets = InternalSubagentCallTest.class.getName(); + Files.writeString( + path, + "agents:\n" + + " - name: parent\n" + + " subagents: [{name: child, agent: echo}]\n" + + " actions:\n" + + " - name: call\n type: java\n trigger_conditions: [input]\n" + + " function: " + + targets + + ":callOnce\n" + + " - name: echo\n actions: [echo]\n" + + "actions:\n" + + " - name: echo\n type: java\n trigger_conditions: [input]\n" + + " function: " + + targets + + "$EchoChildAgent:echo\n"); + StreamExecutionEnvironment flink = StreamExecutionEnvironment.getExecutionEnvironment(); + RemoteExecutionEnvironment env = new RemoteExecutionEnvironment(flink, null); + if (separateFiles) { + String yaml = Files.readString(path); + int childStart = yaml.indexOf(" - name: echo\n"); + Path childPath = + Files.writeString( + directory.resolve("child.yaml"), + "agents:\n" + yaml.substring(childStart)); + Files.writeString(path, yaml.substring(0, childStart)); + env.loadYaml(List.of(childPath, path)); + } else { + env.loadYaml(path); + } + // The child remains independently applicable, including between two parent compilations. + for (String name : List.of("parent", "echo", "parent")) { + env.fromDataStream(flink.fromElements(7L), value -> value).apply(name); + } + AgentPlan original = new AgentPlan(env.getAgents().get("parent")); + ObjectMapper mapper = new ObjectMapper(); + AgentPlan transferred = + mapper.readValue(mapper.writeValueAsString(original), AgentPlan.class); + assertThat(run(transferred, 7L)).containsExactly("echo:7"); + assertThat(CHILD_INVOCATIONS).containsExactly("7"); + } + @Test @Timeout(60) void callerReceivesTheChildsAccumulatedOutput() throws Exception { From 442446ef278349533194e969050080ac1465a86b Mon Sep 17 00:00:00 2001 From: WenjinXie Date: Fri, 16 Oct 2026 13:06:20 +0800 Subject: [PATCH 3/3] [docs] Clarify YAML declarations and document subagents Explain the boundary between YAML configuration and implementation code. Clarify prompt validation, required fields, type applicability and provider aliases. Keep tool injection and skill source behavior in their dedicated guides, and remove maintainer-only mapping and consistency details. Document internal child references and external subagent setups, including examples, input schema configuration, cross-file loading order, name uniqueness and cycle validation. Link the YAML and Subagents guides in both directions. Validated with Hugo and internal links/anchors in both the current tree and a temporary documentation overlay of PR #1261. Subagent examples pass the Pydantic models and generated JSON Schema, with the incremental fragment combined with the earlier single-agent example. Generated-by: Codex CLI 0.153.4 (GPT-6) Co-authored-by: Codex --- .../docs/development/create-agent/yaml.md | 334 +++++++++++------- .../docs/development/multi-agent/subagents.md | 5 + 2 files changed, 210 insertions(+), 129 deletions(-) diff --git a/docs/content/docs/development/create-agent/yaml.md b/docs/content/docs/development/create-agent/yaml.md index b00b0f6546..c0a81e80f8 100644 --- a/docs/content/docs/development/create-agent/yaml.md +++ b/docs/content/docs/development/create-agent/yaml.md @@ -26,15 +26,27 @@ under the License. ## Overview -Beyond the Python and Java programmatic APIs, Flink Agents offers a **declarative YAML API** for describing agents. Compared with the programmatic APIs, the YAML API has the following advantages: +Flink Agents provides a **declarative YAML API** for defining and orchestrating Agents. +It uses the same Events, Actions, and Resources as the Java and Python APIs, described in +[Core Concepts]({{< ref "docs/development/create-agent/overview#core-concepts" >}}). -- **Human-friendly**: low entry barrier and easy to templatize across deployments. -- **Coding-agent-friendly**: a fixed schema reduces token cost, enables strict schema validation, and decouples configuration changes (declared parameters) from logic changes (action code). +Parts of an Agent definition can be expressed directly in YAML, while the +implementation logic remains in Java or Python code: -The YAML API uses the same Events, Actions, and Resources described in -[Core Concepts]({{< ref "docs/development/create-agent/overview#core-concepts" >}}). +**What can be defined in YAML:** + +- Which Actions make up an Agent and which Events trigger them. +- Which Resources are available and how components reference them. +- The existing implementations to use and their configuration. +- Inline prompt text or messages. -A YAML file describes **Resources** and **Actions** declaratively; the **Event** types are referenced by name (built-in event aliases like `input` / `chat_response`, or your own event-type strings). The implementation classes (action functions, tool functions, custom types) still live in your Python or Java code; the YAML file only describes *how* those pieces are wired together. +**What remains in code:** + +- Action and tool logic. +- Custom Resource implementations. +- Custom Event classes. + +This lets you change component configuration and wiring while reusing the same code. If you'd rather declare agents directly in code, see [Agent]({{< ref "docs/development/create-agent/overview" >}}) and [Built-in ReAct Agent]({{< ref "docs/development/react_agent" >}}). @@ -45,6 +57,7 @@ This section walks through: 1. Declaring a single agent in one YAML file. 2. Declaring multiple agents in one YAML file. 3. Declaring shared Resources and Actions that any agent in the file (or any later-loaded file) can reuse. +4. Declaring sub-agents and referencing them from a parent Agent. ### Declaring a single agent @@ -101,28 +114,33 @@ agents: | Field | Required | Description | |-------|----------|-------------| -| `name` | yes | Agent name. Must be unique across the environment. Used to apply the agent by name later. | -| `description` | no | Free-form description of what the agent does. Not surfaced at runtime. | +| `name` | Required | Agent name. Must be unique across the environment. Used to apply the agent by name later. | +| `description` | Optional | Free-form description of what the agent does. Not surfaced at runtime. | #### Resources Declare Resources as lists under the agent, with a `name` unique within each section. The section determines the resource type: for example, `chat_model_setups` registers -`CHAT_MODEL` Resources and `tools` registers `TOOL` Resources. The optional `type: python` -or `type: java` field selects the implementation language. +`CHAT_MODEL` Resources and `tools` registers `TOOL` Resources. For Resources configured +by implementation class and for Tools, the optional `type: python` or `type: java` +field selects the implementation language. Prompts and Skills do not accept `type`. These declarations register Resources for use by name, just like decorators, annotations, or `add_resource` / `addResource`. Actions can get their instances through `ctx.get_resource` / `ctx.getResource`; see [Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}}). -**Prompt** — declarative prompt template; pick exactly one of `text` or `messages`. +**Prompt** — declarative prompt template. Exactly one of `text` or `messages` must be non-empty. | Field | Required | Description | |-------|----------|-------------| -| `name` | yes | Prompt name (referenced by chat-model setups). | -| `text` | one-of | Single-string prompt template. Corresponds to `Prompt.from_text` / `Prompt.fromText`. | -| `messages` | one-of | Multi-turn message template. Corresponds to `Prompt.from_messages` / `Prompt.fromMessages`. Each entry has `role` (`system` / `user` / `assistant` / `tool`) and `content`. | +| `name` | Required | Prompt name (referenced by chat-model setups). | +| `text` | Required if `messages` is unset or empty | Non-empty string prompt template. Corresponds to `Prompt.from_text` / `Prompt.fromText`. | +| `messages` | Required if `text` is unset or empty | Non-empty list of message templates. Corresponds to `Prompt.from_messages` / `Prompt.fromMessages`. Each entry has `role` (`system` / `user` / `assistant` / `tool`) and `content`. | + +Omitted or `null` fields, `text: ""`, and `messages: []` are treated as unset. +If both alternatives are non-empty, or neither is non-empty, YAML loading fails +validation. Neither alternative takes precedence over the other. ```yaml prompts: @@ -140,11 +158,11 @@ prompts: | Field | Required | Description | |-------|----------|-------------| -| `name` | yes | Tool name (referenced from `chat_model_setups[].tools`). | -| `function` | yes | Fully-qualified callable in the form `:`. See [Function references](#function-references). | -| `type` | no | Implementation language: `python` or `java`. Defaults to `python` (see [Selecting the implementation language](#selecting-the-implementation-language)). | -| `parameter_types` | java only | Required for Java tools — one Java type FQN per declared parameter, in order. Forbidden for Python tools (the signature is reflected from the callable). | -| `injected_args` | no | Tool arguments injected by the framework at execution time. Declare a map from tool parameter name to `{source, key}`. These arguments are hidden from the model-facing tool schema. | +| `name` | Required | Tool name (referenced from `chat_model_setups[].tools`). | +| `function` | Required | Fully-qualified callable in the form `:`. See [Function references](#function-references). | +| `type` | Optional | Implementation language: `python` or `java`. Defaults to `python` (see [Selecting the implementation language](#selecting-the-implementation-language)). | +| `parameter_types` | Required for Java / Forbidden for Python | One Java type per declared parameter, in order. Use `[]` for a Java tool with no parameters. Python tool signatures are reflected from the callable. | +| `injected_args` | Optional | Tool arguments injected by the framework at execution time. Declare a map from tool parameter name to `{source, key}`. These arguments are hidden from the model-facing tool schema. | ```yaml tools: @@ -153,7 +171,7 @@ tools: type: python ``` -For **Java tools**, `parameter_types` is required because Java methods can be overloaded — list one FQN per declared parameter, in order. Generic type arguments are not part of the JVM method descriptor and must not be included (`java.util.List`, not `java.util.List`). Boxed primitives (`java.lang.Integer`, etc.) and bare primitives (`int`, `boolean`, ...) are both accepted. +For **Java tools**, `parameter_types` is required because Java methods can be overloaded — list one Java type per declared parameter, in order. ```yaml tools: @@ -163,84 +181,52 @@ tools: parameter_types: [java.lang.Integer, java.lang.Integer] ``` -To hide framework-owned arguments from the model, declare them in `injected_args` and bind each argument to a runtime source: - -```yaml -agents: - - name: order_agent - tools: - - name: query_order - function: my_pkg.tools:query_order - injected_args: - tenant_id: - source: config - key: tenant_id -``` - -The supported `source` values are `config`, `sensory_memory`, and `short_term_memory`. If omitted, `source` defaults to `sensory_memory`. For `config`, `key` is read from the agent execution environment configuration. For memory sources, `key` is the memory path. +{{< hint info >}} +Generic type arguments are not part of the JVM method descriptor and must not be +included (`java.util.List`, not `java.util.List`). Boxed primitives +(`java.lang.Integer`, etc.) and bare primitives (`int`, `boolean`, ...) are both accepted. +{{< /hint >}} -See [Tool Parameter Injection]({{< ref "docs/development/calling-llms/tool_use#tool-parameter-injection" >}}) for runtime value resolution, overriding model-supplied arguments, and missing-value behavior. +`injected_args` maps tool parameter names to injection settings. Each entry must be +an object with the following optional fields. Omit `injected_args` or use `{}` if +you do not need to declare parameter injection in YAML. -For Java tools loaded from YAML, the `injected_args` entry itself marks the parameter -as framework-injected and hides it from the model-facing schema. The Java method may -use `@ToolParam(name = "...")` to provide stable parameter names, but it does not need -`@ToolParam(injected = true)` when injection is declared in YAML. -If both YAML and the tool function declare the same injected parameter, they must use -the same source and key; conflicting declarations fail during agent plan construction. +| Field | Required | Description | +|-------|----------|-------------| +| `source` | Optional | One of `config`, `sensory_memory`, or `short_term_memory`. Defaults to `sensory_memory`. | +| `key` | Optional | Configuration key or memory path. If omitted, `null`, or empty, defaults to the tool parameter name. | -Per-request values, such as a trace id written to sensory memory before tool execution, can be injected the same way: +For example, declare injection for the `tenant_id` parameter of `query_order`: ```yaml -agents: - - name: traced_order_agent - tools: - - name: lookup_order - function: my_pkg.tools:lookup_order - injected_args: - trace_id: - source: sensory_memory - key: request.trace_id +tools: + - name: query_order + function: my_pkg.tools:query_order + injected_args: + tenant_id: + source: config + key: tenant_id ``` -```yaml -# Java-loaded YAML -agents: - - name: java_order_agent - tools: - - name: queryOrder - type: java - function: com.example.OrderTools:queryOrder - parameter_types: [java.lang.String, java.lang.String] - injected_args: - tenant_id: - source: config - key: tenant_id -``` +See [Tool Parameter Injection]({{< ref "docs/development/calling-llms/tool_use#tool-parameter-injection" >}}) +for when to use injection, runtime value resolution, Java annotations, and +conflict and failure rules. -**Skills** — bundles of agent skill assets loaded from one or more sources. At least one of `paths` / `urls` / `url_sources` / `classpath` / `package` must be non-empty; multiple sources can coexist. +**Skills** — bundles of agent skill assets loaded from one or more sources. | Field | Required | Description | |-------|----------|-------------| -| `name` | yes | Skills resource name. | -| `paths` | one-of | `local` scheme: list of directories or `.zip` files. | -| `urls` | one-of | `url` scheme: list of HTTPS URLs without embedded user information, pointing to `.zip` archives. | -| `url_sources` | one-of | `url` scheme: list of `{url, sha256?, allow_insecure_http?}` archive sources without embedded user information. HTTPS is required by default; `sha256` pins the downloaded archive, and `allow_insecure_http: true` is an explicit unsafe compatibility opt-in. | -| `classpath` | one-of | `classpath` scheme (Java runtime only): list of classpath resource paths. | -| `package` | one-of | `package` scheme (Python runtime only): list of `{package, resource}` pairs. | - -`urls` entries must use HTTPS. To keep an existing plain-HTTP source, move it to `url_sources` and opt in explicitly: +| `name` | Required | Skills resource name. | +| `paths` | Optional | `local` scheme: list of directories or `.zip` files. | +| `urls` | Optional | `url` scheme: list of HTTPS URLs pointing to `.zip` archives. | +| `url_sources` | Optional | `url` scheme: list of objects with a required `url`, optional `sha256` digest, and optional `allow_insecure_http` flag (defaults to `false`). Set the flag to `true` to allow plain HTTP. | +| `classpath` | Optional | `classpath` scheme: list of Java classpath resource paths. | +| `package` | Optional | `package` scheme: list of Python `{package, resource}` pairs. | -```yaml -skills: - - name: agent_skills - url_sources: - - url: http://example.com/skills.zip - allow_insecure_http: true -``` - -Use this compatibility option only on trusted development networks; prefer HTTPS. - -URL source hosts must use DNS-compatible ASCII syntax. Names containing underscores, such as `skill_server`, are rejected; use a DNS-compatible name such as `skill-server`. Internationalized hostnames must use IDNA/Punycode rather than raw Unicode, and scoped IPv6 literals with zone identifiers are rejected. This validation is stricter than in earlier releases. +{{< hint info >}} +At least one of `paths`, `urls`, `url_sources`, `classpath`, or `package` must be non-empty. +Multiple sources can be configured together. +{{< /hint >}} ```yaml skills: @@ -259,22 +245,25 @@ skills: resource: skills/ ``` -`classpath` is accepted by the Python parser for schema parity with Java but rejected at runtime, because the Python runtime does not register a `classpath` handler. - -Remote skill archives contain trusted agent instructions and may contain executable scripts. Pin remote sources with `sha256`, or use `Skills.from_url_with_sha256(...)` (Python) / `Skills.fromUrlWithSha256(...)` (Java) in code. +Both `classpath` and `package` sources are supported by Java and Python agents. +See +[Skill Sources]({{< ref "docs/development/calling-llms/skills#skill-sources" >}}) +for URL restrictions, archive integrity and security guidance, and dependency and +cross-language runtime requirements. **Resources configured by implementation class** — `chat_model_connections`, `chat_model_setups`, `embedding_model_connections`, `embedding_model_setups`, `vector_stores`, -and `mcp_servers`. Declare the implementation and its configuration using the fields below. +and `mcp_servers`, plus `subagents` for external sub-agents. Declare the implementation +and its configuration using the fields below. This corresponds to a `ResourceDescriptor` in Java or Python; the framework creates the Resource instance at runtime. | Field | Required | Description | |-------|----------|-------------| -| `name` | yes | Resource name (referenced by other resources or actions). | -| `clazz` | yes | Either a built-in [alias](#class-aliases) (e.g. `ollama`, `openai`) or a fully-qualified class path. | -| `type` | no | Implementation language: `python` or `java`. Defaults to `python` (see [Selecting the implementation language](#selecting-the-implementation-language)). | -| *extras* | no | Additional configuration for the selected implementation (e.g. `base_url`, `endpoint`, `model`, `request_timeout`). | +| `name` | Required | Resource name (referenced by other resources or actions). | +| `clazz` | Required | Either a built-in [alias](#class-aliases) (e.g. `ollama`, `openai`) or a fully-qualified class path. | +| `type` | Optional | Implementation language: `python` or `java`. Defaults to `python` (see [Selecting the implementation language](#selecting-the-implementation-language)). | +| *extras* | Depends on the selected implementation | Additional configuration for the selected implementation (e.g. `base_url`, `endpoint`, `model`, `request_timeout`). See its provider documentation for required fields and defaults. | ```yaml chat_model_connections: @@ -290,6 +279,10 @@ mcp_servers: endpoint: http://127.0.0.1:8000/mcp ``` +**Sub-agents** are registered under `subagents` as `AGENT` Resources. Internal +sub-agents reference another Agent in the same file or already loaded into the same +environment; external sub-agents name a setup implementation. See [Declaring sub-agents](#declaring-sub-agents). + **References between Resources.** Fields such as `connection`, `prompt`, and `tools` refer to registered Resource names. The Agent's own registrations take precedence over shared Resources from the execution environment. In the [single-agent example](#declaring-a-single-agent): @@ -299,6 +292,7 @@ shared Resources from the execution environment. In the [single-agent example](# | `chat_model_setups[].connection` | a `chat_model_connections[]` name | `connection: ollama_server` | | `chat_model_setups[].prompt` | a `prompts[]` name | `prompt: review_analysis_prompt` | | `chat_model_setups[].tools` | a list of `tools[]` names | `tools: [notify_shipping_manager]` | +| `chat_model_setups[].subagents` | a list of `subagents[]` resource names | `subagents: [reviewer]` | | `embedding_model_setups[].connection` | an `embedding_model_connections[]` name | `connection: my_embedding_server` | The referenced Resource must be registered on the Agent or its execution environment. @@ -321,11 +315,11 @@ Inline action (map) fields: | Field | Required | Description | |-------|----------|-------------| -| `name` | yes | Action name (unique within the agent). | -| `function` | yes | Fully-qualified callable in the form `:`. See [Function references](#function-references). | -| `trigger_conditions` | yes | Non-empty list of strings containing exact event types or condition expressions that return a Boolean. Each string must contain at least one non-whitespace character. Built-in [event aliases](#event-aliases) (`input`, `chat_request`, ...) may be used as complete entries. All entries use OR semantics. | -| `type` | no | Implementation language: `python` or `java`. Defaults to `python` (see [Selecting the implementation language](#selecting-the-implementation-language)). | -| `config` | no | Free-form configuration map passed to the action at runtime. | +| `name` | Required | Action name (unique within the agent). | +| `function` | Required | Fully-qualified callable in the form `:`. See [Function references](#function-references). | +| `trigger_conditions` | Required | Non-empty list of strings containing exact event types or condition expressions that return a Boolean. Each string must contain at least one non-whitespace character. Built-in [event aliases](#event-aliases) (`input`, `chat_request`, ...) may be used as complete entries. All entries use OR semantics. | +| `type` | Optional | Implementation language: `python` or `java`. Defaults to `python` (see [Selecting the implementation language](#selecting-the-implementation-language)). | +| `config` | Optional | Free-form configuration map passed to the action at runtime. | For built-in event types, prefer an [event alias](#event-aliases) such as `input`. Trigger-condition matching and expression semantics are shared with annotated and programmatically registered actions; @@ -442,6 +436,107 @@ actions: Composing one agent across multiple YAML files is **not** supported. If two files loaded into the same environment declare the same agent name, or the same shared resource/action name, the second `load_yaml` call raises an error. {{< /hint >}} +### Declaring sub-agents + +Use an internal sub-agent to delegate to another Agent defined in your application, +or an external sub-agent to call an agent hosted by another system. + +#### Internal sub-agents + +To make a child Agent available to a parent Agent, add an entry to the parent's +`subagents` list. Each entry uses the following fields: + +| Field | Required | Description | +|-------|----------|-------------| +| `name` | Required | Resource name the parent uses to call this sub-agent. | +| `agent` | Required | Value of the child Agent's `name` field in the top-level `agents` list. | +| `description` | Optional | Describes when the model should delegate to this child. Defaults to an empty string; a generic description is used if none is supplied. | +| `input_schema` | Required for model-selected calls | JSON Schema written as a YAML mapping describing the child's input arguments. It can be omitted when calling the sub-agent directly from an Action. | + +The following fragment extends the [single-agent example](#declaring-a-single-agent). +Keep the parent's existing configuration and add the declarations shown below. +The child Agent's Actions and Resources are omitted. + +```yaml +agents: + - name: review_analysis_agent + # Existing Actions and Resources omitted. + subagents: + - name: reviewer + agent: review_workflow + description: Analyze product review details + input_schema: + type: object + properties: + review: {type: string} + required: [review] + chat_model_setups: + - name: review_analysis_model + # Keep the existing model configuration. + subagents: [reviewer] + + - name: review_workflow + # Define the child's Actions and Resources here. +``` + +Here, `agent: review_workflow` refers to the child defined with +`name: review_workflow`. The parent model uses the Resource name `reviewer` to call it. + +Multiple parents may reference the same child definition. Declare internal +references under each parent's `subagents`; file-level `subagents` supports only +external setup declarations. + +The referenced child Agent must be defined in the same YAML file or already loaded +into the same execution environment. Within one file, the child may appear before +or after its parent. For Agents split across files, load the child file before the +parent file. Files in a list are loaded in order, and separate loading calls use +the same rule: + +{{< tabs "Load Sub-agent Files" >}} +{{< tab "Python" >}} +```python +agents_env.load_yaml(["child.yaml", "parent.yaml"]) +``` +{{< /tab >}} +{{< tab "Java" >}} +```java +agentsEnv.loadYaml(List.of(Paths.get("child.yaml"), Paths.get("parent.yaml"))); +``` +{{< /tab >}} +{{< /tabs >}} + +An unknown child name fails loading of the referencing file; the loader does not +wait for later files. Agent names must remain unique across the environment. +Self-references and indirect cycles fail when the Agent plan is built. Register the +child's required Resources on the child itself, as described in +[Declare an internal sub-agent]({{< ref "docs/development/multi-agent/subagents#declare-an-internal-sub-agent" >}}). + +For input/output behavior, model delegation, and execution requirements, see +[Subagents]({{< ref "docs/development/multi-agent/subagents" >}}). + +#### External sub-agents + +Use `clazz` to name a custom sub-agent setup implementation. Its additional fields +are passed to that implementation as configuration: + +```yaml +subagents: + - name: remote_reviewer + clazz: my_app.agents.RemoteReviewSetup + type: python + endpoint: https://example.com/review + description: Review code using the remote service + input_schema: '{"type":"object","properties":{"code":{"type":"string"}},"required":["code"]}' +``` + +This example assumes `RemoteReviewSetup` accepts `endpoint`, `description`, and a +JSON string `input_schema`. Use `type: java` and a Java class name when loading an +external Java setup from a Java job. YAML external sub-agent declarations currently +require the setup language to match the loader; an omitted `type` defaults to Python. +Reference `remote_reviewer` in the chat model's `subagents` list to enable model delegation. +See [Declare an external sub-agent]({{< ref "docs/development/multi-agent/subagents#declare-an-external-sub-agent" >}}) +for the implementation contract. + ## Loading and Running YAML describes an Agent but is not submitted to Flink by itself. Create a Java Flink @@ -559,9 +654,12 @@ function: com.example.Outer$Inner:processInput ### Selecting the implementation language -Every resource, tool, and action accepts an optional `type:` field with values `python` or `java`. When omitted, the default is **`python`** — both the Python loader and the Java loader treat a missing `type:` as `python`. +Resources configured by implementation class, Tools, and Actions accept an optional +`type:` field with values `python` or `java`. When omitted, the default is **`python`** +in both loaders. Prompts, Skills, and internal sub-agent references do not accept this field. -This means a YAML file loaded by the **Java loader** must set `type: java` explicitly on every resource, tool, and action whose implementation is in Java. Omitting `type:` on a Java resource in a Java-loaded YAML will make the loader try to build a Python-wrapping descriptor — usually not what you want. +For components that accept `type`, set `type: java` explicitly when selecting a Java +implementation, even when loading the YAML from a Java job. When `type:` resolves to the **opposite** language of the loader, the loader builds a cross-language descriptor that delegates to the other-language implementation — see [Cross-language agents](#cross-language-agents). @@ -592,6 +690,10 @@ For custom event types defined in your code, write the event's full `EVENT_TYPE` #### Class aliases +Class aliases cover built-in providers. For custom providers, set `clazz:` to the +fully-qualified class name in Java or class path in Python; custom alias registration +is not supported. + Aliases for `clazz:` are keyed on resource type **and** implementation language. The same alias (e.g. `ollama`) resolves to a different class for `chat_model_connections` vs `chat_model_setups`, and for `python` vs `java`. Common chat-model aliases: @@ -632,8 +734,6 @@ Vector-store aliases: The full alias tables live in `flink_agents.api.yaml.aliases` (Python) and `org.apache.flink.agents.api.yaml.Aliases` (Java). -If `clazz:` is not a known alias, the loader passes it through as-is — write a fully-qualified class path (e.g. `flink_agents.integrations.chat_models.ollama_chat_model.OllamaChatModelSetup`) for providers you've added yourself. - ### Cross-language agents Setting `type:` to the **opposite** language of the loader bridges Python and Java pieces in the same agent. The loader resolves the alias against the cross-language bucket and wraps the resource in the appropriate cross-language proxy. @@ -694,7 +794,7 @@ in a Java Flink job and use each component's `type:` to select its implementatio Application payloads must use the supported [Java and Python data types]({{< ref "docs/development/create-agent/choose-apis-and-languages#data-passed-between-java-and-python" >}}). -Not every resource type is cross-language. Currently `actions`, `tools`, `chat_model_connections`, `chat_model_setups`, `embedding_model_connections`, `embedding_model_setups`, and `vector_stores` support `type:` on the opposite language; others (e.g. `mcp_servers`) do not. +Not every resource type is cross-language. Currently `actions`, `tools`, `chat_model_connections`, `chat_model_setups`, `embedding_model_connections`, `embedding_model_setups`, and `vector_stores` support `type:` on the opposite language; others (e.g. `mcp_servers` and external `subagents`) do not. ## YAML API Specification @@ -705,31 +805,7 @@ The schema benefits both humans and machines: - It defines an interface contract, so LLMs and external systems can interact with Flink Agents declaratively. - It is language-neutral, so it doubles as the compatibility contract between the Python and the Java loader. -### Python and Java mapping - -When loading a YAML file, both runtimes parse it into typed in-memory representations and validate it against them using the framework's native validation: - -- The **Python loader** parses YAML into Pydantic `BaseModel`s declared in `flink_agents.api.yaml.specs`. Pydantic enforces the schema (required fields, value types, `extra="forbid"` on unknown keys, mutually-exclusive `text`/`messages` on prompts, ...). -- The **Java loader** parses YAML into POJOs declared in `org.apache.flink.agents.api.yaml.spec`, validated by Jackson with equivalent rules. - -### Consistency guarantees - -Because Pydantic models are easier to author and evolve than raw JSON Schema, the **ground truth** is the Pydantic spec — the checked-in `docs/yaml-schema.json` is exported from it. After changing the specs, regenerate the schema with: - -```bash -python -m flink_agents.api.yaml.specs > docs/yaml-schema.json -``` - -The coding-agent skill under `dev/agent-skills/` bundles a copy of the schema so it can work offline, so refresh that copy in the same change. `python3 tools/check-skill-schema.py` reports whether it is current and prints the two edits needed when it is not. - -Continuous tests then verify cross-runtime consistency: - -- the JSON Schema exported by the Pydantic specs matches the checked-in `docs/yaml-schema.json`; -- the Java POJOs match the checked-in `docs/yaml-schema.json` in property names, required fields, and `additionalProperties` strictness; -- the Pydantic specs stay aligned with the Python `Agent` API; -- the Java POJOs stay aligned with the Java `Agent` API. - -These checks keep the YAML structure aligned across loaders. Schema validation does +Schema validation checks the document structure. It does not verify that referenced classes/functions are installed, that a component supports the selected implementation language, or that its provider is reachable. Before running the job, follow [Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}}) diff --git a/docs/content/docs/development/multi-agent/subagents.md b/docs/content/docs/development/multi-agent/subagents.md index 0871c67991..aa0893203a 100644 --- a/docs/content/docs/development/multi-agent/subagents.md +++ b/docs/content/docs/development/multi-agent/subagents.md @@ -28,6 +28,11 @@ A sub-agent handles a task delegated by a parent agent. An internal sub-agent ru an agent workflow in the same job. An external sub-agent calls an agent service through a `SubagentSetup` implementation. +To declare parent and child workflows in YAML, see +[Declaring sub-agents]({{< ref "docs/development/create-agent/yaml#declaring-sub-agents" >}}). +YAML supports references to internal child Agents in the same file and descriptors +for external sub-agents. + ## Declare an internal sub-agent Use `InternalSubagentResourceDescriptorBuilder` to attach a description and input