如果你同时维护多个 AI 代理一个做数据分析一个写文案还有一个负责代码审查一段时间后你会发现最麻烦的根本不是单个模型能不能跑通而是这些代理现在分别在处理什么任务哪个卡住了哪个需要你确认才能继续同一批任务重跑时哪些结果应该保留、哪些必须清空文章标题里这个方案就是用来解决这个问题的Human task board for my agents一个面向多个 AI 代理的人类任务看板。它不是某个公司的商业 SaaS也不是固定版本的开源项目而是一套可以自己实现的轻量任务协调方案。核心目标很明确让人能随时插手让代理按队列执行让每个任务都有状态、有记录、有结果还能被安全地中断和恢复。本文会带你完整过一遍任务看板的数据模型怎么设计、服务怎么启动、代理怎么通过 API 接入、批量任务怎么排队、中断与审批节点怎么实现以及最常见的问题怎么排查。不需要独立 GPU一台普通开发机就能跑完整个链路。适合正在本地搭多代理工具链、需要人工审批节点、或者想给代理集群加一个统一任务总线的开发者。1. 核心能力速览在动手之前先把这套 Human Task Board 方案的能力边界说清楚。能力项说明项目定位面向多个 AI 代理的人类任务管理与协调看板核心目标任务登记、状态跟踪、人工审批、代理中断恢复、批量队列调度建议技术栈Python 3.10、FastAPI、SQLite、轻量前端页面硬件要求普通开发机即可不需要独立 GPU部署方式命令行启动或 Docker 容器启动启动方式uvicorn app:app或docker compose up是否支持 API支持提供 REST API 供代理端轮询或回调是否支持批量任务支持可按队列顺序执行并限制并发数人工审批节点支持代理可上报waiting_human状态等待确认中断与恢复支持可对超时任务做中断和状态回滚适合场景本地多代理实验、内部工具链、审核型任务流核心思路不是做一个复杂的分布式任务调度平台而是先把“人”放进代理工作流里。代理负责干活看板负责记录和暂停人负责审批和兜底。这样即使某个代理突然跑偏你也可以在任务提交后的任意节点打断它。2. 适用场景与使用边界先说实话不是所有代理场景都需要一个人类任务看板。如果你只有一两个代理且任务完全自动化、不需要人在中间确认直接用代码把任务串起来可能更轻。但如果你同时管理多个代理并且有以下任一需求这个看板就有价值任务需要按批次提交每个批次可能有几十上百条其中一部分任务执行到 80% 时需要人类确认才能继续代理执行结果必须存档方便复盘和审计代理异常退出后任务不能卡死在“执行中”状态多个代理要共享同一个任务来源而不是各接各的输入。边界也要说清楚如果业务对延迟要求极高比如用户在线等待秒级返回那这个看板会引入轮询和状态更新开销不一定合适。如果任务完全不需要人参与那么审批节点和中断机制也是多余的设计。需要特别强调的是这类协作看板经常会和内容生成、文档处理、数据分析等能力组合使用执行过程中可能涉及真实业务数据、用户隐私或版权素材。请务必确保输入数据有合法来源代理对数据的访问范围可控批量任务不要对第三方接口造成过大压力。涉及人脸、声音、品牌素材等敏感内容时必须确认授权后再进入任务队列。3. 环境准备与前置条件这是一套纯服务端方案对硬件基本没有要求。准备清单如下操作系统Windows 10/11、macOS、Linux 均可Python 3.10 或更高版本数据库SQLitePython 自带不需要额外安装数据库服务依赖库FastAPI、Uvicorn、Pydantic、Requests。建议使用虚拟环境安装依赖避免污染系统 Python。如果你的机器上同时有多个 Python 版本先用python --version确认版本。# 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate # 安装依赖 pip install fastapi uvicorn pydantic requests端口方面默认使用8000如果本地端口被占用可以修改启动参数换成8010或其他空闲端口。磁盘空间按任务量估算单条任务记录很小日志和结果文件才是大头。建议把data目录和日志目录单独挂载方便备份和清理。4. 任务看板数据模型与状态设计任务看板的核心是状态机。只要状态定义清晰后面加批量、加审批、加中断都会很顺。建议的任务状态状态含义pending任务已创建等待代理领取running代理已领取任务正在执行waiting_human任务需要人工确认暂停执行completed任务执行完成failed任务执行失败cancelled任务被取消状态流转关系pending - running - waiting_human - running - completed ↘ failed ↘ cancelled其中waiting_human是这个看板区别于普通任务队列的关键。代理在执行过程中发现需要人类决策时可以把任务置为该状态人的工作变成“在合适的时机看板、做决定、放行或终止”。数据表结构示意CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, title TEXT NOT NULL, description TEXT, agent_id TEXT, status TEXT DEFAULT pending, priority INTEGER DEFAULT 3, requires_approval INTEGER DEFAULT 0, approval_status TEXT, result TEXT, progress INTEGER DEFAULT 0, created_at TEXT, updated_at TEXT );agent_id字段用来区分任务由哪个代理执行。requires_approval用来标记是否必须在执行前或执行中经过人工确认。approval_status记录审批状态比如pending、approved、rejected。如果任务需要更复杂的阶段记录可以单独加一张task_logs表保存每个阶段的时间戳、状态变化和备注。这张表的价值在排查问题时非常明显能直接回答“任务到底在哪一步出了问题”。5. 启动服务与 Web 看板访问服务部分我用 FastAPI 做一个最小实现。下面这段代码是核心路由不是完整生产代码但可以直接跑。# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import sqlite3 import uuid from datetime import datetime app FastAPI(titleHuman Task Board) DB_PATH ./data/tasks.db def get_db(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn class TaskCreate(BaseModel): title: str description: str agent_id: Optional[str] None priority: int 3 requires_approval: bool False class TaskUpdate(BaseModel): status: Optional[str] None result: Optional[str] None progress: Optional[int] None remark: Optional[str] None app.on_event(startup) def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, title TEXT NOT NULL, description TEXT, agent_id TEXT, status TEXT DEFAULT pending, priority INTEGER DEFAULT 3, requires_approval INTEGER DEFAULT 0, approval_status TEXT, result TEXT, progress INTEGER DEFAULT 0, created_at TEXT, updated_at TEXT ) ) conn.commit() conn.close() app.post(/api/tasks) def create_task(task: TaskCreate): task_id str(uuid.uuid4())[:8] conn get_db() conn.execute( INSERT INTO tasks (id, title, description, agent_id, status, priority, requires_approval, created_at) VALUES (?, ?, ?, ?, pending, ?, ?, ?) , (task_id, task.title, task.description, task.agent_id, task.priority, task.requires_approval, datetime.now().isoformat()), ) conn.commit() conn.close() return {task_id: task_id, status: pending} app.patch(/api/tasks/{task_id}) def update_task(task_id: str, update: TaskUpdate): conn get_db() task conn.execute(SELECT * FROM tasks WHERE id ?, (task_id,)).fetchone() if not task: conn.close() raise HTTPException(status_code404, detailtask not found) fields update.model_dump(exclude_noneTrue) if fields: fields[updated_at] datetime.now().isoformat() sql UPDATE tasks SET , .join(f{k} ? for k in fields) WHERE id ? conn.execute(sql, (*fields.values(), task_id)) conn.commit() conn.close() return {task_id: task_id, **fields} app.get(/api/tasks) def list_tasks(status: Optional[str] None, agent_id: Optional[str] None): conn get_db() sql SELECT * FROM tasks args [] conditions [] if status: conditions.append(status ?) args.append(status) if agent_id: conditions.append(agent_id ?) args.append(agent_id) if conditions: sql WHERE AND .join(conditions) sql ORDER BY created_at DESC rows conn.execute(sql, args).fetchall() conn.close() return [dict(r) for r in rows]启动命令mkdir -p data uvicorn app:app --host 127.0.0.1 --port 8000启动后访问API 文档http://127.0.0.1:8000/docs看板接口http://127.0.0.1:8000/api/tasks第一次启动后建议先用 API 文档里的POST /api/tasks创建一个测试任务再访问GET /api/tasks确认数据写入了。接口能通看板底座就搭好了。前端页面不是必需项但你可以用静态 HTML 扩展一个看板页面定时轮询/api/tasks展示任务列表和状态。这样代理端自动刷新任务人工端也能实时看到进度。6. 代理接入与任务状态上报服务端只是看板真正干活的还是代理。代理端需要做三件事定期拉取属于自己且状态为pending的任务领取任务后把状态改为running执行完成后上报completed和result。下面的 Python 示例是代理客户端的最简实现逻辑上任何语言都能照着写。# agent_client.py import requests import time import random BASE_URL http://127.0.0.1:8000 AGENT_ID analyst-agent def fetch_pending_tasks(): resp requests.get( f{BASE_URL}/api/tasks, params{agent_id: AGENT_ID, status: pending}, timeout10, ) resp.raise_for_status() return resp.json() def update_task(task_id, **kwargs): resp requests.patch( f{BASE_URL}/api/tasks/{task_id}, jsonkwargs, timeout10, ) resp.raise_for_status() return resp.json() def run_agent(): while True: tasks fetch_pending_tasks() for task in tasks: task_id task[id] print(f领取任务: {task_id} - {task[title]}) update_task(task_id, statusrunning, progress10) # 模拟代理执行替换成真实代理逻辑 time.sleep(random.randint(1, 3)) update_task( task_id, statuscompleted, progress100, result任务处理完成结果已写入 examples/, ) time.sleep(5) if __name__ __main__: run_agent()运行代理端python agent_client.py代理端跑起来后打开看板接口页面应该能看到任务状态从pending变成running最后变成completed。整个链路验证通过后你只需要把run_agent里的模拟逻辑替换成真实的 AI 代理调用即可。7. 接口 API 与批量任务队列看板的价值在批量任务场景下会被放大。一次提交几十个任务代理端按队列消费人在看板上监控。以下是常用操作示例。创建单个任务curl -X POST http://127.0.0.1:8000/api/tasks \ -H Content-Type: application/json \ -d { title: 整理本周用户反馈, description: 从数据库导出本周反馈按主题归类, agent_id: analyst-agent, priority: 2 }代理上报需要人工确认curl -X PATCH http://127.0.0.1:8000/api/tasks/{task_id} \ -H Content-Type: application/json \ -d {status: waiting_human, progress: 80}人工确认后放行curl -X PATCH http://127.0.0.1:8000/api/tasks/{task_id} \ -H Content-Type: application/json \ -d {status: running, remark: 已确认继续}批量提交任务时建议用 Python 脚本循环创建任务import requests BASE_URL http://127.0.0.1:8000 tasks [ {title: 竞品分析-产品A, description: 分析产品A最近一次更新, agent_id: analyst-agent, priority: 1}, {title: 竞品分析-产品B, description: 分析产品B最近一次更新, agent_id: analyst-agent, priority: 1}, {title: 周报草稿-市场组, description: 根据市场组数据生成周报草稿, agent_id: write-agent, priority: 2}, ] for item in tasks: resp requests.post(f{BASE_URL}/api/tasks, jsonitem, timeout10) print(resp.json())批量任务的关键不是一次性提交而是控制并发和失败重试。你可以在代理端做并发限制比如同时最多执行 2 个任务剩下的保持pending。也可以在配置文件里统一管理队列参数queue: batch_size: 10 concurrency: 2 retry_count: 3 retry_delay_seconds: 60 task_timeout_seconds: 1800task_timeout_seconds是中断机制的重要参数。当任务执行时间超过阈值看板或监控脚本可以把状态从running改回pending或者直接标记为failed这样代理进程崩溃后任务不会永远卡住。这里要提醒一个点不要设计无限重试。失败超过retry_count的任务应该进入failed状态留给人工处理。批量任务中最常见的灾难就是重试风暴几十个失败任务同时重试把下游接口打挂。8. 资源占用与性能观察这套看板本身非常轻主要消耗来自运行中的 AI 代理。单看 FastAPI 服务和 SQLite几百条任务完全无压力CPU 和内存占用几乎可以忽略。真正需要观察的是代理进程本身。建议观察三个指标任务队列长度pending任务堆积数持续增长说明代理处理不过来。平均任务耗时从running到completed的时间异常增长说明下游接口变慢。失败率failed任务占总数比例大于一定阈值就要人工介入。写一个简单的监控脚本import requests resp requests.get(http://127.0.0.1:8000/api/tasks, timeout10) tasks resp.json() pending [t for t in tasks if t[status] pending] running [t for t in tasks if t[status] running] waiting [t for t in tasks if t[status] waiting_human] failed [t for t in tasks if t[status] failed] print(fpending: {len(pending)}) print(frunning: {len(running)}) print(fwaiting_human: {len(waiting)}) print(ffailed: {len(failed)})从效率优化角度看代理端应该对相同类型的任务做上下文复用避免每处理一条任务就重新加载一遍模型或重新拉取背景资料。合理使用waiting_human可以显著减少无效输出让代理在关键决策点上先停下来问人而不是一口气生成一堆错误结果。9. 常见问题与排查方法问题现象可能原因排查方式解决方案服务启动后接口 404路由未注册或 uvicorn 启动模块路径不对检查 app.py 中的路由装饰器和启动命令使用uvicorn app:app并确认在项目根目录执行代理无法拉取任务没有pending任务或agent_id不匹配调用/api/tasks查看现有数据检查参数创建任务时指定正确的agent_id任务卡在running代理进程被杀、异常退出没有上报状态查看代理日志观察任务最后更新时间增加心跳上报超时自动回滚到pending端口被占用上次服务未正常退出或其他进程占用 8000本机检查端口占用情况换端口启动或结束占用进程API 返回 500数据库路径不存在或数据结构不匹配查看 uvicorn 日志检查data目录确认data目录存在必要时重建数据库批量任务重复执行代理端没有去重逻辑重复拉取同一批pending任务查看任务日志中是否出现重复 task_id代理端增加独占标记或任务锁领取后立即置为running任务等待审批后无反应人工端没有处理waiting_human任务用列表接口筛选waiting_human状态任务增加人工审批页面或告警通知排查思路按照“先从数据看再从代码看”的顺序走。先查/api/tasks接口返回的数据是否在预期状态再去翻代理端和服务端日志。大多数问题出在状态上报逻辑而不是看板本身。10. 最佳实践与使用建议最后给一些工程化建议直接照做可以少踩很多坑。第一第一次验证时用最小参数集跑通全链路。创建 3 个非常简单的任务比如“输出一句话”确认代理能领取、能上报、能完成再逐步加入批量、审批、中断。第二保留一套最小可运行配置。把app.py、虚拟环境依赖、启动命令、配置文件固定下来后续任何改动都在副本上做。这样即使改崩了也能快速回到可用状态。第三目录结构要保持干净。模型文件、输入素材、输出结果、任务日志分开管理不要让代理把结果到处乱写。第四批量任务必须加日志和失败重试。任务每进入一个新状态就记录一行日志这样出问题时能定位到具体任务和步骤。重试次数要有限制次数到顶就进入failed绝对不要进入无限重试。第五接口服务要控制访问范围。如果看板服务暴露在局域网或公网一定要加认证。任务看板里有真实业务描述和中间结果默认只绑定127.0.0.1有条件就加 token 或简单的 API Key。第六涉及人脸、声音、版权素材、个人隐私数据的任务必须确认授权。这一点在代理任务场景里尤其重要因为代理可能会自主执行后续步骤人类介入的节点比纯人工流程更少。第七发布或商用前要做效果复核。代理自动生成的内容不能直接裸奔到生产环境至少要有一个人工抽检步骤甚至可以在看板里增加“复核通过/不通过”字段。11. 总结与下一步Human task board for my agents 这套方案最值得尝试的一点就是把“人机协同”从口头概念变成了实际可跑的任务流。你不需要一上来就搞复杂的分布式调度用 FastAPI 加 SQLite 就能支撑一个个人级的代理任务中心。最先建议验证的是三件事能不能创建任务、代理能不能稳定拉取并更新状态、人工审批节点能不能把任务从waiting_human放行回running。这三步通了说明看板主链路已经跑通。最容易踩的坑有两个一个是任务卡死在running状态解决思路是超时自动回滚另一个是多代理争抢同一个任务解决思路是领取后立即置为running避免重复消费。后续可以继续扩展的方向包括加入前端看板页面、接入消息通知、把任务日志导出成审计报表、给不同代理配置独立的并发额度以及引入向量数据库对历史任务做相似度去重。每加一层你的代理协作链条就会再稳一点。