Botflow

0.2.0 alpha

• 数据流编程框架
• 在协程和线程池中并行化
• 基于类型和内容的路由函数
• 使用 Jupyter Notebook 进行交互式编程

要求

Graphviz <http://brewformulas.org/Graphviz>_。

安装

使用 pip 安装和更新：

pip install -U botflow

文档

http://botflow.readthedocs.io

什么是数据流编程？

所有函数通过管道（队列）连接，并通过数据进行通信。

当数据进入时，函数会被调用并返回结果。

可以联想到 Unix 中的管道操作：ls|grep|sed。

优点：

#. 解耦数据与功能 #. 易于复用

Botflow 提供了管道和路由功能，使数据流编程和强大的数据流处理变得更加容易。

Botflow 是...

• 简单

Botflow 易于使用和维护，不需要配置文件，并且了解 asyncio 以及如何并行化计算。

以下是一个你可以实现的简单应用：

每 2 秒加载一次比特币价格。优势价格聚合示例可以在 这里 <https://github.com/kkyon/Botflow/tree/master/examples>_ 找到。

.. code-block:: python

from botflow import *


def main():
    Pipe(

        Timer(delay=2),  # 每 2 秒向管道发送定时器数据
        "http://api.coindesk.com/v1/bpi/currentprice.json",  # 当定时器触发时，将 URL 发送到管道
        HttpLoader(),  # 读取 URL 并加载 HTTP 响应
        lambda r: r.json['bpi']['USD']['rate_float'],  # 读取 HTTP 响应并解析为 JSON
        print,  # 打印输出

    )

    Bot.render('ex_output/simple_bitcoin_price')
    Bot.run()
main()

或者以链式风格编写

.. code-block:: python

from botflow import *
p_cd_bitcoin=Pipe().Timer(delay=2).Loop("http://api.coindesk.com/v1/bpi/currentprice.json")\
            .HttpLoader().Map(lambda r: r.json['bpi']['USD']['rate_float']).Map(print)

p_cd_bitcoin.run()

• 流图 借助渲染功能： Bot.render('bitcoin_arbitrage') Botflow 会将数据流网络渲染成一个 Graphviz 图像。 以下是 Botflow 生成的流图。它聚合了 6 家交易所的比特币价格，用于交易。

.. image:: docs/bitcoin_arbitrage.png :width: 400

• 快速 节点将以并行方式运行，在处理流数据时表现优异。 :Web 爬虫: Botflow 的速度是 Scrapy 的 10 倍。

• 可重放

启用重放模式后： config.replay_mode=True 当第 N 步出现异常时，你无需从步骤 1 重新运行到 N。 Botflow 会从最近完成的节点（通常是第 N-1 步）开始重放数据。 这将在开发阶段节省大量时间。

路线图

• 将添加 HTTP 服务器支持（REST、WebSocket）。
• 将支持服务器端机器学习模型在线部署。
• 完成 API 参考文档。
• 将项目更名为 Botflow。?

关于 Botflow 的更多信息

数据流编程通常应用于结构化数据流，用于过滤、转换、聚合（如计算统计信息）或调用其他程序。

Botflow 有几个基本概念来实现数据流编程。

• 源 它将数据流输入到管道中。

  • 定时器: 它会按照定时器参数（延迟、最长时间等）向管道发送消息。
  • Pipe.run: 你可以使用 Pipe.run 来触发数据进入管道。默认情况下，它会输入整数 0。

• 函数 它是可调用的单元。任何可调用的函数和对象都可以作为节点。它由数据驱动。自定义函数可用作映射单元。 有一些内置节点：

  • Fetch: （别名：HttpLoader）获取 URL 并返回 HTTP 响应
  • AioFile: 用于文件 I/O。
  • SpeedLimit: 限制流速
  • Delay: 特定秒数的延迟。
  • Map ：作为转换单元工作。
  • Filter ：如果数据不符合某些条件，则将其从管道中丢弃。
  • Flat ：如果数据不符合某些条件，则将其从管道中丢弃。

• 路由 它用于创建复杂的数据流网络，而不仅仅是一个主流程。Botflow 可以在路由内部嵌套路由。 这是一个强大的概念。 有一些预置的路由：

  • Pipe: 这是程序的主要流处理过程。所有单元都在其中工作。
  • Tee : （别名：Branch）将父管道中的数据复制到子管道中，形成分支。
  • Zip : 将多个管道的结果合并为列表。
  • Link: （别名：LinkTo）将流路由到任何节点或路由，以形成循环。

所有单元（管道、节点、路由）通过队列进行通信，并在协程中执行并行计算。 这种抽象使得 Botflow 即使只具备有限的 asyncio 知识也能使用。

贡献

捐赠

链接

Botflow 快速上手指南

Botflow 是一个基于协程和线程池并行的数据流编程框架。它通过管道（队列）连接函数，实现数据驱动的执行模式，类似 Unix 的管道操作（ls|grep|sed）。支持 Jupyter Notebook 交互式编程，并能自动生成数据流可视化图表。

环境准备

在开始之前，请确保您的系统已安装以下依赖：

• Graphviz: 用于生成数据流可视化图表。
  • macOS: brew install graphviz
  • Ubuntu/Debian: sudo apt-get install graphviz
  • Windows: 下载安装包安装并配置环境变量。
• Python: 建议 Python 3.6+（需支持 asyncio）。

安装步骤

使用 pip 安装或更新 Botflow：

pip install -U botflow

提示：国内用户如遇下载缓慢，可使用清华源加速：

pip install -U botflow -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

Botflow 的核心概念是 Pipe（管道），数据在其中流动并被各个节点（Node）处理。

示例：每 2 秒获取一次比特币价格

以下代码展示了如何构建一个简单的数据流：定时器触发 -> 发送 URL -> 加载 HTTP 响应 -> 解析 JSON -> 打印结果。

from botflow import *


def main():
    Pipe(
        Timer(delay=2),  # 每 2 秒向管道发送计时器数据
        "http://api.coindesk.com/v1/bpi/currentprice.json",  # 计时器触发时发送 URL
        HttpLoader(),  # 读取 URL 并加载 HTTP 响应
        lambda r: r.json['bpi']['USD']['rate_float'],  # 解析 JSON 获取美元价格
        print,  # 打印结果
    )

    # 渲染数据流图（可选，需安装 Graphviz）
    Bot.render('ex_output/simple_bitcoin_price')
    
    # 运行管道
    Bot.run()

if __name__ == '__main__':
    main()

链式写法

Botflow 也支持更简洁的链式调用风格：

from botflow import *

# 定义管道流程
p_cd_bitcoin = Pipe()\
    .Timer(delay=2)\
    .Loop("http://api.coindesk.com/v1/bpi/currentprice.json")\
    .HttpLoader()\
    .Map(lambda r: r.json['bpi']['USD']['rate_float'])\
    .Map(print)

# 直接运行
p_cd_bitcoin.run()

核心特性速览

• 可视化: 调用 Bot.render('graph_name') 即可将数据流网络生成 Graphviz 图片，方便调试复杂流程。
• 并行执行: 节点在协程中并行运行，适合处理流式数据，性能优异。
• 可重放 (Replay): 开发阶段开启 config.replay_mode=True，若第 N 步报错，可从最近完成的节点（通常是 N-1）重放数据，无需从头运行，大幅提升调试效率。