已经在虚拟机部署好Apache DolphinScheduler了，想尝试下在Flink新建一个Flink节点，然后用Flink消费Kafka数据。

Apache DolphinScheduler用的是单机部署，具体操作可以参考官方文档：DolphinScheduler | 文档中心(https://dolphinscheduler.apache.org/zh-cn/docs/3.3.2/guide/in...).

• 前置条件：已经安装Java 11、DolphinScheduler 3.3.2、Flink 1.18.1、Kafka 3.6.0，Zookeeper用Kafka内置的。建议这些安装都下载二进制的安装包到虚拟机安装，用命令安装的不可控，我下载的二进制包如下：

配置好Flink的环境变量

1、编辑环境变量：

sudo vim ~/.bashrc

增加Flink的路径

2、使环境变量生效：

#使环境变量生效
source ~/.bashrc
#查看环境变量
echo $Flink_HOME

修改Kafka、Flink以及DolphinScheduler的配置文件

因为用的是虚拟机，为了让外面的主机能够访问到虚拟机的网络，需要修改下配置文件

1. 修改Kafka配置：找到Kafka安装包下的config文件夹，修改config下的server.properties文件，修改listeners是为了外面的主机能够访问到虚拟机的Kafka，还有把advertised.listeners改成虚拟机地址，写样例的时候能连上虚拟机的Kafka地址，不然默认连localhost

broker.id=0
listeners=PLAINTEXT://0.0.0.0:9092
#192.168.146.132修改成虚拟机ip
advertised.listeners=PLAINTEXT://192.168.146.132:9092

2. 修改Flink配置：找到Flink安装包下的conf文件夹，修改conf下的Flink-conf.yaml文件，把里面所有的localhost地址全部改成0.0.0.0，以便主机能访问到虚拟机的Flink。还有增加jobmanager和taskmanager的内存

jobmanager.rpc.address: 0.0.0.0
jobmanager.bind-host: 0.0.0.0
jobmanager.cpu.cores: 1
jobmanager.memory.process.size: 1600m
taskmanager.bind-host: 0.0.0.0
taskmanager.host: 0.0.0.0
taskmanager.memory.process.size: 2048m
taskmanager.cpu.cores: 1

3. 修改Apache DolphinScheduler的配置文件，从Apache DolphinScheduler的启动脚本文件dolphinscheduler-daemon.sh可以看出，配置环境变量用的是bin/env文件夹下的dolphinscheduler_env.sh。

查看dolphinscheduler-daemon.sh文件：

修改dolphinscheduler_env.sh文件，新增JAVA、Flink路径：

#修改成自己的JAVA、Flink路径
export JAVA_HOME=/data/jdk-11.0.29
export Flink_HOME=/data/Flink-1.18.1

关闭防火墙，启动应用

启动应用，包括Zookeeper、Kafka、Flink以及Apache DolphinScheduler。

#关闭防火墙
sudo systemctl stop firewalld
 
# 在 Flink 根目录下，执行以下命令启动 Flink 集群
bin/start-cluster.sh
 
# 启动 ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties &
 
# 启动 Kafka 服务器
bin/Kafka-server-start.sh config/server.properties &
 
#创建 Kafka 主题
bin/Kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
 
#使用命令行生产者发送消息
bin/Kafka-console-producer.sh --topic test --bootstrap-server localhost:9092
 
#消费
bin/Kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092

# 启动 Standalone Server 服务
bash ./bin/dolphinscheduler-daemon.sh start standalone-server

测试

测试Flink、Apache DolphinScheduler是否能访问成功。

1. Flink访问地址：http://localhost:8081/，localhost改成自己虚拟机地址

2. Apache DolphinScheduler访问地址：http://localhost:12345/dolphinscheduler/ui ，localhost改成自己虚拟机地址即可登录系统 UI。默认的用户名和密码是 admin/dolphinscheduler123

编写样例

用Flink消费Kafka数据，然后打包上传到Apache DolphinScheduler，启动Flink任务：

1. 编写样例：

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
    <groupId>com.example</groupId>
    <artifactId>Flink-Kafka-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
 
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <Flink.version>1.18.1</Flink.version>
        <scala.binary.version>2.12</scala.binary.version>
        <Kafka.version>3.6.0</Kafka.version>
    </properties>
 
    <dependencies>
        <!-- Flink核心依赖 -->
        <dependency>
            <groupId>org.apache.Flink</groupId>
            <artifactId>Flink-java</artifactId>
            <version>${Flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.Flink</groupId>
            <artifactId>Flink-streaming-java</artifactId>
            <version>${Flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.Flink</groupId>
            <artifactId>Flink-clients</artifactId>
            <version>${Flink.version}</version>
        </dependency>
 
        <!-- 连接器基础依赖 -->
        <dependency>
            <groupId>org.apache.Flink</groupId>
            <artifactId>Flink-connector-base</artifactId>
            <version>${Flink.version}</version>
        </dependency>
 
        <!-- Kafka连接器（关键修改点） -->
        <dependency>
            <groupId>org.apache.Flink</groupId>
            <artifactId>Flink-connector-Kafka</artifactId>
            <version>3.1.0-1.18</version>
        </dependency>
        <dependency>
            <groupId>org.apache.Kafka</groupId>
            <artifactId>Kafka-clients</artifactId>
            <version>${Kafka.version}</version>
        </dependency>
 
        <!-- 日志依赖 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.36</version>
            <scope>runtime</scope>
        </dependency>
    </dependencies>
 
    <repositories>
        <repository>
            <id>aliyun</id>
            <url>https://maven.aliyun.com/repository/public</url>
            <releases>
                <enabled>true</enabled>
            </releases>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
        <repository>
            <id>apache-releases</id>
            <url>https://repository.apache.org/content/repositories/releases/</url>
        </repository>
    </repositories>
 
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>${maven.compiler.source}</source>
                    <target>${maven.compiler.target}</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.4</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>org.apache.Flink:force-shading</exclude>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

FlinkKafkaConsumerExample.java

import org.apache.Flink.api.common.functions.FlatMapFunction;
import org.apache.Flink.api.java.tuple.Tuple2;
import org.apache.Flink.api.java.utils.ParameterTool;
import org.apache.Flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.Flink.streaming.api.datastream.DataStream;
import org.apache.Flink.streaming.api.functions.ProcessFunction;
import org.apache.Flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.Flink.util.Collector;
import org.apache.Flink.streaming.connectors.Kafka.FlinkKafkaConsumer;
import org.apache.Flink.api.common.serialization.SimpleStringSchema;
import org.apache.Kafka.clients.consumer.ConsumerConfig;
import org.apache.Kafka.common.serialization.StringDeserializer;
 
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
 
 
public class FlinkKafkaConsumerExample {
    private static volatile int messageCount = 0;
    private static volatile boolean shouldStop = false;
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        // Kafka 配置
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.146.132:9092"); // Kafka broker 地址
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); // 消费者组
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
 
        // 创建 Kafka 消费者
        FlinkKafkaConsumer<String> KafkaConsumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
        KafkaConsumer.setStartFromEarliest(); // 从最早的消息开始消费
        DataStream<String> stream = env.addSource(KafkaConsumer);
 
        // 处理数据：分词和计数
        DataStream<Tuple2<String, Integer>> counts = stream
                .flatMap(new Tokenizer())
                .keyBy(value -> value.f0)
                .sum(1);
 
 
        counts.addSink(new RichSinkFunction<Tuple2<String, Integer>>() {
            @Override
            public void invoke(Tuple2<String, Integer> value, Context context) {
                System.out.println(value);
                messageCount++;
 
                // 检查是否达到停止条件
                if (messageCount >= 2 && !shouldStop) {
                    System.out.println("Processed 2 messages, stopping job.");
                    shouldStop = true; // 设置标志位，表示应该停止
                }
            }
        });
 
        // 执行作业并获取 JobClient
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            try {
                // 启动作业并获取 JobClient
                org.apache.Flink.core.execution.JobClient jobClient = env.executeAsync("Flink Kafka WordCount");
                System.out.println("Job ID: " + jobClient.getJobID());
 
                // 监测条件并取消作业
                while (!shouldStop) {
                    Thread.sleep(100); // 每100毫秒检查一次
                }
 
                // 达到停止条件时取消作业
                if (shouldStop) {
                    System.out.println("Cancelling the job...");
                    jobClient.cancel().get(); // 取消作业
                }
 
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
 
        // 在主线程中等待作业结束
        future.join(); // 等待作业完成
    }
 
    // Tokenizer 类用于将输入字符串转化为单词
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] tokens = value.toLowerCase().split("\\W+");
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
 
}

2. 打包上传到Apache DolphinScheduler

3. 新建Flink节点，并启动

在Apache DolphinScheduler的任务实例看启动日志：

在虚拟机启动生产者，输出字符串，然后可以在Flink查看输出Kafka生产的消息：

原文链接：https://blog.csdn.net/Analyze_ing/article/details/156940553

核心原则

OpenManus的工具系统基于「插件化设计」，所有自定义工具需继承框架的BaseTool基类，实现标准化接口，再通过配置文件注册，即可被AI智能体识别和调用。

一、前置准备

1. 确认目录结构：在OpenManus项目根目录下，建议创建custom_tools目录存放自定义工具（便于管理）：

   mkdir custom_tools  # 项目根目录执行

2. 核心依赖：确保已安装基础依赖（无需额外安装，框架自带工具基类）：

   pip install openmanus  # 若未安装框架核心包

二、步骤1：编写自定义工具类（核心）

所有自定义工具必须继承BaseTool基类，并实现3个核心要素：

• name：工具唯一名称（AI通过名称识别工具）
• description：工具描述（关键！AI通过描述判断何时调用该工具，需清晰说明「用途+输入格式」）
• run()：工具执行逻辑（接收输入参数，返回执行结果）

示例：开发「实时天气查询工具」

创建custom_tools/weather_tool.py文件，写入以下代码（含完整注释）：

# 导入框架核心基类和结果封装类
from openmanus.tools.base import BaseTool, ToolResult
# 按需导入第三方依赖（如请求网络需requests）
import requests

# 自定义工具类，必须继承BaseTool
class WeatherQueryTool(BaseTool):
    # 1. 工具唯一名称（不可重复，建议英文）
    name = "WeatherQueryTool"
    
    # 2. 工具描述（核心！需明确：用途+输入格式+输出说明）
    description = """
    用于查询指定城市的实时天气信息，输入格式为「城市名」（如：北京、上海），
    输出格式为「城市名 + 温度 + 天气状况」（如：北京 18℃ 晴）。
    仅当用户询问天气相关问题时调用该工具。
    """

    # 3. 工具执行逻辑，必须实现run方法
    def run(self, city: str) -> ToolResult:
        """
        参数说明：
        - city: 字符串，用户输入的城市名
        返回值：ToolResult对象（封装执行结果，必填）
        """
        try:
            # 步骤1：校验输入（可选，增强健壮性）
            if not city or len(city) > 10:
                return ToolResult(
                    success=False,  # 执行失败标记
                    content="输入无效！请输入正确的城市名（如：北京）"
                )
            
            # 步骤2：核心业务逻辑（调用免费天气API）
            # 替换为可靠的天气API，此处使用wttr.in（无需密钥）
            url = f"http://wttr.in/{city}?format=3"  # 精简格式：城市名: 天气 温度
            response = requests.get(url, timeout=10)
            
            # 步骤3：处理响应并封装结果
            if response.status_code == 200:
                weather_info = response.text.strip()
                return ToolResult(
                    success=True,  # 执行成功标记
                    content=f"✅ {weather_info}"  # 返回给AI的内容
                )
            else:
                return ToolResult(
                    success=False,
                    content=f"❌ 天气查询失败，API响应码：{response.status_code}"
                )
        
        # 异常处理（必加，避免工具崩溃）
        except requests.exceptions.Timeout:
            return ToolResult(success=False, content="❌ 网络超时，无法查询天气")
        except Exception as e:
            return ToolResult(success=False, content=f"❌ 查询出错：{str(e)}")

关键说明：

• ToolResult：框架规定的结果封装类，必须返回该类型，包含success（布尔值）和content（字符串）两个核心字段。

• description的精准性：AI完全依赖这段描述判断「是否调用该工具」，需明确：

  • 工具用途（如「查询指定城市实时天气」）
  • 输入格式（如「输入为城市名，例：北京」）
  • 适用场景（如「仅用户问天气时调用」）

三、步骤2：配置文件注册自定义工具

修改OpenManus的核心配置文件config.yaml（无则从config.example.yaml复制），将自定义工具添加到tools列表中：

1. 复制配置模板（首次需做）

cp config.example.yaml config.yaml  # 项目根目录执行

2. 编辑config.yaml，添加工具配置

找到tools节点，新增自定义工具的配置项：

# config.yaml 核心配置片段
llm:
  type: openai
  model: gpt-4-turbo
  api_key: "sk-xxxxxx"  # 替换为你的LLM密钥
  base_url: "https://api.openai.com/v1"

# 工具注册列表（内置工具 + 自定义工具）
tools:
  # 保留框架内置工具（按需取舍）
  - name: BrowserTool        # 浏览器工具
  - name: CodeExecutorTool   # 代码执行工具
  - name: FileTool           # 文件操作工具
  
  # 新增自定义工具（关键配置）
  - name: WeatherQueryTool   # 必须和工具类的name一致
    path: custom_tools/weather_tool.py  # 工具文件的绝对/相对路径
    enabled: true  # 是否启用该工具（默认true）

配置说明：

• name：必须和自定义工具类中定义的name完全一致（大小写敏感）。
• path：工具文件的路径，支持相对路径（相对于项目根目录）或绝对路径。
• enabled：是否启用该工具，设为false则AI不会调用。

四、步骤3：测试自定义工具

编写测试代码，验证自定义工具是否能被AI智能体识别并调用：

1. 创建测试文件test_custom_tool.py

import asyncio
from openmanus.agent import Agent  # 单智能体
from openmanus.config import Config  # 配置加载类

# 异步测试函数（OpenManus核心逻辑为异步）
async def test_weather_tool():
    # 步骤1：加载配置文件
    config = Config.from_file("config.yaml")
    
    # 步骤2：初始化AI智能体
    agent = Agent(config=config)
    
    # 步骤3：发送包含工具调用的任务指令
    task = "查询深圳市的实时天气"
    
    # 步骤4：执行任务并获取结果
    result = await agent.run(task)
    
    # 步骤5：打印结果
    print("=== 自定义工具调用结果 ===")
    print(result)

# 执行测试
if __name__ == "__main__":
    asyncio.run(test_weather_tool())

2. 运行测试代码

python test_custom_tool.py

预期输出：

=== 自定义工具调用结果 ===
✅ 深圳: 晴 25℃

五、进阶：支持多参数的自定义工具

若工具需要多个输入参数（如「根据城市和日期查询天气预报」），修改工具类的run方法即可：

示例：多参数天气工具

class WeatherQueryTool(BaseTool):
    name = "WeatherQueryTool"
    description = """
    查询指定城市指定日期的天气预报，输入格式为「城市名,日期」（日期格式：YYYY-MM-DD，例：北京,2026-01-20）。
    若未指定日期，则查询实时天气。
    """

    def run(self, input_str: str) -> ToolResult:
        # 解析多参数
        parts = input_str.split(",")
        city = parts[0].strip()
        date = parts[1].strip() if len(parts) > 1 else None
        
        # 核心逻辑（示例）
        if date:
            content = f"✅ {city} {date} 的天气预报：晴 22-30℃"
        else:
            content = f"✅ {city} 实时天气：晴 25℃"
        
        return ToolResult(success=True, content=content)

测试指令可改为：查询上海2026-01-20的天气预报。

六、常见问题与排查

问题1：AI不调用自定义工具

• 原因：description描述不清晰，AI无法判断何时调用；或工具名称/路径配置错误。

• 解决：

  1. 优化description，明确「触发条件+输入格式」；
  2. 检查config.yaml中工具name是否和类名一致；
  3. 测试时指令明确（如「用WeatherQueryTool查询北京天气」）。

问题2：工具执行报错「找不到模块」

• 原因：工具文件路径配置错误，或未继承BaseTool。

• 解决：

  1. 确认config.yaml中path是相对项目根目录的路径；
  2. 检查工具类是否正确导入from openmanus.tools.base import BaseTool。

问题3：工具返回结果为空

• 原因：run方法未正确返回ToolResult对象，或业务逻辑出错。

• 解决：

  1. 确保run方法最后return ToolResult(...)；
  2. 在run方法中添加日志（如print(city)），调试业务逻辑。

总结

1. 核心步骤：自定义工具开发需遵循「继承BaseTool→实现name/description/run→配置文件注册→测试验证」的流程，缺一不可。
2. 关键要点：description是AI调用工具的核心依据，需精准描述用途和输入格式；ToolResult是结果返回的标准格式，必须使用。
3. 扩展技巧：单参数工具直接接收字符串，多参数工具可通过分隔符（如逗号）解析输入，复杂场景可使用JSON格式传参。