joeyutong commented on code in PR #1264: URL: https://github.com/apache/flink-agents/pull/1264#discussion_r4307193625
########## api/src/main/java/org/apache/flink/agents/api/decision/BaseDecisionModelSetup.java: ########## @@ -0,0 +1,40 @@ +/* + * 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.decision; + +import org.apache.flink.agents.api.resource.Resource; +import org.apache.flink.agents.api.resource.ResourceContext; +import org.apache.flink.agents.api.resource.ResourceDescriptor; +import org.apache.flink.agents.api.resource.ResourceType; + +/** A configured decision model. Each call supplies its own evidence and questions. */ +public abstract class BaseDecisionModelSetup extends Resource { + protected BaseDecisionModelSetup( + ResourceDescriptor descriptor, ResourceContext resourceContext) { + super(descriptor, resourceContext); + } + + @Override + public ResourceType getResourceType() { + return ResourceType.DECISION_MODEL; + } + + /** Evaluate named independent questions, preserving typed answers and provider errors. */ + public abstract DecisionResult decide(DecisionRequest request); Review Comment: Added common `connection`/`model` configuration, resolution in `open()`, and default `decide()` delegation. Java providers now only need a Connection; the Python wrapper keeps its overrides. Tests cover shared lifecycle and Java Setup → Python Connection calls. Updated in [c8dd8d7a](https://github.com/apache/flink-agents/commit/c8dd8d7a20575e8eba8f26cbe9409fe2a2f80b14). ########## python/flink_agents/api/decision_models/decision_model.py: ########## @@ -0,0 +1,123 @@ +################################################################################ +# 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. +################################################################################ +"""Resources for making typed decisions from application-supplied evidence.""" + +from abc import ABC, abstractmethod +from typing import Any, Dict + +from pydantic import ConfigDict, Field, field_validator +from typing_extensions import override + +from flink_agents.api.decision_models.decision_request import DecisionRequest +from flink_agents.api.decision_models.decision_result import DecisionResult +from flink_agents.api.resource import Resource, ResourceType + + +class BaseDecisionModelConnection(Resource, ABC): + """Connection shared by decision model setups. + + A connection owns the provider client and its authentication and transport + settings. Use open and close to manage the client lifecycle. + """ + + model_config = ConfigDict(extra="forbid", allow_inf_nan=False) + + @classmethod + @override + def resource_type(cls) -> ResourceType: + """Return the connection resource type.""" + return ResourceType.DECISION_MODEL_CONNECTION + + @abstractmethod + def decide(self, request: DecisionRequest, *, model: str) -> DecisionResult: Review Comment: Added per-call `request_params` / `requestParams` throughout Setup, Connection, Event/Action, and the Java-to-Python bridge. Parameters reach the SDK without leaking across calls; conflicting core fields and unsupported SDK options raise errors. Updated in [c8dd8d7a](https://github.com/apache/flink-agents/commit/c8dd8d7a20575e8eba8f26cbe9409fe2a2f80b14). ########## python/flink_agents/api/decision_models/decision_request.py: ########## @@ -0,0 +1,104 @@ +################################################################################ +# 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. +################################################################################ +"""Inputs and questions for constrained model decisions.""" + +from typing import Annotated, Any, Dict, List, Literal + +from pydantic import BaseModel, ConfigDict, Field + +DecisionInput = str | Dict[str, Any] | List[Any] + + +class _DecisionData(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class PredicateQuestion(_DecisionData): + """Ask for the probability that the supplied condition is true. + + Instructions are required and must define the condition, as text or structured + JSON. Optional criteria supplement the instructions with true/false outcome + descriptions for providers that support them; other providers ignore criteria. + Either outcome may be omitted or left undescribed with None. A false criterion + describes when the condition does not hold; the answer still estimates the + probability of true. + """ + + type: Literal["predicate"] = "predicate" + instructions: DecisionInput + criteria: Dict[Literal["true", "false"], DecisionInput | None] | None = None + + +class ChoiceQuestion(_DecisionData): + """Choose among named, unordered alternatives. + + Criteria use the selected provider's native format: a map of identifiers to + descriptions, or a list of option objects. Map identifiers are strings; + option values may be strings or booleans. Values preserve their type and + string case in the answer. Map descriptions may be text, + a JSON object or array, or None. Option objects follow the provider's schema. + Formats are not converted into one another. The SDK and service determine Review Comment: Unified the public model across providers: - Predicate, Choice, and Score share one request/answer contract, with lightweight JSON criteria. - Answers retain probability distributions and the original Score level descriptions. - Common validation is centralized; adapters handle provider-specific conversion and restrictions. Examples and multi-provider tests now reuse the same request without provider-specific rewrites. Java/Python models, tests, and the guide are aligned. Updated in [c8dd8d7a](https://github.com/apache/flink-agents/commit/c8dd8d7a20575e8eba8f26cbe9409fe2a2f80b14). ########## api/src/main/java/org/apache/flink/agents/api/decision/DecisionRequest.java: ########## @@ -0,0 +1,110 @@ +/* + * 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.decision; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** + * Shared evidence and named independent questions. Input and instructions may be text, a JSON + * object or a JSON array. Strings containing JSON remain strings. Criteria retain the selected + * provider's native format; they are not translated between providers. + */ +public final class DecisionRequest implements Serializable { + private final Object input; + private final Map<String, Question> questions; + + @JsonCreator + public DecisionRequest( + @JsonProperty("input") Object input, + @JsonProperty("questions") Map<String, Question> questions) { + this.input = Objects.requireNonNull(input, "input"); + this.questions = Collections.unmodifiableMap(new LinkedHashMap<>(questions)); + } + + public Object getInput() { + return input; + } + + public Map<String, Question> getQuestions() { + return questions; + } + + /** A predicate, choice or score question, validated by the Python DecisionModel API on call. */ + @JsonInclude(JsonInclude.Include.NON_NULL) + public static final class Question implements Serializable { Review Comment: Added concrete Java `PredicateQuestion`, `ChoiceQuestion`, and `ScoreQuestion` types with early common validation. Existing factories return these types. Tests cover invalid construction, typed JSON round trips, and Java/Python snapshots. Updated in [c8dd8d7a](https://github.com/apache/flink-agents/commit/c8dd8d7a20575e8eba8f26cbe9409fe2a2f80b14). ########## e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/DecisionModelCrossLanguageTest.java: ########## @@ -0,0 +1,382 @@ +/* + * 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.resource.test; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.flink.agents.api.AgentsExecutionEnvironment; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.EventType; +import org.apache.flink.agents.api.InputEvent; +import org.apache.flink.agents.api.OutputEvent; +import org.apache.flink.agents.api.agents.Agent; +import org.apache.flink.agents.api.agents.AgentExecutionOptions; +import org.apache.flink.agents.api.annotation.Action; +import org.apache.flink.agents.api.context.RunnerContext; +import org.apache.flink.agents.api.decision.BaseDecisionModelConnection; +import org.apache.flink.agents.api.decision.BaseDecisionModelSetup; +import org.apache.flink.agents.api.decision.DecisionRequest; +import org.apache.flink.agents.api.decision.DecisionRequest.Question; +import org.apache.flink.agents.api.decision.DecisionResult; +import org.apache.flink.agents.api.decision.DecisionResult.ChoiceAnswer; +import org.apache.flink.agents.api.decision.DecisionResult.PredicateAnswer; +import org.apache.flink.agents.api.decision.DecisionResult.RefusalAnswer; +import org.apache.flink.agents.api.decision.DecisionResult.ScoreAnswer; +import org.apache.flink.agents.api.event.DecisionRequestEvent; +import org.apache.flink.agents.api.event.DecisionResponseEvent; +import org.apache.flink.agents.api.resource.PythonResourceDescriptor; +import org.apache.flink.agents.api.resource.ResourceType; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.DecisionModelAction; +import org.apache.flink.agents.plan.utils.CancellationUtils; +import org.apache.flink.agents.runtime.PythonMCPResourceDiscovery; +import org.apache.flink.agents.runtime.ResourceCache; +import org.apache.flink.agents.runtime.context.RunnerContextImpl; +import org.apache.flink.agents.runtime.metrics.FlinkAgentsMetricGroupImpl; +import org.apache.flink.agents.runtime.python.utils.PythonInterpreterManager; +import org.apache.flink.agents.runtime.python.utils.PythonResourceAdapterImpl; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.util.CloseableIterator; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; +import pemja.core.PythonInterpreter; +import pemja.core.PythonInterpreterConfig; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CancellationException; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Exercises actual Pemja calls and a Flink job without requiring a remote model service. */ +public class DecisionModelCrossLanguageTest { + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final String FIXTURE = + "flink_agents.runtime.tests.decision_model_bridge_fixture"; + + public static class DecisionAgent extends Agent { + public DecisionAgent() { + addResource( Review Comment: Added Java annotations and moved representative invocation/E2E Agents to annotations or decorators, covering discovery and registration alongside invocation. Focused programmatic registration tests remain; documentation uses complete registered Agent examples. Updated in [c8dd8d7a](https://github.com/apache/flink-agents/commit/c8dd8d7a20575e8eba8f26cbe9409fe2a2f80b14). -- 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]
