Apache Flink Agents 0.3 学习教程

┌─────────────────────────────────────────────────────────┐
│                    Agent 层                              │
│  Workflow Agent  |  ReAct Agent  |  YAML Agent          │
├─────────────────────────────────────────────────────────┤
│                  编排与资源层                             │
│  Action  |  Event  |  Prompt  |  Memory  |  Skills     │
├─────────────────────────────────────────────────────────┤
│                  模型与工具层                             │
│  Chat Model  |  Tool  |  MCP  |  Vector Store           │
├─────────────────────────────────────────────────────────┤
│                  Flink 运行时层                           │
│  DataStream API  |  Table API  |  State  |  Checkpoint │
└─────────────────────────────────────────────────────────┘

from pyflink.datastream import StreamExecutionEnvironment
from flink_agents.api.execution_environment import AgentsExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)

curl -fsSL https://raw.githubusercontent.com/apache/flink-agents/main/tools/install.sh | bash

# 下载并解压 Flink
export FLINK_VERSION=<version>
tar -xzf flink-${FLINK_VERSION}-bin-scala_2.12.tgz
export FLINK_HOME=$(pwd)/flink-${FLINK_VERSION}

# 将 flink-python JAR 从 opt 复制到 lib（PyFlink 需要）
cp $FLINK_HOME/opt/flink-python-${FLINK_VERSION}.jar $FLINK_HOME/lib/

python3 -m venv venv
source venv/bin/activate

pip install flink-agents apache-flink==${FLINK_VERSION}

export PYTHONPATH=$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')

