wenjin272 commented on code in PR #1271:
URL: https://github.com/apache/flink-agents/pull/1271#discussion_r4309886831
##########
docs/content/docs/get-started/quickstart/create-agent/single-agent.md:
##########
@@ -187,97 +225,101 @@ class ReviewAnalysisAgent(Agent):
*/
public class ReviewAnalysisAgent extends Agent {
- private static final ObjectMapper MAPPER = new ObjectMapper();
+private static final ObjectMapper MAPPER = new ObjectMapper();
- @Prompt
- public static org.apache.flink.agents.api.prompt.Prompt
reviewAnalysisPrompt() {
- return REVIEW_ANALYSIS_PROMPT;
- }
+@Prompt
+public static org.apache.flink.agents.api.prompt.Prompt reviewAnalysisPrompt()
{
+ return REVIEW_ANALYSIS_PROMPT;
+}
- @ChatModelSetup
- public static ResourceDescriptor reviewAnalysisModel() {
- return
ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
- .addInitialArgument("connection", "ollamaChatModelConnection")
- .addInitialArgument("model", "qwen3:8b")
- .addInitialArgument("prompt", "reviewAnalysisPrompt")
- .addInitialArgument("tools",
Collections.singletonList("notifyShippingManager"))
- .addInitialArgument("extract_reasoning", true)
- .build();
- }
+@ChatModelSetup
+public static ResourceDescriptor reviewAnalysisModel() {
+ return
ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "ollamaChatModelConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .addInitialArgument("prompt", "reviewAnalysisPrompt")
+ .addInitialArgument("tools",
Collections.singletonList("notifyShippingManager"))
+ .addInitialArgument("extract_reasoning", true)
+ .build();
+}
- /**
- * Tool for notifying the shipping manager when product received a
negative review due to
- * shipping damage.
- *
- * @param id The id of the product that received a negative review due to
shipping damage
- * @param review The negative review content
- */
- @Tool(
- description =
- "Notify the shipping manager when product received a
negative review due to shipping damage.")
- public static void notifyShippingManager(
- @ToolParam(name = "id") String id, @ToolParam(name = "review")
String review) {
- CustomTypesAndResources.notifyShippingManager(id, review);
- }
+/**
+ * Tool for notifying the shipping manager when product received a negative
review due to
+ * shipping damage.
+ *
+ * @param id The id of the product that received a negative review due to
shipping damage
+ * @param review The negative review content
+ */
+@Tool(
+ description =
+ "Notify the shipping manager when product received a negative
review due to shipping damage.")
+public static void notifyShippingManager(
+ @ToolParam(name = "id") String id, @ToolParam(name = "review") String
review) {
+ CustomTypesAndResources.notifyShippingManager(id, review);
+}
- /** Process input event and send chat request for review analysis. */
- @Action(EventType.InputEvent)
- public static void processInput(Event event, RunnerContext ctx) throws
Exception {
- InputEvent inputEvent = InputEvent.fromEvent(event);
- String input = (String) inputEvent.getInput();
- MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
- CustomTypesAndResources.ProductReview inputObj =
- MAPPER.readValue(input,
CustomTypesAndResources.ProductReview.class);
-
- ctx.getShortTermMemory().set("id", inputObj.getId());
-
- String content =
- String.format(
- "{\n" + "\"id\": %s,\n" + "\"review\": \"%s\"\n" + "}",
- inputObj.getId(), inputObj.getReview());
- ChatMessage msg = new ChatMessage(MessageRole.USER, "");
-
- ctx.sendEvent(
- new ChatRequestEvent(
- "reviewAnalysisModel", List.of(msg), Map.of("input",
content), null));
- }
+/** Process input event and send chat request for review analysis. */
+@Action(EventType.InputEvent)
+public static void processInput(Event event, RunnerContext ctx) throws
Exception {
+ InputEvent inputEvent = (InputEvent) event;
+ String input = (String) inputEvent.getInput();
+ MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ CustomTypesAndResources.ProductReview inputObj =
+ MAPPER.readValue(input,
CustomTypesAndResources.ProductReview.class);
+
+ ctx.getShortTermMemory().set("id", inputObj.getId());
+
+ String content =
+ String.format(
+ "{\n" + "\"id\": %s,\n" + "\"review\": \"%s\"\n" + "}",
+ inputObj.getId(), inputObj.getReview());
+ ChatMessage msg = new ChatMessage(MessageRole.USER, "");
+
+ ctx.sendEvent(
+ new ChatRequestEvent(
+ "reviewAnalysisModel", List.of(msg), Map.of("input",
content), null));
+}
- @Action(EventType.ChatResponseEvent)
- public static void processChatResponse(Event event, RunnerContext ctx)
- throws Exception {
- ChatResponseEvent chatResponse = ChatResponseEvent.fromEvent(event);
- JsonNode jsonNode =
MAPPER.readTree(chatResponse.getResponse().getText());
- JsonNode scoreNode = jsonNode.findValue("score");
- JsonNode reasonsNode = jsonNode.findValue("reasons");
- if (scoreNode == null || reasonsNode == null) {
- throw new IllegalStateException(
- "Invalid response from LLM: missing 'score' or 'reasons'
field.");
- }
- List<String> result = new ArrayList<>();
- if (reasonsNode.isArray()) {
- for (JsonNode node : reasonsNode) {
- result.add(node.asText());
- }
+@Action(EventType.ChatResponseEvent)
+public static void processChatResponse(Event event, RunnerContext ctx) throws
Exception {
+ ChatResponseEvent chatResponse = (ChatResponseEvent) event;
+ // Fail fast on a malformed LLM response: a parse error here propagates
and fails the
+ // agent, so a dropped input is surfaced instead of silently lost. In
production, choose
+ // the handling that fits your pipeline: raise to fail the input (as
below), emit an
+ // OutputEvent carrying an error sentinel, or send a custom error event so
downstream
+ // operators can detect the failure.
+ JsonNode jsonNode = MAPPER.readTree(chatResponse.getResponse().getText());
+ JsonNode scoreNode = jsonNode.findValue("score");
+ JsonNode reasonsNode = jsonNode.findValue("reasons");
+ if (scoreNode == null || reasonsNode == null) {
+ throw new IllegalStateException(
+ "Invalid response from LLM: missing 'score' or 'reasons'
field.");
+ }
+ List<String> result = new ArrayList<>();
+ if (reasonsNode.isArray()) {
+ for (JsonNode node : reasonsNode) {
+ result.add(node.asText());
}
-
- ctx.sendEvent(
- new OutputEvent(
- new CustomTypesAndResources.ProductReviewAnalysisRes(
-
ctx.getShortTermMemory().get("id").getValue().toString(),
- scoreNode.asInt(),
- result)));
}
+
+ ctx.sendEvent(
+ new OutputEvent(
+ new CustomTypesAndResources.ProductReviewAnalysisRes(
+
ctx.getShortTermMemory().get("id").getValue().toString(),
+ scoreNode.asInt(),
+ result)));
}
-```
+}```
Review Comment:
Could we put the closing code fence on its own line after `}`? In the
rendered page, the three backticks remain inside the Java code block, so
copying the example produces invalid Java code.
##########
docs/content/docs/get-started/setup.md:
##########
@@ -257,28 +319,29 @@ cp
dist/flink-${FLINK_VERSION%.*}/target/flink-agents-dist-flink-${FLINK_VERSION
{{< /tabs >}}
-### Build Environment Variables
+{{< hint info >}}
+**Note:** The quick install performs this step for you, so you can skip it if
you ran
+`install.sh`.
+{{< /hint >}}
-The following environment variables can be used to control how JARs are
resolved during `pip install flink-agents` (from sdist) or `python -m build`:
+## Developing with the Java API
-| Variable | Description |
-|----------|-------------|
-| `FLINK_AGENTS_SKIP_JAR_DOWNLOAD` | Set to `1`, `true`, `yes`, or `on`
(case-insensitive) to skip downloading JARs from Maven Central. Useful when
building from source with `tools/build.sh`, which copies JARs from the local
Maven build. |
-| `FLINK_AGENTS_MAVEN_MIRROR` | Override the Maven repository base URL for JAR
downloads. Defaults to `https://repo1.maven.org/maven2`. Useful for
environments with restricted network access or internal mirrors. |
+A Java Flink job declares its Agent in Java, so it also needs the Flink Agents
artifacts
+on its compile classpath.
Review Comment:
A Java Flink job can also load a YAML-defined Agent, as
`YamlWorkflowAgentExample` demonstrates. Could we avoid tying the Agent
definition language to the hosting job here? For example:
> To develop a Java Flink job, add the Flink Agents artifacts to its compile
classpath.
--
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]
