wenjin272 commented on code in PR #1264: URL: https://github.com/apache/flink-agents/pull/1264#discussion_r4309789204
########## python/flink_agents/integrations/decision_models/openai_decision_model.py: ########## @@ -0,0 +1,266 @@ +################################################################################ +# 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. +################################################################################ +"""Decision resources backed by the official OpenAI Python SDK.""" + +import json +from collections.abc import Mapping +from typing import Any, ClassVar + +from openai import OpenAI +from openai.types import Decision as OpenAIDecision +from openai.types.decision import Answer as OpenAIAnswer +from pydantic import Field, JsonValue, PrivateAttr +from typing_extensions import override + +from flink_agents.api.decision_models import ( + BaseDecisionModelConnection, + BaseDecisionModelSetup, + ChoiceAnswer, + ChoiceQuestion, + DecisionAnswer, + DecisionQuestion, + DecisionRequest, + DecisionResult, + PredicateAnswer, + PredicateQuestion, + RefusalAnswer, + ScoreAnswer, + ScoreQuestion, +) +from flink_agents.api.decision_models.decision_model import _validate_model_name +from flink_agents.integrations.decision_models.decision_model_utils import ( + create_result, + prepare_request_params, +) + + +def _text(value: JsonValue, field: str) -> str: + if value is None: + msg = f"OpenAI {field} is required." + raise ValueError(msg) + return ( + value + if isinstance(value, str) + else json.dumps(value, ensure_ascii=False, allow_nan=False) + ) + + +def _description(value: JsonValue, field: str) -> dict[str, str]: + return {} if value is None else {"description": _text(value, field)} + + +def _questions(request: DecisionRequest) -> list[dict[str, Any]]: + max_questions = OpenAIDecisionModelConnection._MAX_QUESTIONS + if len(request.questions) > max_questions: + msg = f"OpenAI supports at most {max_questions} questions per decision request." + raise ValueError(msg) + questions = [] + for name, question in request.questions.items(): + entry: dict[str, Any] = { + "name": name, + "type": question.type, + "instructions": _text( + question.instructions, f"question {name!r} instructions" + ), + } + if isinstance(question, PredicateQuestion): + criteria = question.criteria + if criteria is not None and ( + criteria.true is not None or criteria.false is not None + ): + msg = f"OpenAI question {name!r} does not support predicate criteria." + raise ValueError(msg) Review Comment: Could we translate Predicate criteria into OpenAI instructions instead of rejecting them, so the same request remains usable across providers? [Pydantic AI](https://github.com/pydantic/pydantic-ai/blob/main/pydantic_ai_slim/pydantic_ai/models/openai_decisions.py#L263-L296) combines the question with `yes`/`no` meanings and JSON-encodes it as instructions; [LiteLLM](https://docs.litellm.ai/docs/decisions#limits) also folds noul criteria into instructions. A documented deterministic conversion in this Connection, with tests for either or both outcome descriptions, would keep this adaptation out of application code. ########## api/src/main/java/org/apache/flink/agents/api/decision/BaseDecisionModelSetup.java: ########## @@ -0,0 +1,124 @@ +/* + * 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; +import org.apache.flink.annotation.Internal; + +import javax.annotation.Nullable; + +import java.util.Map; +import java.util.Objects; + +/** + * A decision model configured with a connection and a model name. + * + * <p>Named connections are resolved through the resource context in {@link #open()}. Directly + * supplied connections remain the caller's responsibility. Closing a setup does not close a shared + * connection. Each call supplies its own evidence, questions, and provider request parameters. + */ +public class BaseDecisionModelSetup extends Resource { + @Nullable private final String connectionName; + private final String model; + @Nullable private BaseDecisionModelConnection connection; + + /** + * Configure a decision model using the descriptor's {@code connection} and {@code model}. + * + * @param descriptor configuration containing a connection resource name or connection instance + * and a non-blank provider model name + * @param resourceContext resource lookup context, required for a named connection + */ + public BaseDecisionModelSetup( + ResourceDescriptor descriptor, @Nullable ResourceContext resourceContext) { + super(Objects.requireNonNull(descriptor, "descriptor"), resourceContext); + Object configuredConnection = descriptor.getArgument("connection"); + if (configuredConnection instanceof BaseDecisionModelConnection) { + this.connectionName = null; + this.connection = (BaseDecisionModelConnection) configuredConnection; + } else { + this.connectionName = requireNonBlankString(configuredConnection, "connection"); + } + this.model = requireNonBlankString(descriptor.getArgument("model"), "model"); Review Comment: [P2] Could we avoid revalidating Python setup configuration from the raw descriptor in the Java wrapper? A Python subclass can validly declare `model: str = "luna"` and be constructed with only `connection="shared"`. I reproduced this through Pemja: Python construction succeeds, but `PythonDecisionModelSetup` then calls this constructor and fails because the original descriptor has no explicit `model`. The same applies to a default connection. Please retain validation for native Java setups while allowing the Python delegate's resolved defaults, and cover this through the bridge. ########## python/flink_agents/integrations/decision_models/decision_model_utils.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. +################################################################################ +"""Shared utilities for decision model integrations.""" + +from collections.abc import Collection, Mapping, Sequence +from decimal import Decimal +from typing import Any + +from pydantic import JsonValue + +from flink_agents.api.decision_models import ( + ChoiceAnswer, + ChoiceQuestion, + DecisionAnswer, + DecisionRequest, + DecisionResult, + PredicateAnswer, + PredicateQuestion, + RefusalAnswer, + ScoreAnswer, + ScoreQuestion, + TokenUsage, +) +from flink_agents.api.decision_models.decision_model import _validate_request_params + + +def prepare_request_params( + request_params: Mapping[str, JsonValue] | None, + *, + reserved_fields: Collection[str], +) -> dict[str, Any]: + """Copy call parameters and protect fields supplied by the framework.""" + params = _validate_request_params(request_params) + conflicts = params.keys() & set(reserved_fields) + extra_body = params.get("extra_body") + if isinstance(extra_body, dict): + conflicts |= extra_body.keys() & set(reserved_fields) + if conflicts: + msg = f"request_params cannot override reserved fields: {', '.join(sorted(conflicts))}." + raise ValueError(msg) + return params + + +def same_json_value(left: JsonValue, right: JsonValue) -> bool: + """Compare JSON content without conflating booleans with numeric values.""" + if isinstance(left, dict): + return ( + isinstance(right, dict) + and left.keys() == right.keys() + and all(same_json_value(value, right[key]) for key, value in left.items()) + ) + if isinstance(left, list): + return ( + isinstance(right, list) + and len(left) == len(right) + and all(same_json_value(a, b) for a, b in zip(left, right, strict=True)) + ) + return isinstance(left, bool) == isinstance(right, bool) and left == right + + +def _rounding_bounds(value: float) -> tuple[Decimal, Decimal]: + displayed = Decimal(str(value)) + places = max(2, -int(displayed.as_tuple().exponent)) + half_unit = Decimal(10) ** -places / 2 + return displayed - half_unit, displayed + half_unit Review Comment: [P2] Could we add a numerical tolerance floor for both distribution and expected-score validation? The digits in `str(float)` do not account for floating-point normalization error: the valid normalized distribution `[0.44962867851783656, 0.5503713214821633]` is rejected despite summing to within about `1e-16` of one. A score computed directly as `sum(i * p)` can also be rejected. I reproduced both through the real OpenAI and TypeSafe SDKs with MockTransport: all four cases fail here and pass on the previous revision. This turns successful provider responses into retries/failures, including Java calls through the Python bridge. Please preserve the reported values and add regression coverage for floating-point normalization and weighted scores. -- 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]