<properties>
    <flink.version>2.2.1</flink.version>
    <flink-agents.version>0.3.0</flink-agents.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-agents-api</artifactId>
        <version>${flink-agents.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <!-- IDE 本地运行时需要 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-agents-ide-support</artifactId>
        <version>${flink-agents.version}</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

from flink_agents.api.agent import Agent
from flink_agents.api.decorators import action
from flink_agents.api.event import Event, InputEvent, OutputEvent
from flink_agents.api.context import RunnerContext

class MyAgent(Agent):
    @action(InputEvent.EVENT_TYPE)
    @staticmethod
    def process_input(event: Event, ctx: RunnerContext) -> None:
        # 从事件中提取输入数据
        input_event = InputEvent.from_event(event)
        # 通过 RunnerContext 发送新事件，触发下游 Action
        ctx.send_event(SomeEvent(...))

# 触发内置的聊天模型 Action
ctx.send_event(ChatRequestEvent(model="my_model", messages=messages))

# 输出结果到下游（OutputEvent 会被框架直接收集并输出）
ctx.send_event(OutputEvent(output=result))

# 发送自定义事件
@action(InputEvent.EVENT_TYPE)
@staticmethod
def create_my_event(event: Event, ctx: RunnerContext) -> None:
    ctx.send_event(
        Event(type="my_event", attributes={"field1": "test", "field2": 42})
    )

# 消费自定义事件
@action("my_event")
@staticmethod
def handle_my_event(event: Event, ctx: RunnerContext) -> None:
    field1: str = event.get_attr("field1")
    field2: int = event.get_attr("field2")

from typing import ClassVar
from typing_extensions import override

class MyEvent(Event):
    EVENT_TYPE: ClassVar[str] = "my_event"

    def __init__(self, value: str) -> None:
        super().__init__(type=MyEvent.EVENT_TYPE, attributes={"value": value})

    @classmethod
    @override
    def from_event(cls, event: Event) -> "MyEvent":
        assert "value" in event.attributes
        result = MyEvent(value=event.attributes["value"])
        result.id = event.id  # 保留基础事件 ID
        return result

    @property
    def value(self) -> str:
        return self.get_attr("value")

input_event = InputEvent.from_event(event)
input_data = MyModel.model_validate(input_event.input)

from flink_agents.api.decorators import chat_model_connection
from flink_agents.api.resource import ResourceDescriptor, ResourceName

class MyAgent(Agent):
    @chat_model_connection
    @staticmethod
    def ollama_connection() -> ResourceDescriptor:
        return ResourceDescriptor(
            clazz=ResourceName.ChatModel.OLLAMA_CONNECTION,
            base_url="http://localhost:11434",
            request_timeout=120.0
        )

@chat_model_setup
@staticmethod
def my_chat_model() -> ResourceDescriptor:
    return ResourceDescriptor(
        clazz=ResourceName.ChatModel.OLLAMA_SETUP,
        connection="ollama_connection",
        model="qwen3:8b",
        temperature=0.7,
        num_ctx=4096,
    )

Action 发送 ChatRequestEvent
        ↓
内置 chat_model_action 处理请求
        ↓
   ┌─────────────────────┐
   │ LLM 返回最终回答？    │
   └─────────────────────┘
     ↓ 是              ↓ 否（请求调用工具）
ChatResponseEvent    ToolRequestEvent
     ↓                    ↓
 Action 处理响应      tool_call_action 执行工具
                         ↓
                    ToolResponseEvent
                         ↓
                    chat_model_action 继续聊天

@chat_model_connection
@staticmethod
def java_chat_model_connection() -> ResourceDescriptor:
    return ResourceDescriptor(
        clazz=ResourceName.ChatModel.JAVA_WRAPPER_CONNECTION,
        java_clazz=ResourceName.ChatModel.Java.OLLAMA_CONNECTION,
        endpoint="http://localhost:11434",
        requestTimeout=120,
    )

from flink_agents.api.chat_model import BaseChatModelConnection, BaseChatModelSetup

class MyChatModelConnection(BaseChatModelConnection):
    def chat(self, messages, tools=None, **kwargs) -> ChatMessage:
        # 核心方法：发送消息到 LLM 并返回响应
        pass

class MyChatModelSetup(BaseChatModelSetup):
    @property
    def model_kwargs(self) -> dict:
        return {"model": self.model, "temperature": 0.7}

from flink_agents.api.decorators import tool

class ReviewAnalysisAgent(Agent):
    @tool
    @staticmethod
    def notify_shipping_manager(id: str, review: str) -> None:
        """Notify the shipping manager when product received a negative review due to
        shipping damage.

        Parameters
        ----------
        id : str
            The id of the product that received a negative review
        review: str
            The negative review content
        """
        notify_shipping_manager(id=id, review=review)

    @chat_model_setup
    @staticmethod
    def review_analysis_model() -> ResourceDescriptor:
        return ResourceDescriptor(
            clazz=ResourceName.ChatModel.OLLAMA_SETUP,
            connection="ollama_server",
            model="qwen3:8b",
            tools=["notify_shipping_manager"],  # 通过名称引用工具
        )

agents_env.add_resource(
    "notify_shipping_manager",
    ResourceType.TOOL,
    Tool.from_callable(notify_shipping_manager)
)

@action(InputEvent.EVENT_TYPE)
@staticmethod
def process_input(event: Event, ctx: RunnerContext) -> None:
    input = ProductReview.model_validate(InputEvent.from_event(event).input)
    # 存储到短期记忆（包含感知记忆功能）
    ctx.short_term_memory.set("id", input.id)

@action(ChatResponseEvent.EVENT_TYPE)
@staticmethod
def process_response(event: Event, ctx: RunnerContext) -> None:
    # 从短期记忆中读取
    product_id = ctx.short_term_memory.get("id")

---
name: math-calculator
description: Calculate mathematical expressions using shell commands. Use when the user asks to perform arithmetic.
license: Apache-2.0
compatibility: Requires bash with bc
---

# Math Calculator Skill

## When to Use
Use this skill whenever the user asks to evaluate a numeric expression.

## Method
Evaluate expressions with the `bc` calculator:

\`\`\`bash
echo "(2 + 3) * 4" | bc
# Output: 20
\`\`\`

from flink_agents.api.decorators import skills
from flink_agents.api.skills import Skills

class MathAgent(Agent):
    @skills
    @staticmethod
    def my_skills() -> Skills:
        return Skills.from_local_dir("/path/to/skills")

@chat_model_setup
@staticmethod
def math_model() -> ResourceDescriptor:
    return ResourceDescriptor(
        clazz=ResourceName.ChatModel.OLLAMA_SETUP,
        connection="ollama_server",
        model="qwen3.5:9b",
        prompt="system_prompt",
        skills=["math-calculator"],          # 按名称启用技能
        allowed_commands=["echo", "bc"],      # 白名单允许的 shell 命令
    )

from flink_agents.api.agents.react_agent import ReActAgent

my_react_agent = ReActAgent(
    chat_model=chat_model_descriptor,
    prompt=my_prompt,
    output_schema=MyBaseModelDataType,
)

from flink_agents.api.prompts.prompt import Prompt
from flink_agents.api.chat_message import ChatMessage, MessageRole

my_prompt = Prompt.from_messages(
    messages=[
        ChatMessage(
            role=MessageRole.SYSTEM,
            content="分析产品评论并生成满意度评分..."
        ),
        ChatMessage(
            role=MessageRole.USER,
            content='{"id": {id}, "review": {review}}'
        ),
    ],
)

from pydantic import BaseModel

class ReviewAnalysisResult(BaseModel):
    id: str
    score: int
    reasons: list[str]

from pyflink.common import WatermarkStrategy

# 1. 创建输入 DataStream
input_stream = env.from_source(
    source=your_source,
    watermark_strategy=WatermarkStrategy.no_watermarks(),
    source_name="your_source_name",
)

# 2. 将 Agent 应用于 DataStream
output_stream = (
    agents_env.from_datastream(
        input=input_stream,
        key_selector=lambda x: x.id  # 必须提供 KeySelector
    )
    .apply(your_agent)
    .to_datastream()
)

# 3. 消费输出
output_stream.print()

from pyflink.common.typeinfo import BasicTypeInfo, ExternalTypeInfo, RowTypeInfo
from pyflink.datastream import KeySelector
from pyflink.table import DataTypes, Schema

# 1. 创建执行环境（包含 Table 环境）
env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(stream_execution_environment=env)
agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env, t_env=t_env)

# 2. 定义 KeySelector
class MyKeySelector(KeySelector):
    def get_key(self, value):
        return value.id

# 3. 定义输出 Schema
output_type = ExternalTypeInfo(RowTypeInfo(
    [BasicTypeInfo.INT_TYPE_INFO()],
    ["result"],
))
schema = Schema.new_builder().column("result", DataTypes.INT()).build()

# 4. 将 Agent 应用于 Table
output_table = (
    agents_env.from_table(input=input_table, key_selector=MyKeySelector())
    .apply(your_agent)
    .to_table(schema=schema, output_type=output_type)
)

from pyflink.datastream import StreamExecutionEnvironment
from flink_agents.api.execution_environment import AgentsExecutionEnvironment
from flink_agents.api.resource import ResourceType

env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)

# 注册 Ollama 聊天模型连接
agents_env.add_resource(
    "ollama_server",
    ResourceType.CHAT_MODEL_CONNECTION,
    ollama_server_descriptor,
)

import json
import logging
from pydantic import BaseModel
from typing import ClassVar

from flink_agents.api.agent import Agent
from flink_agents.api.decorators import action, chat_model_setup, prompt, tool
from flink_agents.api.event import Event, InputEvent, OutputEvent, ChatResponseEvent
from flink_agents.api.context import RunnerContext
from flink_agents.api.chat_message import ChatMessage, MessageRole
from flink_agents.api.chat_event import ChatRequestEvent
from flink_agents.api.resource import ResourceDescriptor, ResourceName
from flink_agents.api.prompts.prompt import Prompt

# 数据模型
class ProductReview(BaseModel):
    id: str
    review: str

class ProductReviewAnalysisRes(BaseModel):
    id: str
    score: int
    reasons: list[str]

# 提示词
review_analysis_prompt = Prompt.from_messages(
    messages=[
        ChatMessage(
            role=MessageRole.SYSTEM,
            content="""你是一个产品评论分析专家。分析给定的产品评论，
            返回JSON格式的结果，包含：
            - score: 满意度评分(1-5)
            - reasons: 不满意原因列表

            示例输出：
            {"score": 2, "reasons": ["物流损坏", "质量不佳"]}
            """
        ),
        ChatMessage(
            role=MessageRole.USER,
            content="{input}"
        ),
    ]
)

class ReviewAnalysisAgent(Agent):
    """使用 LLM 分析产品评论，生成满意度评分和不满意原因。"""

    @prompt
    @staticmethod
    def review_analysis_prompt() -> Prompt:
        return review_analysis_prompt

    @tool
    @staticmethod
    def notify_shipping_manager(id: str, review: str) -> None:
        """当产品因运输损坏收到差评时通知运输经理。

        Parameters
        ----------
        id : str
            收到差评的产品ID
        review : str
            差评内容
        """
        notify_shipping_manager(id=id, review=review)

    @chat_model_setup
    @staticmethod
    def review_analysis_model() -> ResourceDescriptor:
        return ResourceDescriptor(
            clazz=ResourceName.ChatModel.OLLAMA_SETUP,
            connection="ollama_server",
            model="qwen3:8b",
            prompt="review_analysis_prompt",
            tools=["notify_shipping_manager"],
            extract_reasoning=True,
        )

    @action(InputEvent.EVENT_TYPE)
    @staticmethod
    def process_input(event: Event, ctx: RunnerContext) -> None:
        """处理输入事件，发送聊天请求。"""
        input = ProductReview.model_validate(InputEvent.from_event(event).input)
        # 将 id 存入短期记忆，供后续 Action 使用
        ctx.short_term_memory.set("id", input.id)

        content = f"""
            "id": {input.id},
            "review": {input.review}
        """
        msg = ChatMessage(role=MessageRole.USER)
        ctx.send_event(
            ChatRequestEvent(
                model="review_analysis_model",
                messages=[msg],
                prompt_args={"input": content},
            )
        )

    @action(ChatResponseEvent.EVENT_TYPE)
    @staticmethod
    def process_chat_response(event: Event, ctx: RunnerContext) -> None:
        """处理聊天响应，发送输出事件。"""
        chat_response = ChatResponseEvent.from_event(event)
        try:
            json_content = json.loads(chat_response.response.content)
            ctx.send_event(
                OutputEvent(
                    output=ProductReviewAnalysisRes(
                        id=ctx.short_term_memory.get("id"),
                        score=json_content["score"],
                        reasons=json_content["reasons"],
                    )
                )
            )
        except Exception:
            logging.exception(
                f"Error processing chat response {chat_response.response.content}"
            )

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FileSource
from pyflink.common import WatermarkStrategy, Duration
from pyflink.datastream.formats import StreamFormat

# 读取产品评论文件作为流式数据源
product_review_stream = env.from_source(
    source=FileSource.for_record_stream_format(
        StreamFormat.text_line_format(),
        f"file:///{current_dir}/resources/product_review.txt",
    )
    .monitor_continuously(Duration.of_minutes(1))
    .build(),
    watermark_strategy=WatermarkStrategy.no_watermarks(),
    source_name="streaming_agent_example",
).map(
    lambda x: ProductReview.model_validate_json(x)  # 反序列化 JSON
)

# 使用 ReviewAnalysisAgent 分析每条评论
review_analysis_res_stream = (
    agents_env.from_datastream(
        input=product_review_stream,
        key_selector=lambda x: x.id
    )
    .apply(ReviewAnalysisAgent())
    .to_datastream()
)

# 打印分析结果
review_analysis_res_stream.print()

# 执行 Flink 管道
agents_env.execute("Workflow Agent Example Job")

from pyflink.datastream import KeySelector

class TableKeySelector(KeySelector):
    """从 Table 行（dict）中提取分区键。"""
    def get_key(self, value):
        return str(value["id"])

class TableReviewAnalysisAgent(Agent):
    # prompt / tool / chat_model_setup 与 ReviewAnalysisAgent 相同

    @action(InputEvent.EVENT_TYPE)
    @staticmethod
    def process_input(event: Event, ctx: RunnerContext) -> None:
        # Table 输入以 dict 形式到达，而非 POJO
        input_dict = InputEvent.from_event(event).input
        product_id = str(input_dict["id"])
        review_text = str(input_dict["review"])
        # ... 后续逻辑与 ReviewAnalysisAgent 相同

# 通过 Table API 集成
input_table = t_env.from_path("product_reviews")

review_analysis_res_stream = (
    agents_env.from_table(input=input_table, key_selector=TableKeySelector())
    .apply(TableReviewAnalysisAgent())
    .to_datastream()
)

@action(InputEvent.EVENT_TYPE)
@staticmethod
def process_input(event: Event, ctx: RunnerContext) -> None:
    input_event = InputEvent.from_event(event)

    def slow_external_call(data: str) -> str:
        time.sleep(2)
        return f"Processed: {data}"

    # 持久化执行：结果被保存，恢复时重放
    result = ctx.durable_execute(slow_external_call, input_event.input)
    ctx.send_event(OutputEvent(output=result))

def submit_payment(order_id: str) -> str:
    return payment_client.submit(order_id)

def payment_reconciler() -> str:
    """恢复时调用，查询支付状态而非重新提交。"""
    status = payment_client.get_status(order_id)
    if status == "SUCCEEDED":
        return payment_client.lookup_completed_payment(order_id)
    raise payment_client.get_failure(order_id)

result = ctx.durable_execute(
    submit_payment,
    order_id,
    reconciler=payment_reconciler,
)

@action(InputEvent.EVENT_TYPE)
@staticmethod
async def process_with_async(event: Event, ctx: RunnerContext) -> None:
    input_event = InputEvent.from_event(event)

    def slow_external_call(data: str) -> str:
        time.sleep(2)
        return f"Processed: {data}"

    result = await ctx.durable_execute_async(slow_external_call, input_event.input)
    ctx.send_event(OutputEvent(output=result))

from flink_agents.api.function import JavaFunction

class MyAgent(Agent):
    @action(
        InputEvent.EVENT_TYPE,
        target=JavaFunction.for_action("com.example.MyHandlers", "handleInput"),
    )
    @staticmethod
    def handle_input(event: Event, ctx: RunnerContext) -> None:
        raise NotImplementedError("cross-language stub")

# 下载并安装 Ollama（需要 0.9.0 或更高版本）
# 从 https://ollama.com 下载

# 拉取 qwen3:8b 模型
ollama pull qwen3:8b

# 配置 PYTHONPATH
export PYTHONPATH=$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')

# 启动本地独立 Flink 集群
$FLINK_HOME/bin/start-cluster.sh

export PYTHONPATH=$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')

# 运行评论分析示例
$FLINK_HOME/bin/flink run -py ./flink-agents/python/flink_agents/examples/quickstart/workflow_single_agent_example.py

# 运行产品建议示例（多 Agent）
$FLINK_HOME/bin/flink run -py ./flink-agents/python/flink_agents/examples/quickstart/workflow_multiple_agent_example.py