wenjin272 commented on code in PR #1330:
URL: https://github.com/apache/flink-agents/pull/1330#discussion_r4440632447
##########
plan/src/main/java/org/apache/flink/agents/plan/routing/LlmJudgeRoutingExecutor.java:
##########
@@ -195,9 +195,10 @@ private static List<String> candidateNames(RoutingContext
context) {
* The rendering anchor is the router's default candidate (or the first
candidate) — where
* abstains resolve, and in practice the workload-level prompt shared by
the candidates. If the
* anchor can't be resolved or binds no prompt, the raw message list is
used unchanged.
+ * Cancellation and interruption are not failures of the anchor: they
propagate.
*/
- private static List<ChatMessage> effectiveJudgeMessages(
- RoutingContext context, RunnerContext ctx) {
+ static List<ChatMessage> effectiveJudgeMessages(RoutingContext context,
RunnerContext ctx)
Review Comment:
Could we extract these shared input-processing methods into a separate
helper? Message preparation, context budgeting, and content extraction are now
used by both strategies. This would avoid having `DecisionModelRoutingExecutor`
depend on another executor’s internal methods.
##########
examples/src/main/java/org/apache/flink/agents/examples/openai/OpenAiDecisionRoutingExample.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.examples.openai;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
+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.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.model.routing.ModelRouter;
+import org.apache.flink.agents.api.chat.model.routing.Strategies;
+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.DecisionResult;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceName;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Routes each request with a decision model instead of an LLM judge: the
OpenAI Decisions API
+ * answers one choice question over the candidates and reports a probability
per candidate. The
+ * chosen candidate answers unless its probability is below the configured
minimum, in which case
+ * the default model does. Requires {@code OPENAI_API_KEY} in the environment.
+ *
+ * <p>The decision model is a native Java provider: {@link
OpenAiDecisionsConnection} implements
+ * {@link BaseDecisionModelConnection} over the Decisions endpoint, so the job
needs no Python
+ * environment on the TaskManagers. The output lines carry the model that
answered, the decision
+ * source and the routing score, read from the {@code model_routing} attribute
of the response.
+ */
+public class OpenAiDecisionRoutingExample {
+
+ private static ResourceDescriptor openAiModel(String model) {
+ return ResourceDescriptor.Builder.newBuilder(
+ ResourceName.ChatModel.OPENAI_COMPLETIONS_SETUP)
+ .addInitialArgument("connection", "openaiConnection")
+ .addInitialArgument("model", model)
+ .build();
+ }
+
+ /** Runs the example pipeline. */
+ public static void main(String[] args) throws Exception {
+ String apiKey = System.getenv("OPENAI_API_KEY");
+ if (apiKey == null || apiKey.isBlank()) {
+ throw new IllegalStateException("Set OPENAI_API_KEY in the
environment to run.");
+ }
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ AgentsExecutionEnvironment agentsEnv =
+ AgentsExecutionEnvironment.getExecutionEnvironment(env);
+ agentsEnv.getConfig().set(AgentExecutionOptions.NUM_ASYNC_THREADS, 2);
+
+ // OpenAI connection shared by both candidate chat models.
+ agentsEnv.addResource(
+ "openaiConnection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+ ResourceDescriptor.Builder.newBuilder(
+
ResourceName.ChatModel.OPENAI_COMPLETIONS_CONNECTION)
+ .addInitialArgument("api_key", apiKey)
+ .build());
+ agentsEnv
+ .addResource("small", ResourceType.CHAT_MODEL,
openAiModel("gpt-4o-mini"))
+ .addResource("big", ResourceType.CHAT_MODEL,
openAiModel("gpt-4o"));
+
+ // The decision model: a connection to the OpenAI Decisions API and a
setup naming the
+ // decision model. Registered like any other decision model resource.
+ agentsEnv.addResource(
+ "openaiDecisions",
+ ResourceType.DECISION_MODEL_CONNECTION,
+
ResourceDescriptor.Builder.newBuilder(OpenAiDecisionsConnection.class.getName())
+ .addInitialArgument("api_key", apiKey)
+ .build());
+ agentsEnv.addResource(
+ "dm",
+ ResourceType.DECISION_MODEL,
+
ResourceDescriptor.Builder.newBuilder(BaseDecisionModelSetup.class.getName())
+ .addInitialArgument("connection", "openaiDecisions")
+ .addInitialArgument("model", "gpt-6-luna")
+ .build());
+
+ // A router whose decision comes from the decision model; the
descriptions are the choice
+ // criteria. A chosen candidate below 0.6 probability abstains to the
default model.
+ agentsEnv.addResource(
+ "router",
+ ResourceType.MODEL_ROUTER,
+ ModelRouter.of("small", "big")
+ .describe("small", "fast and cheap: chit-chat,
lookups, short factual asks")
+ .describe("big", "expensive: code, SQL, math,
multi-step reasoning")
+
.strategy(Strategies.decision("dm").withMinProbability(0.6))
+ .defaultModel("small")
+ .fallback(true)
+ .build());
+
+ // An easy request (-> small) and a coding request (-> big).
+ DataStream<String> requestStream =
+ env.fromData(
+ "Hi, how are you today?",
+ "Write SQL to select all active users from the users
table.");
+ DataStream<Object> resultStream =
+ agentsEnv
+ .fromDataStream(requestStream, value -> value)
+ .apply(new DecisionRoutingAgent())
+ .toDataStream();
+ resultStream.print();
+ agentsEnv.execute("OpenAI Decision Routing Example");
+ }
+
+ /** Sends each input to the router and reports which model answered and
why. */
+ public static class DecisionRoutingAgent extends Agent {
+
+ @Action(EventType.InputEvent)
+ public static void processInput(InputEvent event, RunnerContext ctx) {
+ ctx.sendEvent(
+ new ChatRequestEvent(
+ "router", List.of(ChatMessage.user((String)
event.getInput()))));
+ }
+
+ @Action(EventType.ChatResponseEvent)
+ public static void processChatResponse(ChatResponseEvent event,
RunnerContext ctx) {
+ if (event.isFailed()) {
+ ctx.sendEvent(new OutputEvent("FAILED: " + event.getError()));
+ return;
+ }
+ Map<?, ?> routing = (Map<?, ?>) event.getAttr("model_routing");
+ String answer = event.getResponse().getText().replace('\n', ' ');
+ ctx.sendEvent(
+ new OutputEvent(
+ String.format(
+ "final_model=%s decision_source=%s
score=%s :: %s",
+ routing.get("final_model"),
+ routing.get("decision_source"),
+ routing.get("score"),
+ answer.length() > 80
+ ? answer.substring(0, 80) + "..."
+ : answer)));
+ }
+ }
+
+ /**
+ * A native Java decision model connection over the OpenAI Decisions API
({@code POST
+ * /v1/decisions}). Supports choice and predicate questions, which is what
routing and most
+ * gating use; score questions are rejected. Descriptor arguments: {@code
api_key} (required)
+ * and {@code base_url} (default {@code https://api.openai.com/v1}).
+ */
+ public static class OpenAiDecisionsConnection extends
BaseDecisionModelConnection {
Review Comment:
Could we reuse the existing Python OpenAI DecisionModel provider in this
example? Implementing a Java HTTP adapter here adds provider-specific
request/response handling to a routing example. If we need native Java support,
I think it would be better to deliver and test it separately as a proper
integration.
--
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]
