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..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; @@ -159,7 +162,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 = @@ -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/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/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/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 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 dec3c52664..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: @@ -196,7 +219,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) @@ -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_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" 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 {