funboost

Imported from ydf0509/funboost (AGENTS.md). Install upstream with npx skills add ydf0509/funboost. Copyright stays with the author.

AGENTS.md — Funboost 项目全貌（AI 专用指南）

Funboost = 一行 @boost 装饰器，让你的任意 Python 函数瞬间获得分布式调度 + FaaS 微服务 + 50 种消息队列 + 5种并发模式 + 30+ 种任务控制功能 + 工作流编排的能力。

AI 在回答 funboost 相关问题前，必须先阅读以下文件：

• @funboost/md_for_ai/funboost_ai_coding_编程指南_rules_and_skills.md （AI 编码铁律 + BoosterParams 全量字段 + 12 个技能模板）
• @funboost/md_for_ai/funboost源码速查for_ai.md （源码定位 + 行号速查 + Broker 映射表）
• @funboost/md_for_ai/funboost教程速查for_ai.md （教程文档速查）
• @funboost/md_for_ai/如何扩展增加新的中间件.md （Broker 扩展指南）
• @funboost_all_docs_and_codes.md （最高事实标准，完整教程 + 源码 + 示例，AI 生成代码前应优先检索此文件以消除幻觉）

二、核心 API

2.1 核心语法

# ✅ 推荐（必须）—— 使用 BoosterParams 对象传参
@boost(BoosterParams(queue_name="my_queue", qps=5, ...))
def my_task(x, y):
    return x + y

# ❌ 禁止使用老式写法
@boost("queue_name", qps=5)       # 过时写法，无代码补全

• BoosterParams 是一个 Pydantic 模型，绝对禁止臆造不存在的字段（如 timeout→正确的是 function_timeout；如 max_retries→正确的是 max_retry_times）
• 支持继承 BoosterParams 减少重复配置（详见 md_for_ai 中 Skills[8]）

2.2 发布消息

2.3 消费启动

注意：

• 连续启动多个消费者：func1.consume(); func2.consume() 即可，不要用 threading.Thread 包装
• AI 调用 funboost 脚本后必须及时 kill 进程（否则被无限消费循环阻塞），应使用 subprocess.run(['python', 'script.py'], timeout=30) 自动终止

2.4 任务上下文获取

from funboost import fct

fct.task_id                        # 当前任务 ID
fct.queue_name                     # 队列名
fct.function_result_status.run_times                     # 运行次数（含重试）
fct.full_msg                       # 完整消息体
fct.function_result_status.function_params                # 函数参数
fct.function_result_status         # FunctionResultStatus 对象
fct.logger                         # 当前任务 logger

禁止在函数参数中用 self 或 bind=True 取上下文（那是 Celery 的思维）

2.5 RPC 模式（获取消费结果）

# 同步
async_result = my_task.push(10, 20)
print(async_result.result)         # 阻塞等待结果

# 异步
aio_result = await my_task.aio_push("http://example.com")
result_status = await AioAsyncResult(aio_result.task_id).status_and_result

条件：BoosterParams 中需设置 is_using_rpc_mode=True ,必须切记这条规则。

2.6 定时任务

from funboost import ApsJobAdder

ApsJobAdder(func, job_store_kind='redis').add_push_job(
    trigger='cron', hour=2, minute=0, kwargs={"type": "daily"}, id='job1'
)
ApsJobAdder(func, job_store_kind='redis').add_push_job(
    trigger='interval', seconds=30, args=(1, 2), id='job2'
)

禁止直接使用 apscheduler.add_job 去执行消费函数，必须使用 ApsJobAdder

2.7 工作流编排

from funboost.workflow import chain, group, chord, WorkflowBoosterParams

workflow = chain(
    download.s("video.mp4"),
    chord(
        group(process.s(resolution=r) for r in ['360p', '720p', '1080p']),
        notify.s(user_id=1001)
    )
)
result = workflow.apply()

2.8 FaaS 微服务

from fastapi import FastAPI
from funboost.faas import fastapi_router

app = FastAPI()
app.include_router(fastapi_router)   # 一键获得 /funboost/publish 等接口

# Flask 版本
from funboost.faas import flask_blueprint

三、BoosterParams 核心字段速览（50+ 字段）

以上为高频字段摘要。完整 50+ 字段定义见 @funboost/md_for_ai/funboost_ai_coding_编程指南_rules_and_skills.md

异步消费要点

• 推荐 concurrent_mode=ConcurrentModeEnum.ASYNC，必须用 await func.aio_push() 发布
• 异步 RPC 结果：必须用 AioAsyncResult + await，禁止用 AsyncResult.result

消费异构系统消息

• 设置 should_check_publish_func_params=False
• 消费函数定义为 def task_fun(**kwargs):（用 **kwargs 解包接收所有字段）
• 禁止用单个 msg 参数接收整个 JSON（如 def task_fun(msg):），必须用 **msg 或 **kwargs

四、Broker 中间件

Redis 模式选型

Broker 扩展

• register_custom_broker — 注册全新中间件（contrib/register_custom_broker_contrib/）
• consumer_override_cls / publisher_override_cls — Mixin 混入定制现有 broker
• 详见 @funboost/md_for_ai/如何扩展增加新的中间件.md

五、扩展模块（Contrib）

六、测试体系

