wenjin272 commented on code in PR #1264: URL: https://github.com/apache/flink-agents/pull/1264#discussion_r4292101634
########## 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: Could we align Java’s `BaseDecisionModelSetup` with the Python implementation by providing the common `connection`/`model` configuration, connection resolution in `open()`, and default `decide()` delegation? This would let Java providers implement only the Connection without repeating Setup logic. `PythonDecisionModelSetup` could continue overriding these methods to delegate to Python, following the existing ChatModel pattern. ########## 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: Could we define a provider-independent request model for Predicate, Choice, and Score questions, with shared types such as `ScoreLevel(label, description)`? Each provider Connection should translate this model into its native format and explicitly reject unsupported semantics. This would let applications and the multi-provider E2E test reuse the same `DecisionRequest` without provider-specific branches, such as the current OpenAI Choice conversion in the test Agent. ########## 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: Could we introduce `PredicateQuestion`, `ChoiceQuestion`, and `ScoreQuestion` types in Java, matching Python’s model and validation rules? The `Question.predicate/choice/score` factories could still return these concrete types. Currently, inputs such as `Question.choice(42, null)` are accepted in Java and only rejected when passed to Python. ########## 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: Could we prefer annotations/decorators in the DecisionModel E2E tests and representative usage tests? This would cover resource discovery and registration alongside actual invocation. Please also add the corresponding Java annotations, while keeping focused coverage for programmatic registration via `addResource` / `add_resource`. ########## 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: Could we accept `Map<String, Object> modelParams` in Java and `**model_kwargs` in Python, following the ChatModel pattern? Setup already passes `model_kwargs`, but Connection only accepts `model`. This would allow provider-specific options without changing the shared signature. The bridge should preserve these parameters, and providers should reject unsupported ones rather than silently ignore them. -- 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]
