最近在给一组 AI Agent 搭建执行流水线时我遇到了一个非常现实的问题Agent 拿到任务后就埋头执行但是它脑子里的任务清单和我想的完全对不上。多个 Agent 并行工作时有的重复处理同一件事有的在需要人类确认的节点上自己猜答案产出的结果我看不到也管不住。后来我换了个思路把“任务”从 Agent 的私有上下文里抽出来放到一个人类和 Agent 都能读、都能写、都能确认状态的任务看板上让任务本身成为协作的唯一事实来源。这篇文章就把这套方案的完整设计、代码实现和运行验证过程分享出来。这套方案适合正在做 Agent 应用、希望让 Agent 更可控的开发者也适合第一次接触 human-in-the-loop人在回路设计的新手。读完你能掌握任务看板的核心数据模型是什么、Agent 如何通过 HTTP API 认领和更新任务、如何设计“Agent 完成 → 人类审批 → 任务归档”的完整闭环。文章不依赖某个特定 Agent 框架提供的是一个通用、可扩展的后端看板并给出 Agent 调用侧的 Python 示例你可以直接拿到自己的项目里修改使用。1. 为什么需要 Human Task Board for Agents1.1 从“Agent 自己记清单”到“人和 Agent 共享看板”很多 Agent 应用的第一版都是把任务直接写死在 Prompt 里或者在 Agent 内部维护一份私有待办列表。这样做在小实验里很顺手但一旦任务变多、Agent 变多问题就会集中爆发。首先是信息孤岛。Agent A 和 Agent B 都以为自己拿到了“最高优先级任务”实际是在做同一件事你说“把日志归档”Agent 把日志删了一半你根本不知道它做了什么。其次是缺乏确认机制。Agent 遇到边界情况时如果没法把问题转交给人类它要么瞎猜要么卡死。第三是审计困难。任务是谁创建的、谁处理的、什么时候进入完成状态的、有没有经过人工批准这些信息全部丢失。Human Task Board 的思路是把任务当作一份可查询、可变更、可追溯的公共数据。人类负责创建任务和审批结果Agent 负责认领任务和上报进度。这个看板不参与 Agent 的具体推理只负责回答三个问题现在有什么任务要做这个任务是谁在处理这个任务需要人类确认吗1.2 Agent 中断Interrupt与人类确认机制最近 Agent 工程领域非常关注 interrupt中断机制。所谓 interrupt就是当 Agent 执行到不确定性较高的节点时主动把控制权交还给人类等待人类的确认、补充或修正再继续执行。这个机制能显著减少 Agent 的无效推理和错误动作。而一个共享的任务看板正是承载 interrupt 的天然载体。当 Agent 发现任务需要人类确认时它不需要在会话里支支吾吾只需要把看板上的任务状态从in_progress改成waiting_approval就等于发出了一次结构化的中断请求。人类打开看板就能看到“这条任务卡在等待审批”审批通过后状态变更为done审批驳回则变为blockedAgent 可以据此决定继续执行还是调整方案。这种设计比用聊天消息传递中断信号更可靠。聊天记录容易丢看板状态不会丢聊天记录只能靠人读看板状态可以被程序解析。对于长时间运行的 Agent 任务来说看板就是 Agent 和人类之间的“通信协议”。1.3 设计目标让 Agent 更高效今天的 Agent 应用经常面临两个极端一是完全放手让 Agent 自由发挥结果不可控二是每个动作都要人类确认效率极低。任务看板的设计目标是在两者之间找到平衡让 Agent 更高效也就是类似 toward efficient agents 这个方向的实践思路。具体来说看板通过优先级和状态让 Agent 只处理当前最重要的任务减少在模糊目标上的反复试探通过requires_human_approval字段让 Agent 提前知道哪些任务需要人工把关而不是执行完再回头补救通过状态机让“人类审批”成为一个明确节点而不是隐藏在 Agent 的推理过程里。所以这个项目表面上是一个待办清单本质上是一层“人机协作控制面”。它把人的判断力和 Agent 的执行力组合起来让人的精力只花在真正需要判断的地方。2. 系统设计2.1 功能需求分析在动手写代码前先把需求拆清楚。这个任务看板需要满足以下能力人类可以创建任务填写标题、描述、优先级并标记该任务完成后是否需要人工审批。Agent 可以获取待处理任务列表认领某个任务并把任务状态修改为处理中。Agent 执行完任务后如果任务要求人工审批Agent 就把状态改为等待审批如果不要求Agent 可以直接把任务标记为完成。人类可以查看全部任务并对等待审批的任务进行批准或驳回。任务的全生命周期需要保留可查询的状态包括创建时间、更新时间、处理 Agent 等。为了聚焦核心逻辑第一版不做用户登录、不做多人权限、不做前端页面。使用方只有两个角色人类和 Agent。两者通过不同的 API Key 区分。2.2 任务状态机设计任务状态不能随意乱填否则 Agent 和人类之间的协作会失控。我把状态定义为六个状态含义可由谁进入todo待处理还没有 Agent 认领人类创建任务时in_progress已被某个 Agent 认领正在处理Agent 认领任务waiting_approvalAgent 已完成等待人类审批Agent 请求审批done已完成或被人类批准通过Agent 直接完成或人类批准blocked被人类驳回或执行过程中遇到阻塞人类驳回或 Agent 标记状态流转的触发动作如下当前状态触发动作执行方目标状态todo认领任务Agentin_progressin_progress执行完成且无需审批Agentdonein_progress请求人类审批Agentwaiting_approvalwaiting_approval人类批准人类donewaiting_approval人类驳回人类blockedblocked重新处理Agentin_progress这里有一个细节值得注意blocked并不意味着任务失败而是“需要人类介入或需要调整方案”。Agent 后续可以根据human_notes里的驳回原因修改策略再重新进入处理中。这样设计能避免 Agent 因为一次审批不通过就彻底放弃任务。2.3 API 设计后端采用 REST API端点划分如下方法路径调用方说明POST/tasks人类创建任务GET/tasks人类 / Agent查看任务列表支持按状态筛选GET/tasks/{task_id}人类 / Agent查看任务详情PATCH/tasks/{task_id}Agent认领任务或更新状态POST/tasks/{task_id}/request-approvalAgent请求人类审批POST/tasks/{task_id}/approve人类批准或驳回任务GET/health任意健康检查权限上人类和 Agent 使用不同的 API Key人类 Key 拥有创建任务、审批任务的权限Agent Key 拥有查看任务、认领任务、更新状态和请求审批的权限。这样即使 Key 泄露影响范围也是有限的符合最小权限原则。2.4 数据模型设计任务表tasks的字段设计如下id主键自增。title任务标题必填长度不超过 200。description任务描述补充说明细节。priority优先级用 P0/P1/P2/P3 表示默认 P2。status当前状态对应上面的状态机。agent_id正在处理该任务的 Agent 标识例如>fastapi uvicorn sqlalchemy pydantic httpx然后执行安装命令pip install -r requirements.txt如果你使用 venv 虚拟环境先创建并激活环境再安装。这一步很基础但能避免不同项目之间的依赖版本冲突。4. 完整代码实现4.1 数据库连接配置文件路径app/database.py# app/database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base SQLALCHEMY_DATABASE_URL sqlite:///./taskboard.db engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base()这里解释两个关键点。第一check_same_threadFalse是 SQLite 配合 FastAPI 常见写法。FastAPI 的请求可能在线程池中执行SQLite 默认只允许创建连接的线程访问数据库关闭这个检查可以避免跨线程访问报错。第二Base是所有 ORM 模型的父类。后续定义 Task 模型时直接继承这个BaseSQLAlchemy 就能通过它统一管理表结构。另外我建议在数据库连接上开启 SQLite 的 WAL 模式。方法是在 engine 创建后通过事件监听器执行 PRAGMA# app/database.py追加以下内容 from sqlalchemy import event event.listens_for(engine, connect) def set_sqlite_pragma(dbapi_connection, connection_record): cursor dbapi_connection.cursor() cursor.execute(PRAGMA journal_modeWAL) cursor.close()WAL 模式允许读操作和写操作并发执行在线程池环境下能显著减少 “database is locked” 错误。4.2 ORM 模型定义文件路径app/models.py# app/models.py from datetime import datetime, timezone from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime from .database import Base def utcnow(): 返回 UTC 时间去掉 tzinfo便于写入 SQLite。 return datetime.now(timezone.utc).replace(tzinfoNone) class Task(Base): __tablename__ tasks id Column(Integer, primary_keyTrue, indexTrue) title Column(String(200), nullableFalse) description Column(Text, default) priority Column(String(20), defaultP2) status Column(String(20), defaulttodo, indexTrue) agent_id Column(String(100), nullableTrue) requires_human_approval Column(Boolean, defaultFalse) approval_requested Human Task Board：构建人类与Agent共享的任务看板
最近在给一组 AI Agent 搭建执行流水线时我遇到了一个非常现实的问题Agent 拿到任务后就埋头执行但是它脑子里的任务清单和我想的完全对不上。多个 Agent 并行工作时有的重复处理同一件事有的在需要人类确认的节点上自己猜答案产出的结果我看不到也管不住。后来我换了个思路把“任务”从 Agent 的私有上下文里抽出来放到一个人类和 Agent 都能读、都能写、都能确认状态的任务看板上让任务本身成为协作的唯一事实来源。这篇文章就把这套方案的完整设计、代码实现和运行验证过程分享出来。这套方案适合正在做 Agent 应用、希望让 Agent 更可控的开发者也适合第一次接触 human-in-the-loop人在回路设计的新手。读完你能掌握任务看板的核心数据模型是什么、Agent 如何通过 HTTP API 认领和更新任务、如何设计“Agent 完成 → 人类审批 → 任务归档”的完整闭环。文章不依赖某个特定 Agent 框架提供的是一个通用、可扩展的后端看板并给出 Agent 调用侧的 Python 示例你可以直接拿到自己的项目里修改使用。1. 为什么需要 Human Task Board for Agents1.1 从“Agent 自己记清单”到“人和 Agent 共享看板”很多 Agent 应用的第一版都是把任务直接写死在 Prompt 里或者在 Agent 内部维护一份私有待办列表。这样做在小实验里很顺手但一旦任务变多、Agent 变多问题就会集中爆发。首先是信息孤岛。Agent A 和 Agent B 都以为自己拿到了“最高优先级任务”实际是在做同一件事你说“把日志归档”Agent 把日志删了一半你根本不知道它做了什么。其次是缺乏确认机制。Agent 遇到边界情况时如果没法把问题转交给人类它要么瞎猜要么卡死。第三是审计困难。任务是谁创建的、谁处理的、什么时候进入完成状态的、有没有经过人工批准这些信息全部丢失。Human Task Board 的思路是把任务当作一份可查询、可变更、可追溯的公共数据。人类负责创建任务和审批结果Agent 负责认领任务和上报进度。这个看板不参与 Agent 的具体推理只负责回答三个问题现在有什么任务要做这个任务是谁在处理这个任务需要人类确认吗1.2 Agent 中断Interrupt与人类确认机制最近 Agent 工程领域非常关注 interrupt中断机制。所谓 interrupt就是当 Agent 执行到不确定性较高的节点时主动把控制权交还给人类等待人类的确认、补充或修正再继续执行。这个机制能显著减少 Agent 的无效推理和错误动作。而一个共享的任务看板正是承载 interrupt 的天然载体。当 Agent 发现任务需要人类确认时它不需要在会话里支支吾吾只需要把看板上的任务状态从in_progress改成waiting_approval就等于发出了一次结构化的中断请求。人类打开看板就能看到“这条任务卡在等待审批”审批通过后状态变更为done审批驳回则变为blockedAgent 可以据此决定继续执行还是调整方案。这种设计比用聊天消息传递中断信号更可靠。聊天记录容易丢看板状态不会丢聊天记录只能靠人读看板状态可以被程序解析。对于长时间运行的 Agent 任务来说看板就是 Agent 和人类之间的“通信协议”。1.3 设计目标让 Agent 更高效今天的 Agent 应用经常面临两个极端一是完全放手让 Agent 自由发挥结果不可控二是每个动作都要人类确认效率极低。任务看板的设计目标是在两者之间找到平衡让 Agent 更高效也就是类似 toward efficient agents 这个方向的实践思路。具体来说看板通过优先级和状态让 Agent 只处理当前最重要的任务减少在模糊目标上的反复试探通过requires_human_approval字段让 Agent 提前知道哪些任务需要人工把关而不是执行完再回头补救通过状态机让“人类审批”成为一个明确节点而不是隐藏在 Agent 的推理过程里。所以这个项目表面上是一个待办清单本质上是一层“人机协作控制面”。它把人的判断力和 Agent 的执行力组合起来让人的精力只花在真正需要判断的地方。2. 系统设计2.1 功能需求分析在动手写代码前先把需求拆清楚。这个任务看板需要满足以下能力人类可以创建任务填写标题、描述、优先级并标记该任务完成后是否需要人工审批。Agent 可以获取待处理任务列表认领某个任务并把任务状态修改为处理中。Agent 执行完任务后如果任务要求人工审批Agent 就把状态改为等待审批如果不要求Agent 可以直接把任务标记为完成。人类可以查看全部任务并对等待审批的任务进行批准或驳回。任务的全生命周期需要保留可查询的状态包括创建时间、更新时间、处理 Agent 等。为了聚焦核心逻辑第一版不做用户登录、不做多人权限、不做前端页面。使用方只有两个角色人类和 Agent。两者通过不同的 API Key 区分。2.2 任务状态机设计任务状态不能随意乱填否则 Agent 和人类之间的协作会失控。我把状态定义为六个状态含义可由谁进入todo待处理还没有 Agent 认领人类创建任务时in_progress已被某个 Agent 认领正在处理Agent 认领任务waiting_approvalAgent 已完成等待人类审批Agent 请求审批done已完成或被人类批准通过Agent 直接完成或人类批准blocked被人类驳回或执行过程中遇到阻塞人类驳回或 Agent 标记状态流转的触发动作如下当前状态触发动作执行方目标状态todo认领任务Agentin_progressin_progress执行完成且无需审批Agentdonein_progress请求人类审批Agentwaiting_approvalwaiting_approval人类批准人类donewaiting_approval人类驳回人类blockedblocked重新处理Agentin_progress这里有一个细节值得注意blocked并不意味着任务失败而是“需要人类介入或需要调整方案”。Agent 后续可以根据human_notes里的驳回原因修改策略再重新进入处理中。这样设计能避免 Agent 因为一次审批不通过就彻底放弃任务。2.3 API 设计后端采用 REST API端点划分如下方法路径调用方说明POST/tasks人类创建任务GET/tasks人类 / Agent查看任务列表支持按状态筛选GET/tasks/{task_id}人类 / Agent查看任务详情PATCH/tasks/{task_id}Agent认领任务或更新状态POST/tasks/{task_id}/request-approvalAgent请求人类审批POST/tasks/{task_id}/approve人类批准或驳回任务GET/health任意健康检查权限上人类和 Agent 使用不同的 API Key人类 Key 拥有创建任务、审批任务的权限Agent Key 拥有查看任务、认领任务、更新状态和请求审批的权限。这样即使 Key 泄露影响范围也是有限的符合最小权限原则。2.4 数据模型设计任务表tasks的字段设计如下id主键自增。title任务标题必填长度不超过 200。description任务描述补充说明细节。priority优先级用 P0/P1/P2/P3 表示默认 P2。status当前状态对应上面的状态机。agent_id正在处理该任务的 Agent 标识例如>fastapi uvicorn sqlalchemy pydantic httpx然后执行安装命令pip install -r requirements.txt如果你使用 venv 虚拟环境先创建并激活环境再安装。这一步很基础但能避免不同项目之间的依赖版本冲突。4. 完整代码实现4.1 数据库连接配置文件路径app/database.py# app/database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base SQLALCHEMY_DATABASE_URL sqlite:///./taskboard.db engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base()这里解释两个关键点。第一check_same_threadFalse是 SQLite 配合 FastAPI 常见写法。FastAPI 的请求可能在线程池中执行SQLite 默认只允许创建连接的线程访问数据库关闭这个检查可以避免跨线程访问报错。第二Base是所有 ORM 模型的父类。后续定义 Task 模型时直接继承这个BaseSQLAlchemy 就能通过它统一管理表结构。另外我建议在数据库连接上开启 SQLite 的 WAL 模式。方法是在 engine 创建后通过事件监听器执行 PRAGMA# app/database.py追加以下内容 from sqlalchemy import event event.listens_for(engine, connect) def set_sqlite_pragma(dbapi_connection, connection_record): cursor dbapi_connection.cursor() cursor.execute(PRAGMA journal_modeWAL) cursor.close()WAL 模式允许读操作和写操作并发执行在线程池环境下能显著减少 “database is locked” 错误。4.2 ORM 模型定义文件路径app/models.py# app/models.py from datetime import datetime, timezone from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime from .database import Base def utcnow(): 返回 UTC 时间去掉 tzinfo便于写入 SQLite。 return datetime.now(timezone.utc).replace(tzinfoNone) class Task(Base): __tablename__ tasks id Column(Integer, primary_keyTrue, indexTrue) title Column(String(200), nullableFalse) description Column(Text, default) priority Column(String(20), defaultP2) status Column(String(20), defaulttodo, indexTrue) agent_id Column(String(100), nullableTrue) requires_human_approval Column(Boolean, defaultFalse) approval_requested