xintongsong commented on code in PR #670: URL: https://github.com/apache/flink-agents/pull/670#discussion_r3246599346
########## python/flink_agents/api/yaml/specs.py: ########## @@ -0,0 +1,208 @@ +################################################################################ +# 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. +################################################################################# +"""Pydantic schema for the declarative YAML API. + +The models in this module define the file-level wire format. Pydantic +validation is the ground truth for the JSON Schema published in +docs/yaml-schema.json. +""" + +import json +import sys +from enum import Enum +from typing import Any, Dict, List, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +Language = Literal["python", "java"] +"""Implementation language of a YAML-declared resource, action, or tool.""" + + +class DescriptorSpec(BaseModel): + """Schema for any ResourceDescriptor-backed resource. + + Required: ``name`` and ``clazz``. ``type`` selects the implementation + language (``"python"`` or ``"java"``; ``None`` means Python). All + remaining fields are forwarded verbatim to ``ResourceDescriptor`` as + kwargs (or as the Java wrapper's kwargs when ``type: java``); the + forwarding and language-aware wrapping is done by ``loader._build_descriptor``. + """ + + model_config = ConfigDict(extra="allow") + + name: str + clazz: str + type: Language | None = None + + +class MessageRole(str, Enum): + """Role of a message in a chat conversation.""" + + SYSTEM = "system" + USER = "user" + ASSISTANT = "assistant" + TOOL = "tool" + + +class PromptMessage(BaseModel): + """One message in a multi-turn prompt template.""" + + model_config = ConfigDict(extra="forbid") + + role: MessageRole = MessageRole.USER + content: str + + +class PromptSpec(BaseModel): + """Declarative prompt: either a single ``text`` template or a list of + role-tagged ``messages``. Exactly one of the two fields must be set. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + text: str | None = None + messages: List[PromptMessage] | None = None + + @model_validator(mode="after") + def _require_exactly_one(self) -> "PromptSpec": + # Treat empty string / empty list as "unset" so that ``text: ""`` and + # ``messages: []`` are rejected rather than silently producing a + # nonsense empty prompt at load time. + if bool(self.text) == bool(self.messages): + msg = "prompt must define exactly one non-empty 'text' or 'messages'" + raise ValueError(msg) + return self + + +class ToolSpec(BaseModel): + """Points ``function:`` at a module attribute that is a callable tool. + + When ``function:`` is omitted, the loader falls back to + ``<namespace>.<name>``. + + ``parameter_types`` is required when ``type: java`` (Java method + signatures vary across tools). + """ + + model_config = ConfigDict(extra="forbid") + + name: str + function: str | None = None + type: Language | None = None + parameter_types: List[str] | None = None Review Comment: This needs a more detailed description. What are the required formats/values of the parameter type strings. ########## python/flink_agents/api/yaml/specs.py: ########## @@ -0,0 +1,208 @@ +################################################################################ +# 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. +################################################################################# +"""Pydantic schema for the declarative YAML API. + +The models in this module define the file-level wire format. Pydantic +validation is the ground truth for the JSON Schema published in +docs/yaml-schema.json. +""" + +import json +import sys +from enum import Enum +from typing import Any, Dict, List, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +Language = Literal["python", "java"] +"""Implementation language of a YAML-declared resource, action, or tool.""" + + +class DescriptorSpec(BaseModel): + """Schema for any ResourceDescriptor-backed resource. + + Required: ``name`` and ``clazz``. ``type`` selects the implementation + language (``"python"`` or ``"java"``; ``None`` means Python). All + remaining fields are forwarded verbatim to ``ResourceDescriptor`` as + kwargs (or as the Java wrapper's kwargs when ``type: java``); the + forwarding and language-aware wrapping is done by ``loader._build_descriptor``. + """ + + model_config = ConfigDict(extra="allow") + + name: str + clazz: str + type: Language | None = None + + +class MessageRole(str, Enum): + """Role of a message in a chat conversation.""" + + SYSTEM = "system" + USER = "user" + ASSISTANT = "assistant" + TOOL = "tool" + + +class PromptMessage(BaseModel): + """One message in a multi-turn prompt template.""" + + model_config = ConfigDict(extra="forbid") + + role: MessageRole = MessageRole.USER + content: str + + +class PromptSpec(BaseModel): + """Declarative prompt: either a single ``text`` template or a list of + role-tagged ``messages``. Exactly one of the two fields must be set. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + text: str | None = None + messages: List[PromptMessage] | None = None + + @model_validator(mode="after") + def _require_exactly_one(self) -> "PromptSpec": + # Treat empty string / empty list as "unset" so that ``text: ""`` and + # ``messages: []`` are rejected rather than silently producing a + # nonsense empty prompt at load time. + if bool(self.text) == bool(self.messages): + msg = "prompt must define exactly one non-empty 'text' or 'messages'" + raise ValueError(msg) + return self + + +class ToolSpec(BaseModel): + """Points ``function:`` at a module attribute that is a callable tool. + + When ``function:`` is omitted, the loader falls back to + ``<namespace>.<name>``. + + ``parameter_types`` is required when ``type: java`` (Java method + signatures vary across tools). + """ + + model_config = ConfigDict(extra="forbid") + + name: str + function: str | None = None + type: Language | None = None + parameter_types: List[str] | None = None + + +class SkillsSpec(BaseModel): + """Declarative Skills resource pointing at one or more skill source + directories on the local filesystem. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + paths: List[str] + + +class ActionSpec(BaseModel): + """An action references a user function and the event types it listens to. + + When ``function:`` is omitted, the loader falls back to + ``<namespace>.<name>``. Action signatures are fixed + (``(Event, RunnerContext)``), so there is no ``parameter_types`` + knob — Python doesn't need it, and the Java action signature is + determined by the action contract. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + function: str | None = None + on: List[str] = Field(..., min_length=1) + config: Dict[str, Any] | None = None + type: Language | None = None + + +class AgentSpec(BaseModel): + """One agent inside a YAML file's ``agents:`` list. + + Holds the agent's own resources and actions. Resources/actions declared + at the file level (siblings of ``agents:``) are merged in by the loader. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + description: str | None = None + namespace: str | None = None Review Comment: The name `namespace` is confusing. Maybe something like `default_func_path`. Anyway, the semantic of this needs well description in documents. ########## python/flink_agents/api/yaml/tests/test_loader.py: ########## @@ -0,0 +1,648 @@ +################################################################################ +# 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. +################################################################################# +from pathlib import Path + +import pytest + +from flink_agents.api.agents.agent import Agent +from flink_agents.api.chat_message import MessageRole +from flink_agents.api.events.chat_event import ChatResponseEvent +from flink_agents.api.events.event import InputEvent +from flink_agents.api.execution_environment import AgentsExecutionEnvironment +from flink_agents.api.function import JavaFunction, PythonFunction +from flink_agents.api.prompts.prompt import LocalPrompt +from flink_agents.api.resource import ResourceDescriptor, ResourceName, ResourceType +from flink_agents.api.skills import Skills +from flink_agents.api.tools.function_tool import FunctionTool +from flink_agents.api.yaml.loader import build_agents, load_yaml, resolve_function +from flink_agents.api.yaml.tests.fixtures import loader_targets + +_FIXTURES = Path(__file__).parent / "fixtures" + +_TARGETS_MODULE = "flink_agents.api.yaml.tests.fixtures.loader_targets" + + +def test_resolve_function_with_fqn_only() -> None: + func = resolve_function( + name="anything", function=f"{_TARGETS_MODULE}.increment", namespace=None + ) + assert isinstance(func, PythonFunction) + assert func.module == _TARGETS_MODULE + assert func.qualname == "increment" + # still callable + assert func.as_callable() is loader_targets.increment + + +def test_resolve_function_with_namespace_and_bare_name() -> None: + func = resolve_function(name="x", function="increment", namespace=_TARGETS_MODULE) + assert isinstance(func, PythonFunction) + assert func.module == _TARGETS_MODULE + assert func.qualname == "increment" + assert func.as_callable() is loader_targets.increment + + +def test_resolve_function_with_namespace_and_fqn() -> None: + func = resolve_function( + name="x", + function=f"{_TARGETS_MODULE}.decrement", + namespace="ignored.module", + ) + assert isinstance(func, PythonFunction) + assert func.module == _TARGETS_MODULE + assert func.qualname == "decrement" + assert func.as_callable() is loader_targets.decrement + + +def test_resolve_function_namespace_fallback() -> None: + func = resolve_function(name="increment", function=None, namespace=_TARGETS_MODULE) + assert isinstance(func, PythonFunction) + assert func.module == _TARGETS_MODULE + assert func.qualname == "increment" + assert func.as_callable() is loader_targets.increment + + +def test_resolve_function_bare_name_without_namespace_fails() -> None: + with pytest.raises(ValueError, match="namespace"): + resolve_function(name="x", function="increment", namespace=None) + + +def test_resolve_function_no_function_no_namespace_fails() -> None: + with pytest.raises(ValueError): + resolve_function(name="x", function=None, namespace=None) + + +def test_resolve_function_python_rejects_leading_dot() -> None: + # ``function: ".foo"`` resolves to a qualified name with no module + # component; we must reject it at YAML resolution time rather than + # letting ``importlib.import_module("")`` fail much later without + # any YAML context. + with pytest.raises(ValueError, match="python function"): + resolve_function(name="x", function=".foo", namespace=None) + + +def test_resolve_function_python_rejects_trailing_dot() -> None: + # Mirror of the leading-dot case: an empty attribute name would build + # a ``PythonFunction`` with ``qualname=""`` that explodes on first use. + with pytest.raises(ValueError, match="python function"): + resolve_function(name="x", function="pkg.mod.", namespace=None) + + +def test_resolve_function_python_rejects_empty_namespace_with_bare_name() -> None: + # An empty (rather than missing) ``namespace`` resolves to a leading-dot + # qualified name. The bare-name-without-namespace branch only rejects + # ``namespace is None``, so the empty-string case must be caught at the + # post-partition guard. + with pytest.raises(ValueError, match="python function"): + resolve_function(name="x", function="foo", namespace="") + + +def test_resolve_function_missing_target_raises_importerror() -> None: + # PythonFunction loads lazily; trigger the import via as_callable(). + func = resolve_function( + name="x", + function=f"{_TARGETS_MODULE}.does_not_exist", + namespace=None, + ) + with pytest.raises((ImportError, AttributeError)): + func.as_callable() + + +def test_build_agents_treats_unquoted_on_as_string_not_bool(tmp_path: Path) -> None: + yaml_text = ( + "agents:\n" + " - name: a\n" + f" namespace: {_TARGETS_MODULE}\n" + " actions:\n" + " - name: increment\n" + " on: [input]\n" + ) + p = tmp_path / "on_keyword.yaml" + p.write_text(yaml_text) + agents, _, _ = build_agents(p) + assert "a" in agents + assert "increment" in agents["a"].actions + + +def test_build_agents_rejects_duplicate_agent_within_file(tmp_path: Path) -> None: + yaml_text = ( + "agents:\n" + " - name: dup\n" + f" namespace: {_TARGETS_MODULE}\n" + " actions:\n" + " - name: increment\n" + " on: [input]\n" + " - name: dup\n" + f" namespace: {_TARGETS_MODULE}\n" + " actions:\n" + " - name: decrement\n" + " on: [input]\n" + ) + p = tmp_path / "dup.yaml" + p.write_text(yaml_text) + with pytest.raises(ValueError, match="dup"): + build_agents(p) + + +def test_build_agents_from_single_agent_yaml() -> None: + agents, shared_resources, shared_actions = build_agents( + _FIXTURES / "single_agent.yaml" + ) + assert list(agents) == ["incrementer"] + agent = agents["incrementer"] + assert isinstance(agent, Agent) + assert "increment" in agent.actions + events, func, config = agent.actions["increment"] + assert events == [InputEvent.EVENT_TYPE] + assert isinstance(func, PythonFunction) + assert func.qualname == "increment" + assert config is None + assert shared_resources == {t: {} for t in shared_resources} + assert shared_actions == {} + + +def test_build_agents_resolves_event_alias_and_clazz_alias() -> None: + agents, _, _ = build_agents(_FIXTURES / "with_descriptors.yaml") + agent = agents["chat_agent"] + + inc_events, _, _ = agent.actions["increment"] + dec_events, _, _ = agent.actions["decrement"] + assert inc_events == [InputEvent.EVENT_TYPE] + assert dec_events == [ChatResponseEvent.EVENT_TYPE] + + conn = agent.resources[ResourceType.CHAT_MODEL_CONNECTION]["ollama_conn"] + assert isinstance(conn, ResourceDescriptor) + expected_module, _, expected_class = ( + ResourceName.ChatModel.OLLAMA_CONNECTION.rpartition(".") + ) + assert conn.target_module == expected_module + assert conn.target_clazz == expected_class + assert conn.arguments == { + "base_url": "http://localhost:11434";, + "request_timeout": 30, + } + + +def test_build_agents_loads_tools_and_prompts() -> None: + agents, _, _ = build_agents(_FIXTURES / "with_tools_and_prompts.yaml") + agent = agents["tool_agent"] + + tool = agent.resources[ResourceType.TOOL]["notify"] + assert isinstance(tool, FunctionTool) + assert isinstance(tool.func, PythonFunction) + assert tool.func.qualname == "notify" + + text_prompt = agent.resources[ResourceType.PROMPT]["text_prompt"] + assert isinstance(text_prompt, LocalPrompt) + assert text_prompt.template == "hello {name}" + + msg_prompt = agent.resources[ResourceType.PROMPT]["messages_prompt"] + assert isinstance(msg_prompt, LocalPrompt) + assert len(msg_prompt.template) == 2 + assert msg_prompt.template[0].role == MessageRole.SYSTEM + assert msg_prompt.template[1].content == "{q}" + + +def test_build_agents_handles_shared_resources_and_actions() -> None: + agents, shared_resources, shared_actions = build_agents( + _FIXTURES / "with_shared.yaml" + ) + + # shared resources surfaced to caller + assert "shared_conn" in shared_resources[ResourceType.CHAT_MODEL_CONNECTION] + # shared actions stored as ActionSpec for cross-agent reference resolution + assert "shared_inc" in shared_actions + + # both a1 and a2 own a copy of shared_inc after caller-side merge? + # NO — build_agents only handles in-file. The merge happens in load_yaml. + # Here we assert build_agents leaves string refs *unresolved* for the caller: + a1 = agents["a1"] + a2 = agents["a2"] + assert "shared_inc" not in a1.actions # not yet merged in + assert "own_dec" in a1.actions + assert "shared_inc" not in a2.actions + + +def test_load_yaml_registers_single_agent_on_env() -> None: + env = AgentsExecutionEnvironment.get_execution_environment() + load_yaml(env, _FIXTURES / "single_agent.yaml") + assert "incrementer" in env._agents + + +def test_load_yaml_registers_multiple_agents() -> None: + env = AgentsExecutionEnvironment.get_execution_environment() + load_yaml(env, _FIXTURES / "multi_agent.yaml") + assert set(env._agents.keys()) == {"a1", "a2"} + + +def test_load_yaml_merges_shared_action_into_agents() -> None: + env = AgentsExecutionEnvironment.get_execution_environment() + load_yaml(env, _FIXTURES / "with_shared.yaml") + a1 = env._agents["a1"] + a2 = env._agents["a2"] + assert "shared_inc" in a1.actions + assert "shared_inc" in a2.actions + events_a1, func_a1, _ = a1.actions["shared_inc"] + events_a2, func_a2, _ = a2.actions["shared_inc"] + assert events_a1 == [InputEvent.EVENT_TYPE] + assert events_a2 == [InputEvent.EVENT_TYPE] + assert isinstance(func_a1, PythonFunction) + assert func_a1.qualname == "increment" + assert isinstance(func_a2, PythonFunction) + assert func_a2.qualname == "increment" + + +def test_load_yaml_registers_shared_resources_on_env() -> None: + env = AgentsExecutionEnvironment.get_execution_environment() + load_yaml(env, _FIXTURES / "with_shared.yaml") + assert "shared_conn" in env.resources[ResourceType.CHAT_MODEL_CONNECTION] + + +def test_load_yaml_string_ref_to_missing_shared_action_errors() -> None: + env = AgentsExecutionEnvironment.get_execution_environment() + bad = _FIXTURES / "bad_missing_shared_action.yaml" Review Comment: Should not write temporal contents in the source directory. ########## python/flink_agents/api/yaml/loader.py: ########## @@ -0,0 +1,496 @@ +################################################################################ +# 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 loader: parse a YAML document and register agents on an execution +environment. +""" + +import re +from pathlib import Path +from typing import TYPE_CHECKING, Any, Dict, List, Tuple + +if TYPE_CHECKING: + from flink_agents.api.execution_environment import AgentsExecutionEnvironment + +import yaml + +from flink_agents.api.agents.agent import Agent +from flink_agents.api.chat_message import ChatMessage, MessageRole +from flink_agents.api.function import Function, JavaFunction, PythonFunction +from flink_agents.api.prompts.prompt import Prompt +from flink_agents.api.resource import ResourceDescriptor, ResourceType +from flink_agents.api.skills import Skills +from flink_agents.api.tools.function_tool import FunctionTool +from flink_agents.api.yaml.aliases import ( + JAVA_WRAPPER_CLAZZ, + resolve_clazz, + resolve_event_type, +) +from flink_agents.api.yaml.specs import ( + ActionSpec, + AgentSpec, + DescriptorSpec, + Language, + PromptSpec, + SkillsSpec, + ToolSpec, + YamlAgentsDocument, +) + + +class _FlinkAgentsYamlLoader(yaml.SafeLoader): + """SafeLoader that recognizes only ``true``/``false`` as booleans. + + PyYAML follows YAML 1.1 by default, where ``on``/``off``/``yes``/``no`` + also coerce to booleans. ``on:`` is the natural way to declare an + action's event listeners in this API, so we strip those legacy + keywords from the bool resolver. + """ + + +_FlinkAgentsYamlLoader.yaml_implicit_resolvers = { + key: [(tag, regexp) for tag, regexp in resolvers if tag != "tag:yaml.org,2002:bool"] + for key, resolvers in yaml.SafeLoader.yaml_implicit_resolvers.items() +} +_FlinkAgentsYamlLoader.add_implicit_resolver( + "tag:yaml.org,2002:bool", + re.compile(r"^(?:true|True|TRUE|false|False|FALSE)$"), + list("tTfF"), +) + +# Default Java parameter types for an action. Action methods in +# flink-agents always have signature (Event, RunnerContext). +_JAVA_ACTION_PARAMETER_TYPES: list[str] = [ + "org.apache.flink.agents.api.Event", + "org.apache.flink.agents.api.context.RunnerContext", +] + +_DESCRIPTOR_TYPES: Dict[str, ResourceType] = { + "chat_model_connections": ResourceType.CHAT_MODEL_CONNECTION, + "chat_model_setups": ResourceType.CHAT_MODEL, + "embedding_model_connections": ResourceType.EMBEDDING_MODEL_CONNECTION, + "embedding_model_setups": ResourceType.EMBEDDING_MODEL, + "vector_stores": ResourceType.VECTOR_STORE, + "mcp_servers": ResourceType.MCP_SERVER, +} + + +def resolve_function( + *, + name: str, + function: str | None, + namespace: str | None, + language: Language | None = None, + parameter_types: List[str] | None = None, +) -> PythonFunction | JavaFunction: + """Resolve a YAML function reference to a flink-agents Function. + + Returns a ``PythonFunction`` when ``language`` is ``"python"`` (or + None — the default). Returns a ``JavaFunction`` when ``language`` + is ``"java"``; the qualified name is split at the last ``.`` into + a class ``qualname`` and ``method_name``. Java parameter types must + be passed in by the caller (actions use a fixed signature; tools + vary per method). + + Resolution rules for the qualified name: + - ``function`` given and contains ``.``: use verbatim. + - ``function`` given without ``.`` and ``namespace`` given: + ``{namespace}.{function}``. + - ``function`` given without ``.`` and no ``namespace``: error. + - ``function`` missing and ``namespace`` given: ``{namespace}.{name}``. + - ``function`` missing and no ``namespace``: error. + """ + if function is not None: + if "." in function: + qualified = function + elif namespace is not None: + qualified = f"{namespace}.{function}" + else: + msg = ( + f"Action/tool {name!r}: 'function' is a bare name " + f"{function!r} but no 'namespace' is set to qualify it." + ) + raise ValueError(msg) + elif namespace is not None: + qualified = f"{namespace}.{name}" + else: + msg = ( + f"Action/tool {name!r}: must set 'function' (fully-qualified) " + "or set 'namespace' on the agent." + ) + raise ValueError(msg) + + if language == "java": + class_qualname, _, method_name = qualified.rpartition(".") + if not class_qualname or not method_name: + msg = ( + f"Action/tool {name!r}: java function {qualified!r} " + "must be of the form '<package>.<Class>.<method>'." + ) + raise ValueError(msg) + return JavaFunction( + qualname=class_qualname, + method_name=method_name, + parameter_types=parameter_types or [], + ) + + module_name, _, attr_name = qualified.rpartition(".") + if not module_name or not attr_name: + msg = ( + f"Action/tool {name!r}: python function {qualified!r} " + "must be of the form '<module>.<callable>'." + ) + raise ValueError(msg) + return PythonFunction(module=module_name, qualname=attr_name) + + +def _load_document(path: Path | str) -> YamlAgentsDocument: + text = Path(path).read_text() + raw = yaml.load(text, Loader=_FlinkAgentsYamlLoader) + if raw is None: + msg = f"YAML file {path} is empty" + raise ValueError(msg) + return YamlAgentsDocument.model_validate(raw) + + +def _build_descriptor( + spec: DescriptorSpec, resource_type: ResourceType +) -> ResourceDescriptor: + kwargs = dict(spec.model_extra or {}) + if spec.type == "java": + if resource_type not in JAVA_WRAPPER_CLAZZ: + msg = ( + f"Resource {spec.name!r}: type='java' is not supported " + f"for {resource_type.value} (no Python-side Java wrapper)." + ) + raise ValueError(msg) + java_fqn = resolve_clazz(spec.clazz, resource_type, "java") + wrapper_clazz = JAVA_WRAPPER_CLAZZ[resource_type] + return ResourceDescriptor(clazz=wrapper_clazz, java_clazz=java_fqn, **kwargs) + python_fqn = resolve_clazz(spec.clazz, resource_type, "python") + return ResourceDescriptor(clazz=python_fqn, **kwargs) + + +def _add_descriptors_to_agent( + agent: Agent, attr_name: str, descriptors: list[DescriptorSpec] +) -> None: + resource_type = _DESCRIPTOR_TYPES[attr_name] + for spec in descriptors: + agent.add_resource( + spec.name, resource_type, _build_descriptor(spec, resource_type) + ) + + +def _resolve_action_function(action: ActionSpec, namespace: str | None) -> Function: + parameter_types = _JAVA_ACTION_PARAMETER_TYPES if action.type == "java" else None + return resolve_function( + name=action.name, + function=action.function, + namespace=namespace, + language=action.type, + parameter_types=parameter_types, + ) + + +def _add_action_to_agent( + agent: Agent, action: ActionSpec, namespace: str | None +) -> None: + func = _resolve_action_function(action, namespace) + events = [resolve_event_type(e) for e in action.on] + config = action.config or {} + agent.add_action(action.name, events, func, **config) + + +def _build_tool(spec: ToolSpec, namespace: str | None) -> FunctionTool: + if spec.type == "java" and spec.parameter_types is None: + msg = f"Tool {spec.name!r}: java tools must declare 'parameter_types' in YAML." + raise ValueError(msg) + func = resolve_function( + name=spec.name, + function=spec.function, + namespace=namespace, + language=spec.type, + parameter_types=spec.parameter_types, + ) + return FunctionTool(func=func) + + +def _build_prompt(spec: PromptSpec) -> Prompt: + if spec.text is not None: + return Prompt.from_text(spec.text) + messages = [ + ChatMessage(role=MessageRole(m.role.value), content=m.content) + for m in (spec.messages or []) + ] + return Prompt.from_messages(messages) + + +def _build_skills(spec: SkillsSpec) -> Skills: + return Skills(paths=list(spec.paths)) + + +def _build_agent(agent_spec: AgentSpec) -> Agent: + agent = Agent() + for attr in _DESCRIPTOR_TYPES: + descriptors = getattr(agent_spec, attr) + _add_descriptors_to_agent(agent, attr, descriptors) + for tool_spec in agent_spec.tools: + agent.add_resource( + tool_spec.name, + ResourceType.TOOL, + _build_tool(tool_spec, agent_spec.namespace), + ) + for prompt_spec in agent_spec.prompts: + agent.add_resource( + prompt_spec.name, ResourceType.PROMPT, _build_prompt(prompt_spec) + ) + for skills_spec in agent_spec.skills: + agent.add_resource( + skills_spec.name, ResourceType.SKILLS, _build_skills(skills_spec) + ) + for action in agent_spec.actions: + if isinstance(action, str): + continue # shared-action references handled by caller + _add_action_to_agent(agent, action, agent_spec.namespace) + return agent + + +def _build_in_file_state( + path: Path | str, +) -> Tuple[ + Dict[str, Agent], + Dict[ResourceType, Dict[str, Any]], + Dict[str, ActionSpec], + Dict[str, AgentSpec], + YamlAgentsDocument, +]: + """Parse one YAML file, perform in-file duplicate detection, and build + the in-memory state without touching any execution environment. + + Returns: + agents: name -> Agent + shared_resources: resource_type -> name -> descriptor/resource + shared_actions: name -> ActionSpec (file-level, for cross-agent reference) + agent_specs: name -> AgentSpec (kept so callers can resolve string + action references back to the originating spec). + + Both :func:`build_agents` and :func:`load_yaml` go through this helper + so the in-file rules (duplicate detection, build order) are defined in + exactly one place. + """ + doc = _load_document(path) + 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) + agent_specs[spec.name] = spec + agents[spec.name] = _build_agent(spec) + + shared_resources: Dict[ResourceType, Dict[str, Any]] = {t: {} for t in ResourceType} + for attr, resource_type in _DESCRIPTOR_TYPES.items(): + for spec in getattr(doc, attr): + if spec.name in shared_resources[resource_type]: + msg = f"Duplicate shared resource name {spec.name!r} in {path}" + raise ValueError(msg) + shared_resources[resource_type][spec.name] = _build_descriptor( + spec, resource_type + ) + for tool_spec in doc.tools: + if tool_spec.name in shared_resources[ResourceType.TOOL]: + msg = f"Duplicate shared tool name {tool_spec.name!r} in {path}" + raise ValueError(msg) + # Shared tools at the file level have no namespace; require an + # explicit function. + shared_resources[ResourceType.TOOL][tool_spec.name] = _build_tool( + tool_spec, namespace=None + ) + for prompt_spec in doc.prompts: + if prompt_spec.name in shared_resources[ResourceType.PROMPT]: + msg = f"Duplicate shared prompt name {prompt_spec.name!r} in {path}" + raise ValueError(msg) + shared_resources[ResourceType.PROMPT][prompt_spec.name] = _build_prompt( + prompt_spec + ) + for skills_spec in doc.skills: + if skills_spec.name in shared_resources[ResourceType.SKILLS]: + msg = f"Duplicate shared skills name {skills_spec.name!r} in {path}" + raise ValueError(msg) + shared_resources[ResourceType.SKILLS][skills_spec.name] = _build_skills( + skills_spec + ) + + shared_actions: Dict[str, ActionSpec] = {} + for action_spec in doc.actions: + if action_spec.name in shared_actions: + msg = f"Duplicate shared action name {action_spec.name!r} in {path}" + raise ValueError(msg) + shared_actions[action_spec.name] = action_spec + + return agents, shared_resources, shared_actions, agent_specs, doc + + +def _assert_has_python_implementation( Review Comment: Is this necessary? What if I implemented an agent in yaml + java, but need to use it in a pyflink job? ########## python/flink_agents/api/yaml/specs.py: ########## @@ -0,0 +1,208 @@ +################################################################################ +# 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. +################################################################################# +"""Pydantic schema for the declarative YAML API. + +The models in this module define the file-level wire format. Pydantic +validation is the ground truth for the JSON Schema published in +docs/yaml-schema.json. +""" + +import json +import sys +from enum import Enum +from typing import Any, Dict, List, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +Language = Literal["python", "java"] +"""Implementation language of a YAML-declared resource, action, or tool.""" + + +class DescriptorSpec(BaseModel): + """Schema for any ResourceDescriptor-backed resource. + + Required: ``name`` and ``clazz``. ``type`` selects the implementation + language (``"python"`` or ``"java"``; ``None`` means Python). All + remaining fields are forwarded verbatim to ``ResourceDescriptor`` as + kwargs (or as the Java wrapper's kwargs when ``type: java``); the + forwarding and language-aware wrapping is done by ``loader._build_descriptor``. + """ + + model_config = ConfigDict(extra="allow") + + name: str + clazz: str + type: Language | None = None + + +class MessageRole(str, Enum): + """Role of a message in a chat conversation.""" + + SYSTEM = "system" + USER = "user" + ASSISTANT = "assistant" + TOOL = "tool" + + +class PromptMessage(BaseModel): + """One message in a multi-turn prompt template.""" + + model_config = ConfigDict(extra="forbid") + + role: MessageRole = MessageRole.USER + content: str + + +class PromptSpec(BaseModel): + """Declarative prompt: either a single ``text`` template or a list of + role-tagged ``messages``. Exactly one of the two fields must be set. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + text: str | None = None + messages: List[PromptMessage] | None = None + + @model_validator(mode="after") + def _require_exactly_one(self) -> "PromptSpec": + # Treat empty string / empty list as "unset" so that ``text: ""`` and + # ``messages: []`` are rejected rather than silently producing a + # nonsense empty prompt at load time. + if bool(self.text) == bool(self.messages): + msg = "prompt must define exactly one non-empty 'text' or 'messages'" + raise ValueError(msg) + return self + + +class ToolSpec(BaseModel): + """Points ``function:`` at a module attribute that is a callable tool. + + When ``function:`` is omitted, the loader falls back to + ``<namespace>.<name>``. + + ``parameter_types`` is required when ``type: java`` (Java method + signatures vary across tools). + """ + + model_config = ConfigDict(extra="forbid") + + name: str + function: str | None = None + type: Language | None = None + parameter_types: List[str] | None = None + + +class SkillsSpec(BaseModel): + """Declarative Skills resource pointing at one or more skill source + directories on the local filesystem. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + paths: List[str] + + +class ActionSpec(BaseModel): + """An action references a user function and the event types it listens to. + + When ``function:`` is omitted, the loader falls back to + ``<namespace>.<name>``. Action signatures are fixed + (``(Event, RunnerContext)``), so there is no ``parameter_types`` + knob — Python doesn't need it, and the Java action signature is + determined by the action contract. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + function: str | None = None + on: List[str] = Field(..., min_length=1) + config: Dict[str, Any] | None = None + type: Language | None = None + + +class AgentSpec(BaseModel): + """One agent inside a YAML file's ``agents:`` list. + + Holds the agent's own resources and actions. Resources/actions declared + at the file level (siblings of ``agents:``) are merged in by the loader. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + description: str | None = None + namespace: str | None = None Review Comment: I think this default needs more careful design, otherwise it may introduce more confusion than convenience. I'd suggest to make this a follow-up and exclude from this PR. ########## python/flink_agents/api/yaml/loader.py: ########## @@ -0,0 +1,496 @@ +################################################################################ +# 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 loader: parse a YAML document and register agents on an execution +environment. +""" + +import re +from pathlib import Path +from typing import TYPE_CHECKING, Any, Dict, List, Tuple + +if TYPE_CHECKING: + from flink_agents.api.execution_environment import AgentsExecutionEnvironment + +import yaml + +from flink_agents.api.agents.agent import Agent +from flink_agents.api.chat_message import ChatMessage, MessageRole +from flink_agents.api.function import Function, JavaFunction, PythonFunction +from flink_agents.api.prompts.prompt import Prompt +from flink_agents.api.resource import ResourceDescriptor, ResourceType +from flink_agents.api.skills import Skills +from flink_agents.api.tools.function_tool import FunctionTool +from flink_agents.api.yaml.aliases import ( + JAVA_WRAPPER_CLAZZ, + resolve_clazz, + resolve_event_type, +) +from flink_agents.api.yaml.specs import ( + ActionSpec, + AgentSpec, + DescriptorSpec, + Language, + PromptSpec, + SkillsSpec, + ToolSpec, + YamlAgentsDocument, +) + + +class _FlinkAgentsYamlLoader(yaml.SafeLoader): + """SafeLoader that recognizes only ``true``/``false`` as booleans. + + PyYAML follows YAML 1.1 by default, where ``on``/``off``/``yes``/``no`` + also coerce to booleans. ``on:`` is the natural way to declare an Review Comment: Can we change to another keyword? ########## python/flink_agents/api/yaml/loader.py: ########## @@ -0,0 +1,496 @@ +################################################################################ +# 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 loader: parse a YAML document and register agents on an execution +environment. +""" + +import re +from pathlib import Path +from typing import TYPE_CHECKING, Any, Dict, List, Tuple + +if TYPE_CHECKING: + from flink_agents.api.execution_environment import AgentsExecutionEnvironment + +import yaml + +from flink_agents.api.agents.agent import Agent +from flink_agents.api.chat_message import ChatMessage, MessageRole +from flink_agents.api.function import Function, JavaFunction, PythonFunction +from flink_agents.api.prompts.prompt import Prompt +from flink_agents.api.resource import ResourceDescriptor, ResourceType +from flink_agents.api.skills import Skills +from flink_agents.api.tools.function_tool import FunctionTool +from flink_agents.api.yaml.aliases import ( + JAVA_WRAPPER_CLAZZ, + resolve_clazz, + resolve_event_type, +) +from flink_agents.api.yaml.specs import ( + ActionSpec, + AgentSpec, + DescriptorSpec, + Language, + PromptSpec, + SkillsSpec, + ToolSpec, + YamlAgentsDocument, +) + + +class _FlinkAgentsYamlLoader(yaml.SafeLoader): + """SafeLoader that recognizes only ``true``/``false`` as booleans. + + PyYAML follows YAML 1.1 by default, where ``on``/``off``/``yes``/``no`` + also coerce to booleans. ``on:`` is the natural way to declare an + action's event listeners in this API, so we strip those legacy + keywords from the bool resolver. + """ + + +_FlinkAgentsYamlLoader.yaml_implicit_resolvers = { + key: [(tag, regexp) for tag, regexp in resolvers if tag != "tag:yaml.org,2002:bool"] + for key, resolvers in yaml.SafeLoader.yaml_implicit_resolvers.items() +} +_FlinkAgentsYamlLoader.add_implicit_resolver( + "tag:yaml.org,2002:bool", + re.compile(r"^(?:true|True|TRUE|false|False|FALSE)$"), + list("tTfF"), +) + +# Default Java parameter types for an action. Action methods in +# flink-agents always have signature (Event, RunnerContext). +_JAVA_ACTION_PARAMETER_TYPES: list[str] = [ + "org.apache.flink.agents.api.Event", + "org.apache.flink.agents.api.context.RunnerContext", +] + +_DESCRIPTOR_TYPES: Dict[str, ResourceType] = { + "chat_model_connections": ResourceType.CHAT_MODEL_CONNECTION, + "chat_model_setups": ResourceType.CHAT_MODEL, + "embedding_model_connections": ResourceType.EMBEDDING_MODEL_CONNECTION, + "embedding_model_setups": ResourceType.EMBEDDING_MODEL, + "vector_stores": ResourceType.VECTOR_STORE, + "mcp_servers": ResourceType.MCP_SERVER, +} + + +def resolve_function( + *, + name: str, + function: str | None, + namespace: str | None, + language: Language | None = None, + parameter_types: List[str] | None = None, +) -> PythonFunction | JavaFunction: + """Resolve a YAML function reference to a flink-agents Function. + + Returns a ``PythonFunction`` when ``language`` is ``"python"`` (or + None — the default). Returns a ``JavaFunction`` when ``language`` + is ``"java"``; the qualified name is split at the last ``.`` into + a class ``qualname`` and ``method_name``. Java parameter types must + be passed in by the caller (actions use a fixed signature; tools + vary per method). + + Resolution rules for the qualified name: + - ``function`` given and contains ``.``: use verbatim. + - ``function`` given without ``.`` and ``namespace`` given: + ``{namespace}.{function}``. + - ``function`` given without ``.`` and no ``namespace``: error. + - ``function`` missing and ``namespace`` given: ``{namespace}.{name}``. + - ``function`` missing and no ``namespace``: error. + """ + if function is not None: + if "." in function: + qualified = function Review Comment: What if `namespace` is `foo` and the function is `foo.bar.Func`? ########## python/flink_agents/api/yaml/specs.py: ########## @@ -0,0 +1,208 @@ +################################################################################ +# 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. +################################################################################# +"""Pydantic schema for the declarative YAML API. + +The models in this module define the file-level wire format. Pydantic +validation is the ground truth for the JSON Schema published in +docs/yaml-schema.json. +""" + +import json +import sys +from enum import Enum +from typing import Any, Dict, List, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +Language = Literal["python", "java"] +"""Implementation language of a YAML-declared resource, action, or tool.""" + + +class DescriptorSpec(BaseModel): + """Schema for any ResourceDescriptor-backed resource. + + Required: ``name`` and ``clazz``. ``type`` selects the implementation + language (``"python"`` or ``"java"``; ``None`` means Python). All + remaining fields are forwarded verbatim to ``ResourceDescriptor`` as + kwargs (or as the Java wrapper's kwargs when ``type: java``); the + forwarding and language-aware wrapping is done by ``loader._build_descriptor``. + """ + + model_config = ConfigDict(extra="allow") + + name: str + clazz: str + type: Language | None = None + + +class MessageRole(str, Enum): + """Role of a message in a chat conversation.""" + + SYSTEM = "system" + USER = "user" + ASSISTANT = "assistant" + TOOL = "tool" + + +class PromptMessage(BaseModel): + """One message in a multi-turn prompt template.""" + + model_config = ConfigDict(extra="forbid") + + role: MessageRole = MessageRole.USER + content: str + + +class PromptSpec(BaseModel): + """Declarative prompt: either a single ``text`` template or a list of + role-tagged ``messages``. Exactly one of the two fields must be set. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + text: str | None = None + messages: List[PromptMessage] | None = None + + @model_validator(mode="after") + def _require_exactly_one(self) -> "PromptSpec": + # Treat empty string / empty list as "unset" so that ``text: ""`` and + # ``messages: []`` are rejected rather than silently producing a + # nonsense empty prompt at load time. + if bool(self.text) == bool(self.messages): + msg = "prompt must define exactly one non-empty 'text' or 'messages'" + raise ValueError(msg) + return self + + +class ToolSpec(BaseModel): + """Points ``function:`` at a module attribute that is a callable tool. + + When ``function:`` is omitted, the loader falls back to + ``<namespace>.<name>``. + + ``parameter_types`` is required when ``type: java`` (Java method + signatures vary across tools). + """ + + model_config = ConfigDict(extra="forbid") + + name: str + function: str | None = None + type: Language | None = None + parameter_types: List[str] | None = None + + +class SkillsSpec(BaseModel): + """Declarative Skills resource pointing at one or more skill source + directories on the local filesystem. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + paths: List[str] + + +class ActionSpec(BaseModel): + """An action references a user function and the event types it listens to. + + When ``function:`` is omitted, the loader falls back to + ``<namespace>.<name>``. Action signatures are fixed + (``(Event, RunnerContext)``), so there is no ``parameter_types`` + knob — Python doesn't need it, and the Java action signature is + determined by the action contract. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + function: str | None = None + on: List[str] = Field(..., min_length=1) + config: Dict[str, Any] | None = None + type: Language | None = None + + +class AgentSpec(BaseModel): + """One agent inside a YAML file's ``agents:`` list. + + Holds the agent's own resources and actions. Resources/actions declared + at the file level (siblings of ``agents:``) are merged in by the loader. + """ + + model_config = ConfigDict(extra="forbid") + + name: str + description: str | None = None + namespace: str | None = None Review Comment: Do we need separate defaults for python and java? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
