[python][examples] Add quickstart workflow examples

python/flink_agents/api/execution_environment.py

@@ -162,7 +162,7 @@ def from_list(self, input: List[Dict[str, Any]]) -> AgentBuilder:
 
     @abstractmethod
     def from_datastream(
-        self, input: DataStream, key_selector: KeySelector | None = None
+        self, input: DataStream, key_selector: KeySelector | Callable | None = None
     ) -> AgentBuilder:
         """Set input for agents. Used for remote execution.
 
@@ -184,7 +184,7 @@ def from_table(
         self,
         input: Table,
         t_env: StreamTableEnvironment,
-        key_selector: KeySelector | None = None,
+        key_selector: KeySelector | Callable | None = None,
     ) -> AgentBuilder:
         """Set input for agents. Used for remote execution.
 
@@ -325,7 +325,10 @@ def add_embedding_model_connection(
             msg = f"Embedding model connection {name} already defined"
             raise ValueError(msg)
         kwargs["name"] = name
-        self._resources[ResourceType.EMBEDDING_MODEL_CONNECTION][name] = (connection, kwargs)
+        self._resources[ResourceType.EMBEDDING_MODEL_CONNECTION][name] = (
+            connection,
+            kwargs,
+        )
         return self
 
     def add_embedding_model_setup(
@@ -354,7 +357,9 @@ def add_embedding_model_setup(
         self._resources[ResourceType.EMBEDDING_MODEL][name] = (embedding_model, kwargs)
         return self
 
-    def add_mcp_server(self, name: str, mcp_server: MCPServer) -> "AgentsExecutionEnvironment":
+    def add_mcp_server(
+        self, name: str, mcp_server: MCPServer
+    ) -> "AgentsExecutionEnvironment":
         """Add an MCP server to the agent execution environment.
 
         Parameters

python/flink_agents/examples/quickstart/NOTICE

@@ -0,0 +1,8 @@
+This project uses data generated from the Amazon Review Data (2018) dataset.
+
+Citation:
+Justifying recommendations using distantly-labeled reviews and fined-grained aspects
+Jianmo Ni, Jiacheng Li, Julian McAuley
+Empirical Methods in Natural Language Processing (EMNLP), 2019
+
+Dataset URL: https://nijianmo.github.io/amazon/index.html#complete-data

python/flink_agents/examples/quickstart/__init__.py

@@ -0,0 +1,17 @@
+################################################################################
+#  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.
+#################################################################################

python/flink_agents/examples/quickstart/agents/__init__.py

@@ -0,0 +1,17 @@
+################################################################################
+#  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.
+#################################################################################

python/flink_agents/examples/quickstart/agents/product_suggestion_agent.py

@@ -0,0 +1,166 @@
+################################################################################
+#  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.
+#################################################################################
+import json
+import logging
+from typing import Any, Dict, List, Tuple, Type
+
+from pydantic import BaseModel
+
+from flink_agents.api.agent import Agent
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import (
+    BaseChatModelSetup,
+)
+from flink_agents.api.decorators import (
+    action,
+    chat_model_setup,
+    prompt,
+)
+from flink_agents.api.events.chat_event import ChatRequestEvent, ChatResponseEvent
+from flink_agents.api.events.event import InputEvent, OutputEvent
+from flink_agents.api.prompts.prompt import Prompt
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.integrations.chat_models.ollama_chat_model import (
+    OllamaChatModelSetup,
+)
+
+
+class ProductReviewSummary(BaseModel):
+    """Aggregates multiple reviews and insights using LLM for a product.
+
+    Attributes:
+        id (str): The unique identifier of the product.
+        score_hist (List[str]): A collection of rating scores from various reviews.
+        unsatisfied_reasons (List[str]): A list of reasons or insights generated by LLM
+            to explain the rating.
+    """
+
+    id: str
+    score_hist: List[str]
+    unsatisfied_reasons: List[str]
+
+
+class ProductSuggestion(BaseModel):
+    """Provides a summary of review data including suggestions for improvement.
+
+    Attributes:
+        id (str): The unique identifier of the product.
+        score_histogram (List[int]): A collection of rating scores from various reviews.
+        suggestions (List[str]): Suggestions or recommendations generated as a result of
+            review analysis.
+    """
+
+    id: str
+    score_hist: List[str]
+    suggestions: List[str]
+
+
+class ProductSuggestionAgent(Agent):
+    """An agent that uses a large language model (LLM) to generate actionable product
+    improvement suggestions from aggregated product review data.
+
+    This agent receives a summary of product reviews, including a rating distribution
+    and a list of user dissatisfaction reasons, and produces concrete suggestions for
+    product enhancement. It handles prompt construction, LLM interaction, and output
+    parsing.
+    """
+
+    @prompt
+    @staticmethod
+    def generate_suggestion_prompt() -> Prompt:
+        """Generate product suggestions based on the rating distribution and user
+        dissatisfaction reasons.
+        """
+        prompt_str = """
+        Based on the rating distribution and user dissatisfaction reasons, generate three actionable suggestions for product improvement.
+
+        Input format:
+        {{
+            "id": "1",
+            "score_histogram": ["10%", "20%", "10%", "15%", "45%"],
+            "unsatisfied_reasons": ["reason1", "reason2", "reason3"]
+        }}
+
+        Ensure that your response can be parsed by Python json,use the following format as an example:
+        {{
+            "suggestion_list": [
+                "suggestion1",
+                "suggestion2",
+                "suggestion3"
+            ]
+        }}
+
+        input:
+        {input}
+        """
+        return Prompt.from_text("generate_suggestion_prompt", prompt_str)
+
+    @chat_model_setup
+    @staticmethod
+    def generate_suggestion_model() -> Tuple[Type[BaseChatModelSetup], Dict[str, Any]]:
+        """ChatModel which focus on generating product suggestions."""
+        return OllamaChatModelSetup, {
+            "name": "generate_suggestion_model",
+            "connection": "ollama_server",
+            "prompt": "generate_suggestion_prompt",
+            "extract_reasoning": True,
+        }
+
+    @action(InputEvent)
+    @staticmethod
+    def process_input(event: InputEvent, ctx: RunnerContext) -> None:
+        """Process input event."""
+        input: ProductReviewSummary = event.input
+        ctx.get_short_term_memory().set("id", input.id)
+        ctx.get_short_term_memory().set("score_hist", input.score_hist)
+
+        content = f"""
+            "id": {input.id},
+            "score_histogram": {input.score_hist},
+            "unsatisfied_reasons": {input.unsatisfied_reasons}
+        """
+        ctx.send_event(
+            ChatRequestEvent(
+                model="generate_suggestion_model",
+                messages=[
+                    ChatMessage(role=MessageRole.USER, extra_args={"input": content})
+                ],
+            )
+        )
+
+    @action(ChatResponseEvent)
+    @staticmethod
+    def process_chat_response(event: ChatResponseEvent, ctx: RunnerContext) -> None:
+        """Process chat response event."""
+        try:
+            json_content = json.loads(event.response.content)
+            ctx.send_event(
+                OutputEvent(
+                    output=ProductSuggestion(
+                        id=ctx.get_short_term_memory().get("id"),
+                        score_hist=ctx.get_short_term_memory().get("score_hist"),
+                        suggestions=json_content["suggestions"],
+                    )
+                )
+            )
+        except Exception:
+            logging.exception(
+                f"Error processing chat response {event.response.content}"
+            )
+
+            # To fail the agent, you can raise an exception here.