⚠️ 禁止直接 python script.py 跑测试脚本，必须用 subprocess.run + timeout 自动终止。

七、配置机制

使用 is_using_rpc_mode=True、is_using_distributed_frequency_control=True 或 do_task_filtering=True 时必须配置 Redis

十、项目结构

funboost/
├── core/              # 核心引擎（booster.py / func_params_model.py / current_task.py）
├── consumers/         # 49 个消费者实现
├── publishers/        # 45 个发布者实现
├── factories/         # broker_kind → (Publisher, Consumer) 映射
├── concurrent_pool/   # 5 种并发池
├── workflow/          # Chain / Group / Chord
├── faas/              # FastAPI / Flask / Django 集成
├── timing_job/        # APScheduler 集成
├── contrib/           # 扩展（Mixin / 自定义 Broker / 爬虫 / CDC）
├── funweb/            # Web 管理界面（`python -m funboost.funweb.app` 启动）
└── md_for_ai/         # AI 参考文档

十一、给 AI 的特别提醒 for funboost

1. funboost_all_docs_and_codes.md 是最高事实标准 — 生成任何代码前，先检索它
2. 禁止臆造参数名 — BoosterParams 的所有字段都是固定的，必须从 md_for_ai 中查找正确字段名
3. 禁止使用 Celery 思维 — 获取上下文用 fct，不要用 self 或 bind=True
4. push 和 publish 的区别 — push 传业务参数，publish 传字典 + 框架控制参数
5. 实例方法的 push 语法特殊 — 必须写成 ClassName.method.push(obj, args)，第一个参数传对象实例

十二、 ai自己启动运行脚本的注意事项

1. 消费启动后不会自动结束 — 有两种方式运行和检查：
   • 第一种方式（CMD/PowerShell + timeout）：AI 直接用命令行运行，靠 timeout 超时自动终止进程，脚本里不需要手写 os._exit。AI 通过检查控制台输出内容判断是否符合预期。
     • CMD 语法：set PYTHONPATH=项目根目录 & cmd /c "timeout /t 30 /nobreak >nul & python script.py"
     • PowerShell 语法：$env:PYTHONPATH="项目根目录"; cmd /c "timeout /t 30 /nobreak >nul & python script.py"
     • timeout 时间评估：AI 应自行评估合理时间，一般要大于 10 秒（框架启动需要一定时间）。根据消息个数、qps、并发数量、函数耗时预估：
       • 默认值：concurrent_num=50，qps=None（不限制）
       • 不设置 qps：实际吞吐量 ≈ concurrent_num / 函数耗时
       • 设置 qps：实际吞吐量 ≈ min(qps, concurrent_num / 函数耗时)
       • 合理时间 ≈ 框架启动时间(5-10秒) + (消息个数 / 实际吞吐量) + 缓冲时间(2-5秒)
       • timeout 最大不超过 50 秒。
   • 第二种方式（推荐，参考 for_ai_run_demo.py）：脚本开头设置 LOG_PATH、PRINT_WRTIE_FILE_NAME、SYS_STD_FILE_NAME 三个环境变量，脚本末尾写 time.sleep(休眠秒数); os._exit(66) 自动退出。AI 运行后主动去读取 D:\pythonlogs\ai_console_outs 下生成的 print 和 std 文件来检查输出，这种方式不需要用 subprocess.run + timeout。
     • 运行前必须设置：$env:PYTHONPATH="项目根目录"（PowerShell）或 set PYTHONPATH=项目根目录（CMD）
     • 重要：每次运行时，PRINT_WRTIE_FILE_NAME 和 SYS_STD_FILE_NAME 必须设置不同的文件名后缀（如加时间戳或序号），防止 AI 读取到上一次的老日志内容。
     • time.sleep 时间评估：AI 应自行评估合理休眠时间，一般要大于 10 秒（框架启动需要一定时间）。根据消息个数、qps、并发数量、函数耗时预估：
       • 默认值：concurrent_num=50，qps=None（不限制）
       • 不设置 qps：实际吞吐量 ≈ concurrent_num / 函数耗时
       • 设置 qps：实际吞吐量 ≈ min(qps, concurrent_num / 函数耗时)
       • 合理时间 ≈ 框架启动时间(5-10秒) + (消息个数 / 实际吞吐量) + 缓冲时间(2-5秒)
2. ai启动funboost任何脚本之前，一定先设置 PYTHONPATH=当前项目根目录
   • CMD 语法：set PYTHONPATH=D:\codes\funboost
   • PowerShell 语法：$env:PYTHONPATH="D:\codes\funboost"

十三、 ai 新写代码位置规范

1. ai如果是被要求修改项目框架，写完后，请写单元测试/回归测试，请放在tests/ai_codes/regression_testing文件件下面，ai要运行和修改相关的测试用例
2. 如果是让ai写demo或者例子或者测试验证脚本，而不是修改项目和框架本身，ai要把代码写到 tests/ai_codes/ai_demos/ ，并且是在 ai_demos文件夹下新建合理的子文件夹，在子文件夹里面新增例子脚本。

十四、 ai禁止行为

1. 禁止ai flush redis，ai不要主动去清空redis。

十五 funboost文档仓库

D:/codes/funboost_docs/ ,文档教程在 source/articles 目录下

十六 funboost开发和使用的skills在 .agents/skills目录下