尧图网页设计 
最近在梳理 Agent 工程化落地时我反复被一个问题卡住Agent 可以自动执行任务但当它需要做决策、改配置、删数据或者对外回复时真的应该让它独立完成吗答案是在大多数生产场景里不应该。如果团队里同时跑着三四个 Agent有的在写代码有的在更新需求文档有的在处理客户工单你怎么知道它们现在在干什么哪些任务需要你确认哪些任务已经卡住超过一小时了这时候缺的不是更强的模型而是一块“人工任务看板”。这篇文章我会从工程角度拆解 Human Task Board 的设计与实现包括任务状态机、Agent 侧接入、人工审核流程、Agent 中断与恢复机制并给出完整的 FastAPI SQLite 示例代码。读完你应该能照着搭出一个最小可用的“人审台”把它接到自己的 Agent 工作流里。1. 为什么 Agent 需要一块人工任务看板如果只看 DemoAgent 似乎无所不能它能规划、能调用工具、能写完一段代码。但一旦进入真实业务问题就变了——Agent 的每一步行动是否被授权、是否可回滚、是否有人复核这些问题 Demo 里根本不会出现。举一个实际场景。一个自动化运维 Agent 检测到磁盘使用率超过 85%按照预设策略应该清理日志目录。它执行了rm -rf /var/log/old。假设路径拼接有误或者清理范围超出了预期这个操作会带来什么后果如果旁边有一块任务看板Agent 在执行前先生成一个“高危操作审批”任务等人工确认后再执行风险边界就完全不一样了。再比如写代码场景。Agent 改动了一个核心模块的接口签名关联的调用方可能有十几个。它自称“已经全量修改完毕”但你敢直接合入主干吗更稳妥的做法是Agent 完成改动后在看板上生成一个“请人工检查代码改动”的任务开发者 review 通过后再进入下一步。所以Human Task Board 本质上解决的不是“Agent 能不能干活”而是“Agent 干了活之后谁负责、谁确认、谁兜底”。它是人机协作的一个中间层让 Agent 的自主性被约束在可控范围内。从材料中的热词也能看出这个趋势。toward efficient agents强调的是 Agent 的效率但效率不等于完全自主deep agents interrupt聊的则是 Agent 运行过程中的中断与插话机制。这两个方向合在一起指向的正是 Human-in-the-Loop人在回路的工程化效率靠 Agent安全靠任务看板 中断机制。一句话判断如果你的 Agent 只是本地玩具任务看板可有可无但只要 Agent 要接触真实数据、真实系统、真实用户人工任务看板就不是加分项而是必选项。2. Human Task Board 的核心概念与适用场景2.1 什么是 Human Task BoardHuman Task Board直译是“人工任务看板”。它不是一个特定产品而是一套机制为 Agent 产生的、需要人类参与的任务提供统一的登记、展示、审核、追踪能力。它和普通任务管理软件的区别在于普通任务管理软件的服务对象是人类而 Human Task Board 的服务对象是 Agent Human 的混合流程。任务可能由 Agent 创建由 Agent 推进只在关键节点停下来等人类确认然后再由 Agent 继续执行。2.2 五个核心对象一个最小可用的任务看板需要管理以下对象对象说明示例Task任务一次需要人工参与的独立工作单元“请审核高危操作清理日志目录”State状态任务当前所处阶段pending、await_human、approvedArtifact产物Agent 执行过程中产生的内容代码 diff、日志、报告、参数快照Comment评论人类或 Agent 对任务的补充说明“请补充影响范围再审核”Audit Log审计日志任务全生命周期的操作记录谁在什么时间批准了什么后面实现时我们会把 Artifact 和 Comment 作为任务的关联数据存储而不是单独建模这样对小型项目更省事。2.3 任务状态机从创建到关闭任务状态设计是整个看板最重要的部分。状态少了表达不清楚状态多了维护成本高。一组经过实践检验的状态定义如下pending 任务已创建等待调度或 Agent 开始处理 in_progress 任务已被 Agent 领取并正在执行 await_human 任务执行到关键节点等待人工审核或提供信息 approved 人工审核通过Agent 可以继续执行 rejected 人工驳回Agent 需要修改后重新提交或终止 cancelled 任务被取消流程终止 completed 任务已全部完成并关闭 failed 任务执行失败需要人工介入分析转换规则需要严格控制例如await_human只能变为approved、rejected、cancelled。approved之后 Agent 继续执行最终变为completed或failed。rejected之后 Agent 可以选择修改并重新提交为await_human也可以放弃任务。这样的状态机设计有一个直接好处任何时刻你都能回答“现在有多少任务在等人工”“哪些 Agent 正在执行什么”这两个问题。2.4 什么场景下最需要它高危操作审批删除数据、修改权限、发布上线、转账支付。代码变更审查Agent 生成代码 diff人工 review 后再合入。内容发布审核Agent 生成营销文案、客服回复人工确认后对外发送。多 Agent 协作协调多个 Agent 产出结果有依赖需要人工裁决优先级。数据标注与质检Agent 预标注人工抽检或修正。这些场景的共同点是操作不可逆或者对外有影响或者错误成本高。遇到这类场景不要赌 Agent 的“聪明”要依赖流程的“稳”。3. 整体架构设计看板放在 Agent 和系统之间在动手写代码前先明确架构位置。一个典型的 Agent 应用架构如下User / Human Reviewer ↑ | 审核、批准、驳回 ----------------------------- Human Task Board (看板) ----------------------------- ↑ 创建任务、更新状态、请求审批 ----------------------------- Agent Orchestrator (编排器) | | Agent A Agent B | | Tools Tools (代码库) (数据库) (外部API)看板层是 Agent 与外部世界之间的“闸门”。Agent 不直接执行高危操作而是先在看板上登记意图等人批准后再真正执行。这个模式有时被称为Human-in-the-Loop Gateway。这里有一个关键设计决策看板是同步阻塞还是异步通知。同步阻塞Agent 发起审批请求后挂起直到人工响应才继续。适合高危操作但 Agent 会一直占用资源。异步通知Agent 发起请求后去做别的任务人工响应后通过事件机制唤醒。适合常规审批但实现复杂度更高。对于第一版建议先做同步阻塞。原因很简单同步模型容易理解、容易排错而且高危操作的频率通常不高不会成为性能瓶颈。后面再升级为异步事件驱动。4. 环境准备与项目结构本文示例使用以下技术栈。版本以你本机实际安装为准思路和代码模式是通用的Python 3.9FastAPISQLAlchemy或直接使用 sqlite3本文为了减少依赖使用 sqlite3 标准库 Pydantic 做数据校验uvicornrequests 或 httpxAgent 侧调用看板 API 用如果你的 Agent 本身用 Python直接复用先创建项目目录mkdir human-task-board cd human-task-board mkdir task_board touch task_board/__init__.py最终项目结构如下human-task-board/ ├── task_board/ │ ├── __init__.py │ ├── models.py # 数据模型与状态定义 │ ├── storage.py # SQLite 存储层 │ ├── api.py # FastAPI 接口 │ └── cli_review.py # 命令行审核工具 ├── agent_side.py # 模拟 Agent 侧接入示例 ├── requirements.txt └── README.md依赖文件fastapi uvicorn pydantic httpx安装命令pip install -r requirements.txt5. 核心实现任务模型、存储层与 API5.1 任务数据模型先定义状态枚举与任务模型。这里用 Pydantic 做请求参数校验用标准库 sqlite3 做持久化尽量保持代码可读性。# 文件路径task_board/models.py from enum import Enum from typing import Optional, List from pydantic import BaseModel, Field from datetime import datetime class TaskState(str, Enum): PENDING pending IN_PROGRESS in_progress AWAIT_HUMAN await_human APPROVED approved REJECTED rejected CANCELLED cancelled COMPLETED completed FAILED failed class TaskCreate(BaseModel): title: str Field(..., min_length1, max_length200, description任务标题) description: str Field(, description任务详细说明) agent_name: str Field(..., description发起任务的 Agent 名称) task_type: str Field(general, description任务类型如 high_risk, code_review, content_publish) payload: dict Field(default_factorydict, description任务附带的数据如命令、参数、文件路径) priority: int Field(1, ge0, le10, description优先级数值越大越紧急) class TaskUpdate(BaseModel): state: Optional[TaskState] None comment: Optional[str] None reviewer: Optional[str] None class Task(TaskCreate): id: str state: TaskState created_at: str updated_at: str reviewer: Optional[str] None review_comment: Optional[str] None设计说明payload是字典类型用来携带 Agent 执行前后的上下文。比如高危操作场景可以放{command: rm -rf /var/log/old, target_host: 192.168.1.10}。task_type用于前端做不同渲染也方便统计不同类别任务的审核耗时。reviewer记录谁做了审核是审计的基础字段。5.2 SQLite 存储层存储层不需要复杂重点是保证读写字段完整、状态转换可追踪。# 文件路径task_board/storage.py import sqlite3 import json import uuid from datetime import datetime from .models import Task, TaskCreate, TaskState, TaskUpdate DB_PATH task_board.db def get_conn(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_conn() conn.execute( CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, title TEXT NOT NULL, description TEXT, agent_name TEXT NOT NULL, task_type TEXT, payload TEXT, priority INTEGER DEFAULT 1, state TEXT NOT NULL, reviewer TEXT, review_comment TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) ) conn.commit() conn.close() def now_str() - str: return datetime.now().isoformat(timespecseconds) def create_task(data: TaskCreate) - Task: task_id uuid.uuid4().hex[:12] created now_str() conn get_conn() conn.execute( INSERT INTO tasks (id, title, description, agent_name, task_type, payload, priority, state, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( task_id, data.title, data.description, data.agent_name, data.task_type, json.dumps(data.payload, ensure_asciiFalse), data.priority, TaskState.PENDING.value, created, created, ), ) conn.commit() conn.close() return get_task(task_id) def get_task(task_id: str) - Task | None: conn get_conn() row conn.execute(SELECT * FROM tasks WHERE id ?, (task_id,)).fetchone() conn.close() if row is None: return None return row_to_task(row) def list_tasks(state: str | None None, limit: int 50) - list[Task]: conn get_conn() if state: rows conn.execute( SELECT * FROM tasks WHERE state ? ORDER BY priority DESC, created_at DESC LIMIT ?, (state, limit), ).fetchall() else: rows conn.execute( SELECT * FROM tasks ORDER BY priority DESC, created_at DESC LIMIT ?, (limit,), ).fetchall() conn.close() return [row_to_task(row) for row in rows] def update_task(task_id: str, update: TaskUpdate) - Task | None: task get_task(task_id) if task is None: return None new_state update.state.value if update.state else task.state reviewer update.reviewer if update.reviewer else task.reviewer review_comment update.review_comment if update.review_comment else task.review_comment conn get_conn() conn.execute( UPDATE tasks SET state ?, reviewer ?, review_comment ?, updated_at ? WHERE id ? , (new_state, reviewer, review_comment, now_str(), task_id), ) conn.commit() conn.close() return get_task(task_id) def row_to_task(row: sqlite3.Row) - Task: return Task( idrow[id], titlerow[title], descriptionrow[description], agent_namerow[agent_name], task_typerow[task_type], payloadjson.loads(row[payload]), priorityrow[priority], stateTaskState(row[state]), reviewerrow[reviewer], review_commentrow[review_comment], created_atrow[created_at], updated_atrow[updated_at], )这个存储层用 JSON 序列化payload字段好处是不需要为不同任务类型建不同表坏处是不能直接用 SQL 对 payload 里的字段做复杂查询。对于任务看板这种低频场景完全够用。5.3 FastAPI 接口接口层提供四个核心能力创建任务Agent 调用。查询任务列表看板前端或 CLI 调用。查询单个任务详情页使用。更新任务状态人工审核使用。# 文件路径task_board/api.py from fastapi import FastAPI, HTTPException, Query from .models import Task, TaskCreate, TaskUpdate from . import storage app FastAPI(titleHuman Task Board API) app.on_event(startup) def startup(): storage.init_db() app.post(/tasks, response_modelTask, status_code201) def create_task(data: TaskCreate): Agent 创建一个人工审核任务 task storage.create_task(data) return task app.get(/tasks, response_modellist[Task]) def get_tasks(state: str | None Query(defaultNone), limit: int Query(default50, le200)): 查询任务列表可按状态过滤 if state and state not in [s.value for s in storage.TaskState]: raise HTTPException(status_code400, detailf非法状态: {state}) return storage.list_tasks(statestate, limitlimit) app.get(/tasks/{task_id}, response_modelTask) def get_task(task_id: str): task storage.get_task(task_id) if task is None: raise HTTPException(status_code404, detail任务不存在) return task app.post(/tasks/{task_id}/review, response_modelTask) def review_task(task_id: str, update: TaskUpdate): 人工审核接口批准、驳回或取消任务 if not update.state: raise HTTPException(status_code400, detail请指定目标状态) if update.state.value not in [approved, rejected, cancelled]: raise HTTPException(status_code400, detail审核操作只能为 approved / rejected / cancelled) task storage.update_task(task_id, update) if task is None: raise HTTPException(status_code404, detail任务不存在) return task这里把审核接口单独拆成/tasks/{task_id}/review而不是用通用更新接口是为了在语义上明确“这是人工审核动作”。后续如果要在审核时触发通知、写审计日志都可以在这个接口里扩展。启动服务uvicorn task_board.api:app --host 0.0.0.0 --port 8000 --reload启动后FastAPI 自带的文档地址是http://127.0.0.1:8000/docs可以直接在浏览器里手动创建和审核任务这对快速联调很有用。5.4 Agent 侧接入示例现在模拟一个 Agent 在遇到高危操作时先暂停执行并请求人工审批的接入方式。# 文件路径agent_side.py 模拟 Agent 使用 Human Task Board 的接入示例。 流程 1. Agent 分析问题形成执行计划。 2. 发现存在高危操作创建审批任务。 3. 轮询任务状态等待人工审核。 4. 审核通过则执行真实操作否则放弃操作并记录原因。 import time import httpx API_BASE http://127.0.0.1:8000 POLL_INTERVAL 3 # 轮询间隔单位秒 MAX_WAIT 120 # 最大等待时间单位秒 def request_human_approval(title: str, description: str, payload: dict) - str: 创建人工审批任务返回 task_id resp httpx.post( f{API_BASE}/tasks, json{ title: title, description: description, agent_name: ops-agent, task_type: high_risk, payload: payload, priority: 5, }, timeout10, ) resp.raise_for_status() return resp.json()[id] def wait_for_review(task_id: str) - dict: 阻塞等待人工审核结果 start time.time() while time.time() - start MAX_WAIT: resp httpx.get(f{API_BASE}/tasks/{task_id}, timeout10) resp.raise_for_status() task resp.json() if task[state] in (approved, rejected, cancelled): return task print(f[Agent] 等待审核中... 当前状态: {task[state]}) time.sleep(POLL_INTERVAL) raise TimeoutError(等待人工审核超时) def main(): payload { command: rm -rf /tmp/app_old_logs, target: prod-web-01, reason: 磁盘使用率超过 85%需要清理历史日志, } print([Agent] 检测到磁盘空间不足准备执行清理操作。) print([Agent] 该操作属于高危操作先请求人工审批。) task_id request_human_approval( title高危操作审批清理 prod-web-01 历史日志, descriptionAgent 检测到磁盘使用率超过 85%建议执行日志清理。, payloadpayload, ) print(f[Agent] 已创建审批任务: {task_id}) try: review_result wait_for_review(task_id) except TimeoutError as e: print(f[Agent] 审核超时放弃执行。{e}) return if review_result[state] approved: print([Agent] 人工审核通过开始执行清理操作。) # 在这里执行真实操作例如 subprocess.run(payload[command], shellTrue) print(f[Agent] 执行命令: {payload[command]}) # 执行完成后更新任务为 completed httpx.post( f{API_BASE}/tasks/{task_id}/review, json{state: completed, reviewer: ops-agent, review_comment: 清理任务已执行完成}, timeout10, ) elif review_result[state] rejected: print([Agent] 人工驳回了该操作不执行清理。) else: print([Agent] 任务被取消不执行清理。) if __name__ __main__: main()这段代码展示了最关键的接入模式先请求、再等待、通过后才执行。注意几个细节任务创建后Agent 进入轮询等待。生产环境中Agent 不一定真的 sleep 阻塞可以用回调或事件机制但轮询模式最容易理解和排错。审核通过后Agent 执行真实操作完成后把任务更新为completed形成闭环。如果审核超时Agent 必须有一个兜底策略。这里简单抛出异常生产环境可以改为“放弃执行并通知值班人员”。5.5 命令行人工审核工具人工侧需要一个简单的审核工具方便不打开浏览器时快速处理任务。# 文件路径task_board/cli_review.py 命令行审核工具 用法 python -m task_board.cli_review list python -m task_board.cli_review list --state await_human python -m task_board.cli_review approve task_id python -m task_board.cli_review reject task_id --comment 理由 import argparse import httpx API_BASE http://127.0.0.1:8000 def cmd_list(args): params {limit: args.limit} if args.state: params[state] args.state resp httpx.get(f{API_BASE}/tasks, paramsparams, timeout10) tasks resp.json() if not tasks: print(当前没有任务。) return for t in tasks: print(f[{t[id]}] {t[state]:12s} {t[priority]} {t[title]} agent{t[agent_name]}) def cmd_approve(args): resp httpx.post( f{API_BASE}/tasks/{args.task_id}/review, json{state: approved, reviewer: args.reviewer, review_comment: args.comment}, timeout10, ) if resp.status_code 404: print(任务不存在。) return resp.raise_for_status() print(f任务 {args.task_id} 已通过审核。) def cmd_reject(args): resp httpx.post( f{API_BASE}/tasks/{args.task_id}/review, json{state: rejected, reviewer: args.reviewer, review_comment: args.comment}, timeout10, ) if resp.status_code 404: print(任务不存在。) return resp.raise_for_status() print(f任务 {args.task_id} 已驳回。) def main(): parser argparse.ArgumentParser(descriptionHuman Task Board CLI) subparsers parser.add_subparsers(destcommand) list_parser subparsers.add_parser(list, help列出任务) list_parser.add_argument(--state, defaultNone, help按状态过滤) list_parser.add_argument(--limit, typeint, default20, help最大数量) list_parser.set_defaults(funccmd_list) approve_parser subparsers.add_parser(approve, help批准任务) approve_parser.add_argument(task_id, help任务 ID) approve_parser.add_argument(--reviewer, defaulthuman, help审核人) approve_parser.add_argument(--comment, default, help审核意见) approve_parser.set_defaults(funccmd_approve) reject_parser subparsers.add_parser(reject, help驳回任务) reject_parser.add_argument(task_id, help任务 ID) reject_parser.add_argument(--reviewer, defaulthuman, help审核人) reject_parser.add_argument(--comment, default, help驳回原因) reject_parser.set_defaults(funccmd_reject) args parser.parse_args() if not args.command: parser.print_help() return args.func(args) if __name__ __main__: main()这个 CLI 刻意保持轻量只实现列表、批准、驳回三个操作。实际团队使用中可以继续扩展“取消”“重新打开”“指派给某人”等命令也可以把同样的逻辑封装成 Web 前端。6. 关键机制Agent 中断与恢复前面实现了“Agent 等待人工审核”这个流程但还有一个更深入的问题没有解决当 Agent 被中断时它如何记住自己做到哪一步了恢复后从哪里继续这在 Agent 工程里通常叫interrupt机制。简单说Agent 的执行过程不应该是一段不可分割的“大函数”而应该是一系列可暂停、可恢复的步骤。6.1 中断的本质是持久化执行上下文我第一次实现中断机制时踩过一个坑以为中断就是让 Agent 进程睡眠等审核通过后再唤醒。但实际上Agent 进程可能因为超时被回收容器可能被重启审核人员可能几小时后才处理。所以中断绝不能依赖进程内存。正确做法是在每个可中断步骤前把 Agent 的执行上下文持久化到存储中。上下文至少包含当前任务 ID。当前步骤 ID 或步骤索引。已收集到的输入数据。下一步需要调用什么工具、传入什么参数。依赖人工确认的审批任务 ID。这样即使 Agent 进程重启也能从存储中恢复上下文继续执行。6.2 用“步骤表”代替“流程函数”我推荐的模式是把 Agent 的任务流程拆成显式的步骤表每一步记录状态。# 文件路径agent_ctx.py示意 STEPS [ {id: collect_metrics, action: gather_disk_usage, requires_human: False}, {id: request_cleanup, action: create_human_task, requires_human: True}, {id: exec_cleanup, action: run_clean_command, requires_human: False}, {id: report_result, action: notify_ops, requires_human: False}, ]Agent 主循环伪代码如下def run_agent_flow(task_id: str, ctx: dict): step_index ctx.get(step_index, 0) while step_index len(STEPS): step STEPS[step_index] if step[requires_human]: # 检查对应人工审批任务是否已通过 approval get_approval_by_agent_step(task_id, step[id]) if approval.state pending: # 未处理保存进度并返回等待被唤醒 save_ctx(task_id, {step_index: step_index}) return if approval.state rejected: # 人工驳回停止流程 mark_task_failed(task_id, 人工驳回) return # approved继续向下执行 execute_step(step, ctx) step_index 1 save_ctx(task_id, {step_index: step_index}) mark_task_completed(task_id)这个模式的要点是每一步执行后立即保存进度遇到需要人工确认的步骤就退出主循环外部审核通过后再触发一次 Agent 执行。这样 Agent 的“中断”就不再依赖阻塞等待而是变成了一种带持久化上下文的“挂起-唤醒”模型。6.3 唤醒方式的选择挂起之后如何唤醒常见有三种方式方式实现难度实时性适用场景轮询低有延迟内部工具任务量不大Webhook 回调中高看板与 Agent 同属一个系统消息队列事件较高高大规模多 Agent 集群第一版推荐轮询因为看板本身的数据量很小人工审核频率也不高轮询间隔 3 到 5 秒完全够用。如果后续 Agent 数量上来了再换成 Webhook 或消息队列。7. 运行流程与验证7.1 第一步启动看板服务在一个终端启动 API 服务cd human-task-board uvicorn task_board.api:app --host 0.0.0.0 --port 8000 --reload看到类似输出说明启动成功INFO: Uvicorn running on http://127.0.0.1:8000 INFO: Application startup complete.7.2 第二步创建一个人工审批任务另开一个终端用 curl 创建任务curl -X POST http://127.0.0.1:8000/tasks \ -H Content-Type: application/json \ -d { title: 高危操作审批清理 prod-web-01 历史日志, description: 磁盘使用率超过 85%需要清理日志, agent_name: ops-agent, task_type: high_risk, priority: 5, payload: { command: rm -rf /tmp/app_old_logs, target: prod-web-01 } }预期返回{ id: 3f1a9c2b6d4e, title: 高危操作审批清理 prod-web-01 历史日志, state: pending, agent_name: ops-agent, task_type: high_risk, priority: 5, payload: { command: rm -rf /tmp/app_old_logs, target: prod-web-01 }, created_at: 2025-01-01T10:00:00 }注意id字段后续操作需要用到。7.3 第三步查询待审核任务curl http://127.0.0.1:8000/tasks?stateawait_human这里你会发现刚创建的任务还是pending状态。在实际接入中Agent 领取任务后会把状态改成in_progress到达人工节点时再改成await_human。为了方便测试可以直接手动把状态改成await_human也可以让 Agent 在创建任务时就把初始状态设为await_human。7.4 第四步人工审核使用 CLI 完成审核# 查看任务列表 python -m task_board.cli_review list # 批准任务 python -m task_board.cli_review approve 3f1a9c2b6d4e --reviewer zhangsan --comment 确认清理不影响核心数据预期输出任务 3f1a9c2b6d4e 已通过审核。再查询任务详情状态应变为approvedcurl http://127.0.0.1:8000/tasks/3f1a9c2b6d4e返回中可以看到reviewer和review_comment字段已经被填充。这说明人工审核信息已经完整记录了。7.5 第五步跑通 Agent 接入示例启动服务后另开一个终端运行python agent_side.py此时 Agent 会创建审批任务进入轮询等待。你可以在另一个终端执行批准操作观察 Agent 的输出是否按预期打印“人工审核通过开始执行清理操作”。如果一直不批准Agent 会在MAX_WAIT时间后超时退出并打印“审核超时放弃执行”。这个行为就是我们前面说的兜底策略。7.6 判断成功与否的标准一个完整的闭环应该满足三个条件任务从await_human变为approved且reviewer有值。Agent 收到审核结果后只执行了被批准的操作没有提前执行。执行完成后任务最终变为completed或failed状态有终态。如果任务停留在中间状态超过预期时间优先检查 Agent 的轮询逻辑和 API 是否正常。8. 常见问题与排查思路问题现象可能原因排查方式解决方案创建任务时返回 422请求体缺少必填字段或字段类型错误查看 FastAPI 返回的错误详情检查title、agent_name是否为空payload是否为 JSON 对象Agent 一直显示等待审核但任务已经批准轮询间隔内未抓到最新状态或请求了错误的 API 地址手动 curl 查询任务详情确认API_BASE配置正确缩短轮询间隔检查是否有缓存层审核接口返回 400传入了非法目标状态查看接口文档中的状态枚举审核操作只能传approved、rejected、cancelled之一数据库文件被锁写入失败多个并发请求同时写 SQLite或存在长事务查看 SQLite 报错信息任务看板并发量不高时 SQLite 够用如果并发高换 PostgreSQL 并使用连接池Agent 进程重启后忘记恢复进度没有持久化步骤上下文查看步骤表数据是否存储引入步骤索引持久化每次执行后写入 storage审核人误操作批准了不该批的任务缺少二次确认和历史记录查询任务审计日志增加审核页面的二次确认弹窗记录操作人、时间和 IP便于事后追溯9. 生产环境落地建议9.1 先规划最小可用状态机不要一上来就设计十几个状态。状态越多Agent 和人工的维护成本越高。建议从“两个状态必保留”出发await_human和approved。其他状态按需扩展。9.2 把审核人和操作人分开记录审计是任务看板的核心价值。每次审核都要记录reviewer、时间、意见和操作内容。如果有条件再加一层不可篡改的审计日志例如将操作事件写入独立的task_audit_log表。这样一旦出现问题可以回答“谁在什么时间批准了什么”。9.3 高危操作必须做最小权限授权这里要特别提醒任务看板负责“审批”但不负责“提权”。生产环境中Agent 的执行账号应该遵循最小权限原则。比如清理日志的操作Agent 账号只能删除指定目录下的文件不能有整机删除权限。即使审核通过了Agent 能做的事也必须在预设边界内。9.4 为每个审批任务设置超时策略人工不会 7×24 小时在线。任务挂起超过一定时间必须有自动升级机制先通知值班群再过一段时间没响应就自动终止任务或降级执行。这个超时策略在MAX_WAIT参数的基础上还应该支持按任务类型差异化配置。9.5 用任务类型区分安全等级可以把任务类型分为三级L1无副作用Agent 自动执行不需要人工。L2有间接影响需要事后通知可批量处理。L3高风险必须事前审批。任务看板初期主要服务 L3。随着团队信任度提升再逐步开放 L2 的自动执行权限。9.6 警惕“审核疲劳”当审核任务积压过多时人的注意力会下降可能机械式点击批准。对策是控制进入人工审核的任务量合并同类项提供上下文摘要。如果 Agent 连续提交大量相似任务应该由编排器聚合后一次提交而不是每个小动作都请求审批。9.7 关注 Agent 多任务并发时的资源占用如果一个 Agent 的多个流程都处于“等待人工审核”状态这些流程如果都采用阻塞式等待会浪费大量资源。建议用一个“挂起-唤醒”模式替代线程阻塞并在每次挂起时释放相关资源。10. 从看板到真正的 Agent 工作流从本文的示例代码可以看出Human Task Board 本身并不复杂核心就是一个状态机加几个 HTTP 接口。真正难的不是写代码而是想清楚三件事哪一步需要人工介入、介入后 Agent 怎么恢复、介入过程有没有被审计。如果你现在正在做 Agent 项目可以从一个最小的场景开始验证选一个你认为风险最高的操作把它接到看板上先人工审批跑一周。你会发现这个流程虽然让“效率”看起来降低了但它换来的可控性和安全性在真实业务里远比那几秒钟更重要。下一步值得深入的方向有三个第一把看板从 SQLite 换成 PostgreSQL并补充 Web 前端让非技术人员也能参与审核。第二把轮询机制升级为事件驱动通过消息队列或 Webhook 实现审核完成后的即时唤醒。第三结合deep agents interrupt这类正在出现的框架能力把中断与恢复从“自己实现的轮询 步骤表”升级为框架级的原生支持。Agent 技术迭代很快但“人机协作需要可控边界”这个原则不会变。一块任务看板就是你给 Agent 划出的那条边界。建议先收藏备用等你真正开始做 Agent 工程化时按本文的框架搭起来试一次你会很快体会到它的价值。 Agent工程化必备：Human Task Board人工任务看板设计与实现
最近在梳理 Agent 工程化落地时我反复被一个问题卡住Agent 可以自动执行任务但当它需要做决策、改配置、删数据或者对外回复时真的应该让它独立完成吗答案是在大多数生产场景里不应该。如果团队里同时跑着三四个 Agent有的在写代码有的在更新需求文档有的在处理客户工单你怎么知道它们现在在干什么哪些任务需要你确认哪些任务已经卡住超过一小时了这时候缺的不是更强的模型而是一块“人工任务看板”。这篇文章我会从工程角度拆解 Human Task Board 的设计与实现包括任务状态机、Agent 侧接入、人工审核流程、Agent 中断与恢复机制并给出完整的 FastAPI SQLite 示例代码。读完你应该能照着搭出一个最小可用的“人审台”把它接到自己的 Agent 工作流里。1. 为什么 Agent 需要一块人工任务看板如果只看 DemoAgent 似乎无所不能它能规划、能调用工具、能写完一段代码。但一旦进入真实业务问题就变了——Agent 的每一步行动是否被授权、是否可回滚、是否有人复核这些问题 Demo 里根本不会出现。举一个实际场景。一个自动化运维 Agent 检测到磁盘使用率超过 85%按照预设策略应该清理日志目录。它执行了rm -rf /var/log/old。假设路径拼接有误或者清理范围超出了预期这个操作会带来什么后果如果旁边有一块任务看板Agent 在执行前先生成一个“高危操作审批”任务等人工确认后再执行风险边界就完全不一样了。再比如写代码场景。Agent 改动了一个核心模块的接口签名关联的调用方可能有十几个。它自称“已经全量修改完毕”但你敢直接合入主干吗更稳妥的做法是Agent 完成改动后在看板上生成一个“请人工检查代码改动”的任务开发者 review 通过后再进入下一步。所以Human Task Board 本质上解决的不是“Agent 能不能干活”而是“Agent 干了活之后谁负责、谁确认、谁兜底”。它是人机协作的一个中间层让 Agent 的自主性被约束在可控范围内。从材料中的热词也能看出这个趋势。toward efficient agents强调的是 Agent 的效率但效率不等于完全自主deep agents interrupt聊的则是 Agent 运行过程中的中断与插话机制。这两个方向合在一起指向的正是 Human-in-the-Loop人在回路的工程化效率靠 Agent安全靠任务看板 中断机制。一句话判断如果你的 Agent 只是本地玩具任务看板可有可无但只要 Agent 要接触真实数据、真实系统、真实用户人工任务看板就不是加分项而是必选项。2. Human Task Board 的核心概念与适用场景2.1 什么是 Human Task BoardHuman Task Board直译是“人工任务看板”。它不是一个特定产品而是一套机制为 Agent 产生的、需要人类参与的任务提供统一的登记、展示、审核、追踪能力。它和普通任务管理软件的区别在于普通任务管理软件的服务对象是人类而 Human Task Board 的服务对象是 Agent Human 的混合流程。任务可能由 Agent 创建由 Agent 推进只在关键节点停下来等人类确认然后再由 Agent 继续执行。2.2 五个核心对象一个最小可用的任务看板需要管理以下对象对象说明示例Task任务一次需要人工参与的独立工作单元“请审核高危操作清理日志目录”State状态任务当前所处阶段pending、await_human、approvedArtifact产物Agent 执行过程中产生的内容代码 diff、日志、报告、参数快照Comment评论人类或 Agent 对任务的补充说明“请补充影响范围再审核”Audit Log审计日志任务全生命周期的操作记录谁在什么时间批准了什么后面实现时我们会把 Artifact 和 Comment 作为任务的关联数据存储而不是单独建模这样对小型项目更省事。2.3 任务状态机从创建到关闭任务状态设计是整个看板最重要的部分。状态少了表达不清楚状态多了维护成本高。一组经过实践检验的状态定义如下pending 任务已创建等待调度或 Agent 开始处理 in_progress 任务已被 Agent 领取并正在执行 await_human 任务执行到关键节点等待人工审核或提供信息 approved 人工审核通过Agent 可以继续执行 rejected 人工驳回Agent 需要修改后重新提交或终止 cancelled 任务被取消流程终止 completed 任务已全部完成并关闭 failed 任务执行失败需要人工介入分析转换规则需要严格控制例如await_human只能变为approved、rejected、cancelled。approved之后 Agent 继续执行最终变为completed或failed。rejected之后 Agent 可以选择修改并重新提交为await_human也可以放弃任务。这样的状态机设计有一个直接好处任何时刻你都能回答“现在有多少任务在等人工”“哪些 Agent 正在执行什么”这两个问题。2.4 什么场景下最需要它高危操作审批删除数据、修改权限、发布上线、转账支付。代码变更审查Agent 生成代码 diff人工 review 后再合入。内容发布审核Agent 生成营销文案、客服回复人工确认后对外发送。多 Agent 协作协调多个 Agent 产出结果有依赖需要人工裁决优先级。数据标注与质检Agent 预标注人工抽检或修正。这些场景的共同点是操作不可逆或者对外有影响或者错误成本高。遇到这类场景不要赌 Agent 的“聪明”要依赖流程的“稳”。3. 整体架构设计看板放在 Agent 和系统之间在动手写代码前先明确架构位置。一个典型的 Agent 应用架构如下User / Human Reviewer ↑ | 审核、批准、驳回 ----------------------------- Human Task Board (看板) ----------------------------- ↑ 创建任务、更新状态、请求审批 ----------------------------- Agent Orchestrator (编排器) | | Agent A Agent B | | Tools Tools (代码库) (数据库) (外部API)看板层是 Agent 与外部世界之间的“闸门”。Agent 不直接执行高危操作而是先在看板上登记意图等人批准后再真正执行。这个模式有时被称为Human-in-the-Loop Gateway。这里有一个关键设计决策看板是同步阻塞还是异步通知。同步阻塞Agent 发起审批请求后挂起直到人工响应才继续。适合高危操作但 Agent 会一直占用资源。异步通知Agent 发起请求后去做别的任务人工响应后通过事件机制唤醒。适合常规审批但实现复杂度更高。对于第一版建议先做同步阻塞。原因很简单同步模型容易理解、容易排错而且高危操作的频率通常不高不会成为性能瓶颈。后面再升级为异步事件驱动。4. 环境准备与项目结构本文示例使用以下技术栈。版本以你本机实际安装为准思路和代码模式是通用的Python 3.9FastAPISQLAlchemy或直接使用 sqlite3本文为了减少依赖使用 sqlite3 标准库 Pydantic 做数据校验uvicornrequests 或 httpxAgent 侧调用看板 API 用如果你的 Agent 本身用 Python直接复用先创建项目目录mkdir human-task-board cd human-task-board mkdir task_board touch task_board/__init__.py最终项目结构如下human-task-board/ ├── task_board/ │ ├── __init__.py │ ├── models.py # 数据模型与状态定义 │ ├── storage.py # SQLite 存储层 │ ├── api.py # FastAPI 接口 │ └── cli_review.py # 命令行审核工具 ├── agent_side.py # 模拟 Agent 侧接入示例 ├── requirements.txt └── README.md依赖文件fastapi uvicorn pydantic httpx安装命令pip install -r requirements.txt5. 核心实现任务模型、存储层与 API5.1 任务数据模型先定义状态枚举与任务模型。这里用 Pydantic 做请求参数校验用标准库 sqlite3 做持久化尽量保持代码可读性。# 文件路径task_board/models.py from enum import Enum from typing import Optional, List from pydantic import BaseModel, Field from datetime import datetime class TaskState(str, Enum): PENDING pending IN_PROGRESS in_progress AWAIT_HUMAN await_human APPROVED approved REJECTED rejected CANCELLED cancelled COMPLETED completed FAILED failed class TaskCreate(BaseModel): title: str Field(..., min_length1, max_length200, description任务标题) description: str Field(, description任务详细说明) agent_name: str Field(..., description发起任务的 Agent 名称) task_type: str Field(general, description任务类型如 high_risk, code_review, content_publish) payload: dict Field(default_factorydict, description任务附带的数据如命令、参数、文件路径) priority: int Field(1, ge0, le10, description优先级数值越大越紧急) class TaskUpdate(BaseModel): state: Optional[TaskState] None comment: Optional[str] None reviewer: Optional[str] None class Task(TaskCreate): id: str state: TaskState created_at: str updated_at: str reviewer: Optional[str] None review_comment: Optional[str] None设计说明payload是字典类型用来携带 Agent 执行前后的上下文。比如高危操作场景可以放{command: rm -rf /var/log/old, target_host: 192.168.1.10}。task_type用于前端做不同渲染也方便统计不同类别任务的审核耗时。reviewer记录谁做了审核是审计的基础字段。5.2 SQLite 存储层存储层不需要复杂重点是保证读写字段完整、状态转换可追踪。# 文件路径task_board/storage.py import sqlite3 import json import uuid from datetime import datetime from .models import Task, TaskCreate, TaskState, TaskUpdate DB_PATH task_board.db def get_conn(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_conn() conn.execute( CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, title TEXT NOT NULL, description TEXT, agent_name TEXT NOT NULL, task_type TEXT, payload TEXT, priority INTEGER DEFAULT 1, state TEXT NOT NULL, reviewer TEXT, review_comment TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) ) conn.commit() conn.close() def now_str() - str: return datetime.now().isoformat(timespecseconds) def create_task(data: TaskCreate) - Task: task_id uuid.uuid4().hex[:12] created now_str() conn get_conn() conn.execute( INSERT INTO tasks (id, title, description, agent_name, task_type, payload, priority, state, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( task_id, data.title, data.description, data.agent_name, data.task_type, json.dumps(data.payload, ensure_asciiFalse), data.priority, TaskState.PENDING.value, created, created, ), ) conn.commit() conn.close() return get_task(task_id) def get_task(task_id: str) - Task | None: conn get_conn() row conn.execute(SELECT * FROM tasks WHERE id ?, (task_id,)).fetchone() conn.close() if row is None: return None return row_to_task(row) def list_tasks(state: str | None None, limit: int 50) - list[Task]: conn get_conn() if state: rows conn.execute( SELECT * FROM tasks WHERE state ? ORDER BY priority DESC, created_at DESC LIMIT ?, (state, limit), ).fetchall() else: rows conn.execute( SELECT * FROM tasks ORDER BY priority DESC, created_at DESC LIMIT ?, (limit,), ).fetchall() conn.close() return [row_to_task(row) for row in rows] def update_task(task_id: str, update: TaskUpdate) - Task | None: task get_task(task_id) if task is None: return None new_state update.state.value if update.state else task.state reviewer update.reviewer if update.reviewer else task.reviewer review_comment update.review_comment if update.review_comment else task.review_comment conn get_conn() conn.execute( UPDATE tasks SET state ?, reviewer ?, review_comment ?, updated_at ? WHERE id ? , (new_state, reviewer, review_comment, now_str(), task_id), ) conn.commit() conn.close() return get_task(task_id) def row_to_task(row: sqlite3.Row) - Task: return Task( idrow[id], titlerow[title], descriptionrow[description], agent_namerow[agent_name], task_typerow[task_type], payloadjson.loads(row[payload]), priorityrow[priority], stateTaskState(row[state]), reviewerrow[reviewer], review_commentrow[review_comment], created_atrow[created_at], updated_atrow[updated_at], )这个存储层用 JSON 序列化payload字段好处是不需要为不同任务类型建不同表坏处是不能直接用 SQL 对 payload 里的字段做复杂查询。对于任务看板这种低频场景完全够用。5.3 FastAPI 接口接口层提供四个核心能力创建任务Agent 调用。查询任务列表看板前端或 CLI 调用。查询单个任务详情页使用。更新任务状态人工审核使用。# 文件路径task_board/api.py from fastapi import FastAPI, HTTPException, Query from .models import Task, TaskCreate, TaskUpdate from . import storage app FastAPI(titleHuman Task Board API) app.on_event(startup) def startup(): storage.init_db() app.post(/tasks, response_modelTask, status_code201) def create_task(data: TaskCreate): Agent 创建一个人工审核任务 task storage.create_task(data) return task app.get(/tasks, response_modellist[Task]) def get_tasks(state: str | None Query(defaultNone), limit: int Query(default50, le200)): 查询任务列表可按状态过滤 if state and state not in [s.value for s in storage.TaskState]: raise HTTPException(status_code400, detailf非法状态: {state}) return storage.list_tasks(statestate, limitlimit) app.get(/tasks/{task_id}, response_modelTask) def get_task(task_id: str): task storage.get_task(task_id) if task is None: raise HTTPException(status_code404, detail任务不存在) return task app.post(/tasks/{task_id}/review, response_modelTask) def review_task(task_id: str, update: TaskUpdate): 人工审核接口批准、驳回或取消任务 if not update.state: raise HTTPException(status_code400, detail请指定目标状态) if update.state.value not in [approved, rejected, cancelled]: raise HTTPException(status_code400, detail审核操作只能为 approved / rejected / cancelled) task storage.update_task(task_id, update) if task is None: raise HTTPException(status_code404, detail任务不存在) return task这里把审核接口单独拆成/tasks/{task_id}/review而不是用通用更新接口是为了在语义上明确“这是人工审核动作”。后续如果要在审核时触发通知、写审计日志都可以在这个接口里扩展。启动服务uvicorn task_board.api:app --host 0.0.0.0 --port 8000 --reload启动后FastAPI 自带的文档地址是http://127.0.0.1:8000/docs可以直接在浏览器里手动创建和审核任务这对快速联调很有用。5.4 Agent 侧接入示例现在模拟一个 Agent 在遇到高危操作时先暂停执行并请求人工审批的接入方式。# 文件路径agent_side.py 模拟 Agent 使用 Human Task Board 的接入示例。 流程 1. Agent 分析问题形成执行计划。 2. 发现存在高危操作创建审批任务。 3. 轮询任务状态等待人工审核。 4. 审核通过则执行真实操作否则放弃操作并记录原因。 import time import httpx API_BASE http://127.0.0.1:8000 POLL_INTERVAL 3 # 轮询间隔单位秒 MAX_WAIT 120 # 最大等待时间单位秒 def request_human_approval(title: str, description: str, payload: dict) - str: 创建人工审批任务返回 task_id resp httpx.post( f{API_BASE}/tasks, json{ title: title, description: description, agent_name: ops-agent, task_type: high_risk, payload: payload, priority: 5, }, timeout10, ) resp.raise_for_status() return resp.json()[id] def wait_for_review(task_id: str) - dict: 阻塞等待人工审核结果 start time.time() while time.time() - start MAX_WAIT: resp httpx.get(f{API_BASE}/tasks/{task_id}, timeout10) resp.raise_for_status() task resp.json() if task[state] in (approved, rejected, cancelled): return task print(f[Agent] 等待审核中... 当前状态: {task[state]}) time.sleep(POLL_INTERVAL) raise TimeoutError(等待人工审核超时) def main(): payload { command: rm -rf /tmp/app_old_logs, target: prod-web-01, reason: 磁盘使用率超过 85%需要清理历史日志, } print([Agent] 检测到磁盘空间不足准备执行清理操作。) print([Agent] 该操作属于高危操作先请求人工审批。) task_id request_human_approval( title高危操作审批清理 prod-web-01 历史日志, descriptionAgent 检测到磁盘使用率超过 85%建议执行日志清理。, payloadpayload, ) print(f[Agent] 已创建审批任务: {task_id}) try: review_result wait_for_review(task_id) except TimeoutError as e: print(f[Agent] 审核超时放弃执行。{e}) return if review_result[state] approved: print([Agent] 人工审核通过开始执行清理操作。) # 在这里执行真实操作例如 subprocess.run(payload[command], shellTrue) print(f[Agent] 执行命令: {payload[command]}) # 执行完成后更新任务为 completed httpx.post( f{API_BASE}/tasks/{task_id}/review, json{state: completed, reviewer: ops-agent, review_comment: 清理任务已执行完成}, timeout10, ) elif review_result[state] rejected: print([Agent] 人工驳回了该操作不执行清理。) else: print([Agent] 任务被取消不执行清理。) if __name__ __main__: main()这段代码展示了最关键的接入模式先请求、再等待、通过后才执行。注意几个细节任务创建后Agent 进入轮询等待。生产环境中Agent 不一定真的 sleep 阻塞可以用回调或事件机制但轮询模式最容易理解和排错。审核通过后Agent 执行真实操作完成后把任务更新为completed形成闭环。如果审核超时Agent 必须有一个兜底策略。这里简单抛出异常生产环境可以改为“放弃执行并通知值班人员”。5.5 命令行人工审核工具人工侧需要一个简单的审核工具方便不打开浏览器时快速处理任务。# 文件路径task_board/cli_review.py 命令行审核工具 用法 python -m task_board.cli_review list python -m task_board.cli_review list --state await_human python -m task_board.cli_review approve task_id python -m task_board.cli_review reject task_id --comment 理由 import argparse import httpx API_BASE http://127.0.0.1:8000 def cmd_list(args): params {limit: args.limit} if args.state: params[state] args.state resp httpx.get(f{API_BASE}/tasks, paramsparams, timeout10) tasks resp.json() if not tasks: print(当前没有任务。) return for t in tasks: print(f[{t[id]}] {t[state]:12s} {t[priority]} {t[title]} agent{t[agent_name]}) def cmd_approve(args): resp httpx.post( f{API_BASE}/tasks/{args.task_id}/review, json{state: approved, reviewer: args.reviewer, review_comment: args.comment}, timeout10, ) if resp.status_code 404: print(任务不存在。) return resp.raise_for_status() print(f任务 {args.task_id} 已通过审核。) def cmd_reject(args): resp httpx.post( f{API_BASE}/tasks/{args.task_id}/review, json{state: rejected, reviewer: args.reviewer, review_comment: args.comment}, timeout10, ) if resp.status_code 404: print(任务不存在。) return resp.raise_for_status() print(f任务 {args.task_id} 已驳回。) def main(): parser argparse.ArgumentParser(descriptionHuman Task Board CLI) subparsers parser.add_subparsers(destcommand) list_parser subparsers.add_parser(list, help列出任务) list_parser.add_argument(--state, defaultNone, help按状态过滤) list_parser.add_argument(--limit, typeint, default20, help最大数量) list_parser.set_defaults(funccmd_list) approve_parser subparsers.add_parser(approve, help批准任务) approve_parser.add_argument(task_id, help任务 ID) approve_parser.add_argument(--reviewer, defaulthuman, help审核人) approve_parser.add_argument(--comment, default, help审核意见) approve_parser.set_defaults(funccmd_approve) reject_parser subparsers.add_parser(reject, help驳回任务) reject_parser.add_argument(task_id, help任务 ID) reject_parser.add_argument(--reviewer, defaulthuman, help审核人) reject_parser.add_argument(--comment, default, help驳回原因) reject_parser.set_defaults(funccmd_reject) args parser.parse_args() if not args.command: parser.print_help() return args.func(args) if __name__ __main__: main()这个 CLI 刻意保持轻量只实现列表、批准、驳回三个操作。实际团队使用中可以继续扩展“取消”“重新打开”“指派给某人”等命令也可以把同样的逻辑封装成 Web 前端。6. 关键机制Agent 中断与恢复前面实现了“Agent 等待人工审核”这个流程但还有一个更深入的问题没有解决当 Agent 被中断时它如何记住自己做到哪一步了恢复后从哪里继续这在 Agent 工程里通常叫interrupt机制。简单说Agent 的执行过程不应该是一段不可分割的“大函数”而应该是一系列可暂停、可恢复的步骤。6.1 中断的本质是持久化执行上下文我第一次实现中断机制时踩过一个坑以为中断就是让 Agent 进程睡眠等审核通过后再唤醒。但实际上Agent 进程可能因为超时被回收容器可能被重启审核人员可能几小时后才处理。所以中断绝不能依赖进程内存。正确做法是在每个可中断步骤前把 Agent 的执行上下文持久化到存储中。上下文至少包含当前任务 ID。当前步骤 ID 或步骤索引。已收集到的输入数据。下一步需要调用什么工具、传入什么参数。依赖人工确认的审批任务 ID。这样即使 Agent 进程重启也能从存储中恢复上下文继续执行。6.2 用“步骤表”代替“流程函数”我推荐的模式是把 Agent 的任务流程拆成显式的步骤表每一步记录状态。# 文件路径agent_ctx.py示意 STEPS [ {id: collect_metrics, action: gather_disk_usage, requires_human: False}, {id: request_cleanup, action: create_human_task, requires_human: True}, {id: exec_cleanup, action: run_clean_command, requires_human: False}, {id: report_result, action: notify_ops, requires_human: False}, ]Agent 主循环伪代码如下def run_agent_flow(task_id: str, ctx: dict): step_index ctx.get(step_index, 0) while step_index len(STEPS): step STEPS[step_index] if step[requires_human]: # 检查对应人工审批任务是否已通过 approval get_approval_by_agent_step(task_id, step[id]) if approval.state pending: # 未处理保存进度并返回等待被唤醒 save_ctx(task_id, {step_index: step_index}) return if approval.state rejected: # 人工驳回停止流程 mark_task_failed(task_id, 人工驳回) return # approved继续向下执行 execute_step(step, ctx) step_index 1 save_ctx(task_id, {step_index: step_index}) mark_task_completed(task_id)这个模式的要点是每一步执行后立即保存进度遇到需要人工确认的步骤就退出主循环外部审核通过后再触发一次 Agent 执行。这样 Agent 的“中断”就不再依赖阻塞等待而是变成了一种带持久化上下文的“挂起-唤醒”模型。6.3 唤醒方式的选择挂起之后如何唤醒常见有三种方式方式实现难度实时性适用场景轮询低有延迟内部工具任务量不大Webhook 回调中高看板与 Agent 同属一个系统消息队列事件较高高大规模多 Agent 集群第一版推荐轮询因为看板本身的数据量很小人工审核频率也不高轮询间隔 3 到 5 秒完全够用。如果后续 Agent 数量上来了再换成 Webhook 或消息队列。7. 运行流程与验证7.1 第一步启动看板服务在一个终端启动 API 服务cd human-task-board uvicorn task_board.api:app --host 0.0.0.0 --port 8000 --reload看到类似输出说明启动成功INFO: Uvicorn running on http://127.0.0.1:8000 INFO: Application startup complete.7.2 第二步创建一个人工审批任务另开一个终端用 curl 创建任务curl -X POST http://127.0.0.1:8000/tasks \ -H Content-Type: application/json \ -d { title: 高危操作审批清理 prod-web-01 历史日志, description: 磁盘使用率超过 85%需要清理日志, agent_name: ops-agent, task_type: high_risk, priority: 5, payload: { command: rm -rf /tmp/app_old_logs, target: prod-web-01 } }预期返回{ id: 3f1a9c2b6d4e, title: 高危操作审批清理 prod-web-01 历史日志, state: pending, agent_name: ops-agent, task_type: high_risk, priority: 5, payload: { command: rm -rf /tmp/app_old_logs, target: prod-web-01 }, created_at: 2025-01-01T10:00:00 }注意id字段后续操作需要用到。7.3 第三步查询待审核任务curl http://127.0.0.1:8000/tasks?stateawait_human这里你会发现刚创建的任务还是pending状态。在实际接入中Agent 领取任务后会把状态改成in_progress到达人工节点时再改成await_human。为了方便测试可以直接手动把状态改成await_human也可以让 Agent 在创建任务时就把初始状态设为await_human。7.4 第四步人工审核使用 CLI 完成审核# 查看任务列表 python -m task_board.cli_review list # 批准任务 python -m task_board.cli_review approve 3f1a9c2b6d4e --reviewer zhangsan --comment 确认清理不影响核心数据预期输出任务 3f1a9c2b6d4e 已通过审核。再查询任务详情状态应变为approvedcurl http://127.0.0.1:8000/tasks/3f1a9c2b6d4e返回中可以看到reviewer和review_comment字段已经被填充。这说明人工审核信息已经完整记录了。7.5 第五步跑通 Agent 接入示例启动服务后另开一个终端运行python agent_side.py此时 Agent 会创建审批任务进入轮询等待。你可以在另一个终端执行批准操作观察 Agent 的输出是否按预期打印“人工审核通过开始执行清理操作”。如果一直不批准Agent 会在MAX_WAIT时间后超时退出并打印“审核超时放弃执行”。这个行为就是我们前面说的兜底策略。7.6 判断成功与否的标准一个完整的闭环应该满足三个条件任务从await_human变为approved且reviewer有值。Agent 收到审核结果后只执行了被批准的操作没有提前执行。执行完成后任务最终变为completed或failed状态有终态。如果任务停留在中间状态超过预期时间优先检查 Agent 的轮询逻辑和 API 是否正常。8. 常见问题与排查思路问题现象可能原因排查方式解决方案创建任务时返回 422请求体缺少必填字段或字段类型错误查看 FastAPI 返回的错误详情检查title、agent_name是否为空payload是否为 JSON 对象Agent 一直显示等待审核但任务已经批准轮询间隔内未抓到最新状态或请求了错误的 API 地址手动 curl 查询任务详情确认API_BASE配置正确缩短轮询间隔检查是否有缓存层审核接口返回 400传入了非法目标状态查看接口文档中的状态枚举审核操作只能传approved、rejected、cancelled之一数据库文件被锁写入失败多个并发请求同时写 SQLite或存在长事务查看 SQLite 报错信息任务看板并发量不高时 SQLite 够用如果并发高换 PostgreSQL 并使用连接池Agent 进程重启后忘记恢复进度没有持久化步骤上下文查看步骤表数据是否存储引入步骤索引持久化每次执行后写入 storage审核人误操作批准了不该批的任务缺少二次确认和历史记录查询任务审计日志增加审核页面的二次确认弹窗记录操作人、时间和 IP便于事后追溯9. 生产环境落地建议9.1 先规划最小可用状态机不要一上来就设计十几个状态。状态越多Agent 和人工的维护成本越高。建议从“两个状态必保留”出发await_human和approved。其他状态按需扩展。9.2 把审核人和操作人分开记录审计是任务看板的核心价值。每次审核都要记录reviewer、时间、意见和操作内容。如果有条件再加一层不可篡改的审计日志例如将操作事件写入独立的task_audit_log表。这样一旦出现问题可以回答“谁在什么时间批准了什么”。9.3 高危操作必须做最小权限授权这里要特别提醒任务看板负责“审批”但不负责“提权”。生产环境中Agent 的执行账号应该遵循最小权限原则。比如清理日志的操作Agent 账号只能删除指定目录下的文件不能有整机删除权限。即使审核通过了Agent 能做的事也必须在预设边界内。9.4 为每个审批任务设置超时策略人工不会 7×24 小时在线。任务挂起超过一定时间必须有自动升级机制先通知值班群再过一段时间没响应就自动终止任务或降级执行。这个超时策略在MAX_WAIT参数的基础上还应该支持按任务类型差异化配置。9.5 用任务类型区分安全等级可以把任务类型分为三级L1无副作用Agent 自动执行不需要人工。L2有间接影响需要事后通知可批量处理。L3高风险必须事前审批。任务看板初期主要服务 L3。随着团队信任度提升再逐步开放 L2 的自动执行权限。9.6 警惕“审核疲劳”当审核任务积压过多时人的注意力会下降可能机械式点击批准。对策是控制进入人工审核的任务量合并同类项提供上下文摘要。如果 Agent 连续提交大量相似任务应该由编排器聚合后一次提交而不是每个小动作都请求审批。9.7 关注 Agent 多任务并发时的资源占用如果一个 Agent 的多个流程都处于“等待人工审核”状态这些流程如果都采用阻塞式等待会浪费大量资源。建议用一个“挂起-唤醒”模式替代线程阻塞并在每次挂起时释放相关资源。10. 从看板到真正的 Agent 工作流从本文的示例代码可以看出Human Task Board 本身并不复杂核心就是一个状态机加几个 HTTP 接口。真正难的不是写代码而是想清楚三件事哪一步需要人工介入、介入后 Agent 怎么恢复、介入过程有没有被审计。如果你现在正在做 Agent 项目可以从一个最小的场景开始验证选一个你认为风险最高的操作把它接到看板上先人工审批跑一周。你会发现这个流程虽然让“效率”看起来降低了但它换来的可控性和安全性在真实业务里远比那几秒钟更重要。下一步值得深入的方向有三个第一把看板从 SQLite 换成 PostgreSQL并补充 Web 前端让非技术人员也能参与审核。第二把轮询机制升级为事件驱动通过消息队列或 Webhook 实现审核完成后的即时唤醒。第三结合deep agents interrupt这类正在出现的框架能力把中断与恢复从“自己实现的轮询 步骤表”升级为框架级的原生支持。Agent 技术迭代很快但“人机协作需要可控边界”这个原则不会变。一块任务看板就是你给 Agent 划出的那条边界。建议先收藏备用等你真正开始做 Agent 工程化时按本文的框架搭起来试一次你会很快体会到它的价值。