wenjin272 commented on code in PR #830: URL: https://github.com/apache/flink-agents/pull/830#discussion_r3401363752
########## docs/content/docs/get-started/quickstart/parallel_llm.md: ########## @@ -0,0 +1,461 @@ +--- +title: 'Parallel LLM Calls' +weight: 4 +type: docs +--- +<!-- +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. +--> + +## Overview + +Flink Agents supports parallel LLM invocations via multi-action fan-out. By emitting multiple `ChatRequestEvent` events from a single action, the framework's built-in chat action executes the corresponding LLM calls concurrently — no external orchestration is required. + +This quickstart introduces an example that demonstrates how to build a parallel LLM workflow with Flink Agents: + +The **Parallel Sentiment Analysis** agent processes a restaurant review and judges sentiment along three dimensions (taste / service / price) in parallel, then aggregates the results into a one-line summary with a final LLM call. The end-to-end wall clock time is roughly "slowest single branch + aggregation call", rather than the sum of all four calls. + +{{< hint info >}} +**JDK version note (Java only):** On JDK 21+, the framework uses the Continuation API to execute concurrent chat actions in parallel. On JDK < 21, the framework silently falls back to sequential execution — the result is identical, but the LLM calls run one after another. Python uses native coroutines and always executes in parallel regardless of the JDK version. +{{< /hint >}} + +## Code Walkthrough + +### Prepare Agents Execution Environment + +Create the agents execution environment, and register the available chat model connection to the environment. + +{{< tabs "Prepare Agents Execution Environment" >}} + +{{< tab "Python" >}} +```python +# Set up the Flink streaming environment and the Agents execution environment. +env = StreamExecutionEnvironment.get_execution_environment() +env.set_parallelism(1) +agents_env = AgentsExecutionEnvironment.get_execution_environment(env) + +# Add Ollama chat model connection to be used by the ParallelChatAgent. +agents_env.add_resource( + "ollama_server", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor( + clazz=ResourceName.ChatModel.OLLAMA_CONNECTION, + request_timeout=240.0, + ), +) +``` +{{< /tab >}} + +{{< tab "Java" >}} +```Java +// Set up the Flink streaming environment and the Agents execution environment. +StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); +env.setParallelism(1); +AgentsExecutionEnvironment agentsEnv = + AgentsExecutionEnvironment.getExecutionEnvironment(env); + +// Add Ollama chat model connection to be used by the ParallelChatAgent. +agentsEnv.addResource( + "ollamaChatModelConnection", + ResourceType.CHAT_MODEL_CONNECTION, + ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_CONNECTION) + .addInitialArgument("endpoint", "http://localhost:11434";) + .addInitialArgument("requestTimeout", 240) + .build()); +``` +{{< /tab >}} + +{{< /tabs >}} + +### Create the Agent + +Below is the example code for the `ParallelChatAgent`. Two system prompts — `PARALLEL_SYSTEM_PROMPT` for the parallel aspect judgments and `AGGREGATE_SYSTEM_PROMPT` for the aggregation call — are packaged into `ChatRequestEvent`s by `_build_aspect_request` / `buildAspectRequest` and `_build_summarize_request` / `buildSummarizeRequest`. The agent defines a chat model and two actions: `request_aspect_judgments` pre-builds all requests and records a `{request_id → aspect}` map for reliable correlation, and `handle_response` looks up the dispatched aspect by `request_id`, accumulates the results, and emits the final output. For more details, please refer to the [Workflow Agent]({{< ref "docs/development/workflow_agent" >}}) documentation. + +{{< tabs "Create the Agent" >}} + +{{< tab "Python" >}} +```python +PARALLEL_SYSTEM_PROMPT = ( + "You are a sentiment analysis assistant. Return JSON: " + '{"aspect":"<dimension>", "result":"<positive|negative|not_mentioned>"}' + " — no explanation, no extra fields." +) +AGGREGATE_SYSTEM_PROMPT = ( + "You are a summary assistant. Based on the sentiment judgments for three " + "dimensions, compose a brief one-line evaluation. Return JSON: " + '{"summary":"taste:<positive/negative/not_mentioned>, ' + "service:<positive/negative/not_mentioned>, " + 'price:<positive/negative/not_mentioned>"} — return only this JSON.' +) + + +def _build_aspect_request(text: str, aspect: str) -> ChatRequestEvent: + """Build a ChatRequestEvent for a single aspect dimension.""" + return ChatRequestEvent( + model="sentiment_model", + messages=[ + ChatMessage(role=MessageRole.SYSTEM, content=PARALLEL_SYSTEM_PROMPT), + ChatMessage( + role=MessageRole.USER, + content=f'Judge the "{aspect}" dimension: {text}', + ), + ], + output_schema=OutputSchema(output_schema=AspectResponse), + ) + + +def _build_summarize_request(row: Dict[str, Any]) -> ChatRequestEvent: + """Build a ChatRequestEvent for the aggregation phase.""" + sentiments = row["sentiments"] + body = ( + f"Original: {row['text']}\n" + + "Judgments: " + + " ".join(f"{a}:{sentiments[a]}" for a in ASPECTS) Review Comment: The definition of `ASPECTS` is provided on the Java example but not on the Python side. I believe adding it to the code would improve readability. -- 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]
