PraisonAI Recipe Async Jobs：把 Agent 任务提交到 Jobs Server 异步执行完整指南

PraisonAI Recipe Async Jobs把 Agent 任务提交到 Jobs Server 异步执行完整指南【免费下载链接】PraisonAIPraisonAI — Hire a 24/7 AI Workforce. Stop writing boilerplate and start shipping autonomous self-improving agents that research, plan, code, and execute tasks. Deployed in 5 lines of code with built-in memory, RAG, and support for 100 LLMs.项目地址: https://gitcode.com/GitHub_Trending/pr/PraisonAI导读本文围绕仓库文档 recipes_jobs/README.md 展开系统讲解 PraisonAI 的 Recipe Async Jobs 能力将耗时较长的 Agent/Recipe 任务以异步 Job 的形式提交到独立的 Jobs Server 上执行支持状态轮询、实时进度流、Webhook 完成通知与幂等去重。读完本文你将掌握 Jobs Server 的启动与配置、Python SDK 与 CLI 的完整用法以及底层 FastAPI 路由、执行器与存储层的实现原理可直接在本地把长任务改造成可靠、可追踪的异步作业。什么是 Recipe Async JobsPraisonAI 是一个让开发者用少量代码编排自主 AI 代理Agent的框架。当任务的执行时间较长例如深度研究、多步工作流、复杂 Recipe时同步等待会阻塞调用方。Recipe Async Jobs 正是为此设计它把任务提交请求投递给一个独立的 Jobs Server由服务器在后台调度执行客户端可以随时查询状态、拉取结果或订阅进度流且 Job 状态跨进程重启依然可恢复。从仓库结构看异步作业的核心实现集中在 src/praisonai/praisonai/jobs/ 目录下由四个模块协作模块职责server.pyFastAPI 应用工厂create_app负责中间件、鉴权与存储选择router.py暴露/api/v1/runs下的提交、查询、取消、流式等 HTTP 端点executor.py后台并发执行 Job处理超时、取消、进度回调和 Webhookstore.pyJob 存储后端抽象提供内存版与 SQLite 持久化版与之配套的示例代码位于 recipes_jobs/example_jobs.py演示了从提交到轮询再到取结果的完整流程。前置条件与安装根据原文档使用 Recipe Async Jobs 需要安装以下依赖并配置 API Keypip install praisonai praisonaiagents httpx uvicorn export OPENAI_API_KEYyour-api-keypraisonai核心框架包含 Jobs Server 与 CLIpraisonaiagents底层 Agent 运行库Job 执行时通过 executor.py 中的arun/PraisonAI.run驱动httpx服务器发送 Webhook 通知时使用见 executor.py 中_send_webhook的实现uvicornJobs Server 的 ASGI 服务器缺少时 server.py 会提示pip install uvicorn。启动 Jobs Server按照原文档在终端执行python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory这条命令以工厂模式--factory加载 server.py 中的create_app()默认绑定127.0.0.1:8005。应用启动时通过lifespan钩子调用executor.start()拉起后台执行器并在关闭时优雅停止。存储后端与持久化Jobs Server 的服务端持久化特性取决于存储后端的选择逻辑在 server.py 的_build_default_store()中设置环境变量PRAISONAI_JOBS_DB_PATH例如/var/lib/praisonai/jobs.db时使用SqliteJobStoreJob 状态与幂等键跨重启保留未设置时默认使用InMemoryJobStore(max_jobs1000)重启后 Job 全部丢失仅适合本地开发当ENVIRONMENTproduction且未配置PRAISONAI_JOBS_DB_PATH时服务器会拒绝启动避免生产环境静默丢失幂等键。鉴权与 CORS从 server.py 可以读出两条安全策略绑定地址非 localhost由PRAISONAI_JOBS_BIND_HOST决定时必须设置PRAISONAI_JOBS_API_KEY否则除/health外所有请求返回 401设置后则通过praisonai._api_auth中的 API Key 中间件校验请求CORS 默认只放行本地开发地址localhost:3000/8000、127.0.0.1:3000/8000生产环境ENVIRONMENTproduction默认不放行任何来源可通过JOBS_CORS_ORIGINS逗号分隔显式配置。可配置环境变量结合 server.py 与 executor.py服务器支持的环境变量如下环境变量默认值说明PRAISONAI_JOBS_DB_PATH未设置设置后启用 SQLite 持久化存储PRAISONAI_JOBS_API_KEY未设置非本机绑定时必填的 API KeyPRAISONAI_JOBS_BIND_HOST127.0.0.1绑定地址决定是否需要鉴权PRAISONAI_MAX_CONCURRENT_JOBS10并发执行的任务数上限PRAISONAI_JOB_TIMEOUT3600单个 Job 默认超时秒JOBS_CORS_ORIGINS未设置允许的 CORS 来源列表逗号分隔ENVIRONMENT未设置置为production时启用更严格的安全默认值启动后可通过GET /health查看服务器状态、存储类型与执行器统计通过GET /stats查看执行器与存储层统计信息。Python SDK 示例提交与轮询原文档要求运行仓库自带的示例脚本python example_jobs.py该脚本位于 recipes_jobs/example_jobs.py其完整流程展示了 SDK 编程的核心用法创建处理器实例化JobsHandler指定服务器地址与verbose开关from praisonai.cli.features.jobs import JobsHandler handler JobsHandler( api_urlhttp://127.0.0.1:8005, verboseTrue )提交任务调用handler.submit()支持framework、timeout、wait、output_json等参数返回结果中携带job_id与statusresult handler.submit( promptWhat are the key trends in AI for 2024?, frameworkpraisonai, timeout300, waitFalse, # 不阻塞等待改为手动轮询 output_jsonFalse ) job_id result.get(job_id)轮询状态循环调用handler.status(job_id)读取status与progress百分比在终态succeeded/failed/cancelled时退出示例中以 5 秒为间隔最多轮询 60 次for i in range(max_polls): status_result handler.status(job_id) status status_result.get(status) progress status_result.get(progress, 0) if status in [succeeded, failed, cancelled]: break time.sleep(5)获取结果handler.result(job_id)返回终态与result内容示例打印结果预览的前 200 个字符。列出任务handler.list_jobs(page1, page_size10)分页展示全部 Job。CLI 命令全解原文档提供了完整的命令行用法以下逐一展开并补充参数说明。提交任务# 基础提交 praisonai run submit Analyze AI trends # 携带 Recipe 提交 praisonai run submit Analyze news --recipe news-analyzer # 等待完成阻塞直到终态 praisonai run submit Quick task --wait # 流式查看进度 praisonai run submit Long task --stream # 注册 Webhook 回调 praisonai run submit Task --webhook-url https://example.com/callback # 幂等提交防止重复 praisonai run submit Task --idempotency-key order-123 # 附加元数据 praisonai run submit Task --metadata userjohn --metadata priorityhigh其中--recipe指定要执行的 Recipe 名称与--agent-file互斥见 models.py 中的JobSubmitRequest--metadata可多次传入作为自定义键值附加到 Job 上。查询状态praisonai run status job_id praisonai run status job_id --json返回内容包括当前状态、进度百分比、当前步骤、创建/开始/完成时间戳、agent_id、run_id用于链路追踪、session_id以及retry_after建议轮询间隔。其中queued状态建议 2 秒后重试running状态建议 5 秒后重试见 models.py 中to_status_response()。获取结果praisonai run result job_id praisonai run result job_id --json返回终态、result、metrics、duration_seconds与错误信息。若 Job 尚未进入终态HTTP 层会返回 409 Conflict。流式查看进度praisonai run stream job_id订阅基于 Server-Sent EventsSSE的实时进度流事件类型包括status、progress、result、error、cancelled与心跳heartbeat每 5 秒一次见 router.py 的stream_job端点。列出任务praisonai run list praisonai run list --status running支持按状态过滤queued/running/succeeded/failed/cancelled也支持按session_id过滤与分页page、page_size每页最多 100 条见 router.py 的list_jobs端点。取消任务praisonai run cancel job_id对非终态 Job 生效执行器会将 Job 标记为cancelled并取消对应 asyncio 任务见 executor.py 的cancel()对已完成的 Job 取消会返回 409。底层 HTTP API 与状态机CLI 与 SDK 底层都通过 router.py 定义的 HTTP API 与服务器通信路由前缀为/api/v1/runs方法路径说明POST/api/v1/runs提交 Job返回 202 与Location/Retry-After头GET/api/v1/runs分页列出 Job支持status、session_id过滤GET/api/v1/runs/{job_id}查询状态GET/api/v1/runs/{job_id}/result获取结果未完成时返回 409POST/api/v1/runs/{job_id}/cancel取消任务DELETE/api/v1/runs/{job_id}删除已完成的 Job运行中返回 409GET/api/v1/runs/{job_id}/streamSSE 实时进度流Job 的完整状态机在 models.py 的JobStatus中定义queued → running → succeeded / failed / cancelled。每个 Job 内部维护进度字段progress_percentage、current_step、steps_completed、steps_total由执行器在初始化 Agent → 解析 Recipe → 执行 → 收尾等阶段逐级更新并通过进度回调扇出给所有订阅者——同一 Job 可以被多个客户端如多个 Dashboard 标签页同时订阅。执行器的并发与背压executor.py 中JobExecutor用 asyncio 信号量控制并发度默认max_concurrent10并用max_queued默认max_concurrent * 10作为准入上限当排队与运行中的任务总数达到上限时提交请求会抛出JobQueueFull由路由层映射为 HTTP 503 并携带Retry-After响应头向调用方传递诚实的背压而非无限堆积内存。此外执行器每 300 秒清理一次超过 24 小时的已完成 Jobcleanup_old_jobs(max_age_seconds86400)防止存储无限膨胀。关键特性深入原文档总结了五大特性这里结合源码逐项印证。服务端持久化Server-basedJob 由独立服务器托管存储层可切换为 SQLitePRAISONAI_JOBS_DB_PATH状态与幂等键跨重启保留。存储层抽象定义在 store.py 的JobStore基类中save_if_absent()保证同一幂等键只插入一次SqliteJobStore在数据库层利用唯一约束实现真正的原子去重避免并发提交竞态。Webhook 完成通知提交时通过--webhook-url指定回调地址Job 成功或失败后执行器都会向该地址 POST JSON 载荷包含job_id、status、result/error、completed_at与duration_seconds见 executor.py 的_send_webhook()。值得注意的是models.py 对webhook_url做了严格的 SSRF 防护校验仅允许http/https协议拒绝 localhost 地址以及解析到私网、回环、链路本地或组播地址的域名避免服务器被诱导访问内网资源。幂等去重Idempotency通过--idempotency-key或 HTTP 头Idempotency-Key提交时若该键已存在则直接返回已有 Job 而不是重复执行。这一机制在提交端点中实现先查存储再通过save_if_absent原子认领键确保副作用只执行一次适用于支付订单、定时报告等对重复执行敏感的场景见 router.py 的submit_job与 store.py 的save_if_absent。原文档还提到提交请求中可携带idempotency_scopenone/session/global以控制去重范围。实时进度流Streaming/api/v1/runs/{job_id}/stream基于 SSE 推送status、progress、result等事件进度回调采用每 Job 多订阅者模型多个客户端可同时订阅同一 Job 的进度且各自断开时只注销自己的回调互不影响见 executor.py 的register_progress_callback/unregister_progress_callback。元数据Metadata提交请求支持config等自定义字段CLI 通过--metadata keyvalue附加业务数据Job 会原样保存并在查询时返回便于在作业系统中打标签、做归属统计或按业务维度筛选。Safe Defaults 参考原文档给出的安全默认值如下这也是 models.py 与 server.py 中实际使用的默认值设置默认值说明timeout3600单个 Job 最长执行时间1 小时超时标记为 failedapi_urlhttp://127.0.0.1:8005Jobs Server 地址在此基础上结合源码还可补充几个重要默认值并发执行上限max_concurrent为 10、准入上限max_queued为 100max_concurrent * 10、Job ID 格式为run_ 12 位十六进制见 models.py 的Job定义、进度字段范围被约束在 0100 之间、列表接口默认每页 20 条且最大 100 条。小结与使用建议Recipe Async Jobs 让 PraisonAI 的长任务具备了完整的异步作业生命周期通过praisonai run系列命令或JobsHandlerSDK 提交任务用status/result/stream追踪执行用cancel中断用 Webhook 与幂等键对接外部系统。本地验证时建议按以下顺序操作安装依赖并导出OPENAI_API_KEY启动 Jobs Server默认内存存储即可若需跨重启保留任务设置PRAISONAI_JOBS_DB_PATH运行 example_jobs.py 或 CLI 命令完成提交、轮询与取结果的闭环涉及外部回调时为--webhook-url提供公网可达的 HTTPS 地址内网地址会被服务器校验拒绝需要对接生产时将ENVIRONMENTproduction并显式配置PRAISONAI_JOBS_API_KEY与PRAISONAI_JOBS_DB_PATH遵循服务器的安全默认值。更完整的参考实现可继续阅读 jobs 模块源码 与示例目录 recipes_jobs。【免费下载链接】PraisonAIPraisonAI — Hire a 24/7 AI Workforce. Stop writing boilerplate and start shipping autonomous self-improving agents that research, plan, code, and execute tasks. Deployed in 5 lines of code with built-in memory, RAG, and support for 100 LLMs.项目地址: https://gitcode.com/GitHub_Trending/pr/PraisonAI创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考