用FastAPI构建Agent人工审批任务看板：从状态机到中断机制

现在很多人做 Agent 应用最纠结的问题不是“模型够不够聪明”而是Agent 到底该不该自己决定一切我在实际项目中见过两类极端。一类是给 Agent 绑定一堆工具让它“放手去干”结果它在测试环境删了数据库的一张表另一类是 Agent 每走一步都来问人五分钟一次弹窗最后使用方烦到直接关掉整个自动化流程。显然完全自主和处处打断都不对。真正的解法是把“人的确认”做成 Agent 工作流里一个可编程的检查点Agent 需要执行高风险操作时先在任务看板上挂起一个“待审批任务”人看一眼就知道它想干什么、涉及哪些文件或接口点一下通过它继续跑点一下拒绝它换条路。整个过程既不打断流程又把底线握在手里。这篇文章要讲的就是如何从零实现这样一个Human Task Board人类任务看板。我会用一个 FastAPI SQLite 原生前端的最小可运行项目带你跑通“Agent 提交任务 - 人工审批 - Agent 执行 - 状态实时更新 - 随时中断”的完整链路。看完你不仅能照着搭出一套看板服务还能理解这类系统在真实工程里的架构取舍和坑点。1. 为什么你的 Agent 需要一个任务看板先说一个反直觉的判断Agent 自动化程度越高越需要人工介入点而不是越少。原因很简单。Agent 的能力边界在快速扩展它可以读写文件、调用 API、操作浏览器、在服务器上执行命令但模型在单次会话里对上下文的理解有上限它很难判断“删除这条线上数据”和“删除这条测试数据”背后的业务后果。本地开发还能容忍犯错一旦接入生产系统一次擅自执行就可能造成不可逆损失。任务看板解决的正是这个矛盾。它把“Agent 是否执行某个操作”从模型的概率决策变成“人机协作的流程决策”。具体来说一套看板能带来四层价值第一层可见性。你不再需要翻 Agent 的日志才知道它刚才干了什么。看板上每一条任务都记录了哪个 Agent、在什么时间、提交了什么动作、当前处于什么状态。审计和回溯变得非常直接。第二层可控性。Agent 遇到需要确认的场景时不是自己硬扛而是把任务推送到看板进入 pending 状态。人工审批通过后它才继续。这就是近期社区里经常讨论的 “deep agents interrupt” 理念落地——中断不应该靠截停进程而应该靠流程内的检查点。第三层效率。把“人审”从实时打断改成异步审批人的注意力不用一直挂在看板上。Agent 可以先处理不需要审批的任务审批通过后再继续下一步。对需要人工深度参与的复杂任务这个模型明显更高效也正契合 “toward efficient agents” 的方向。第四层安全边界。我们可以在看板层做策略控制比如某些动作类型必须审批、某些 Agent 只能提交只读操作、异常任务可以直接终止。这些策略如果散落在各 Agent 代码里很难统一维护放在看板层就是一个集中的安全网关。2. 核心概念人在环中与任务状态机要设计一套任务看板先要理解三个核心概念。2.1 Human-in-the-Loop人在环中这是整套系统的设计哲学。它的意思是AI 系统在执行关键决策时需要把人类纳入决策闭环而不是完全替代人类。放在 Agent 场景里具体体现就是Agent 可以自主完成“分析、拆解、尝试”但在执行“删除、发送、支付、创建”这类有副作用的操作时必须经过审批。和它相对的是 Human-on-the-Loop后者指人类只做监控不参与每个决策只在异常时介入。我们的看板系统其实同时支持两种模式低风险任务可以配置为自动放行人只看记录高风险任务必须审批人是决策链的一环。2.2 任务状态机看板的核心不是界面而是任务状态机。状态设计得好不好直接决定系统复杂度和后续扩展空间。我这里用到的状态集合是状态含义允许的迁移pendingAgent 已请求等待人工审批- approved / rejectedapproved人工已批准执行- running / rejectedrejected人工已拒绝终态runningAgent 正在执行- completed / failed / interruptedinterrupted人工主动中断终态completed执行完成终态failed执行失败- pending重试关键设计点在两个地方。一是interrupted 必须是独立状态而不是标记位。因为中断可能发生在运行中的任意时刻Agent 需要知道自己是“正常完成”还是“被终断”才能决定要不要回滚局部操作。二是failed 可以回到 pending这样人工看过失败原因后可以选择让 Agent 重试或彻底拒绝。2.3 审批策略审批策略决定了哪些任务需要人审哪些可以自动放行。一般有三档always所有动作都必须审批最安全但人的负担重。risk-based按动作类型、资源路径、目标环境动态判断。比如删除文件需要审批读取文件不需要。auto-pass白名单动作直接放行只记录不打扰。在最小实现里我们先做成“所有任务都必须审批”把策略层留到最佳实践部分讨论。3. 架构设计与数据模型整体架构我尽量精简但保留了一个真实看板系统该有的组成部分。3.1 组件划分模块职责技术选型Task API任务提交、审批、状态查询、中断FastAPI存储层任务数据持久化SQLite生产可换 PostgreSQLAgent SDKAgent 侧调用看板的封装Python requests看板前端人工操作界面原生 HTML JavaScript轮询刷新Agent 模拟器演示用的虚拟 AgentPython 脚本这里要说明两个选型理由。为什么用轮询而不是 WebSocket因为看板更新的实时性要求不高几秒延迟完全可接受轮询实现简单、部署方便还能省掉连接管理逻辑。真实项目如果任务量很大再换成 WebSocket 或 SSE 也不迟。为什么前端用原生 JS因为我们要把注意力和代码量集中在后端流程上。原生 JS 足够完成看板渲染和轮询还省去 Node 工具链的安装成本。3.2 数据表设计任务表是唯一的核心表字段如下CREATE TABLE tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_name TEXT NOT NULL, title TEXT NOT NULL, description TEXT NOT NULL, action_type TEXT NOT NULL, target_resource TEXT NOT NULL DEFAULT , status TEXT NOT NULL DEFAULT pending, human_note TEXT NOT NULL DEFAULT , created_at TEXT NOT NULL, updated_at TEXT NOT NULL );字段拆解action_typeAgent 想执行的动作类型如delete_file、send_email、run_shell。target_resource动作目标如文件路径、接口地址、命令内容。human_note人工审批时填写的备注拒绝原因也可以写这里。created_at/updated_at用于时间线排序和审计。这张表麻雀虽小五脏俱全后续加“审批人”“超时时间”“重试次数”都只是加字段的事。3.3 交互流程核心链路是Agent 组装任务信息调用POST /api/tasks提交任务。任务进入pending状态看板前端轮询到新任务后展示。人工查看任务详情点击“通过”或“拒绝”调用审批接口。Agent 通过轮询或回调拿到审批结果继续执行或放弃。Agent 开始执行后把状态更新为running完成后更新为completed或failed。人工随时可以在看板上点击“中断”把任务从running变为interrupted。4. 环境准备与项目结构4.1 运行环境Python 3.9 或更高版本版本请以实际环境为准本文演示通用思路无需安装 Node.js前端是纯 HTML/JS操作系统不限Windows / macOS / Linux 均可4.2 依赖安装使用 pip 安装三个依赖即可pip install fastapi uvicorn requests三个依赖各自的作用fastapi提供路由、请求体校验和自动生成的 API 文档。uvicornASGI 服务器负责启动 FastAPI 应用。requestsAgent 模拟器调用 API 用。4.3 项目结构agent-task-board/ ├── backend/ │ ├── __init__.py │ ├── main.py # FastAPI 入口和全部接口 │ └── database.py # SQLite 初始化与连接 ├── agent/ │ └── simulate_agent.py # 模拟 Agent提交任务、等待审批、执行 ├── frontend/ │ ├── index.html # 看板页面 │ └── app.js # 看板逻辑 └── requirements.txt整个项目只有这几个文件方便你快速拷贝跑通。我会把核心代码都放在main.py里减少跳转。5. 后端实现FastAPI 任务服务5.1 数据库初始化先把数据库基础写好。文件路径backend/database.pyimport sqlite3 from pathlib import Path DB_PATH Path(__file__).parent / tasks.db def get_connection(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_connection() conn.execute( CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_name TEXT NOT NULL, title TEXT NOT NULL, description TEXT NOT NULL, action_type TEXT NOT NULL, target_resource TEXT NOT NULL DEFAULT , status TEXT NOT NULL DEFAULT pending, human_note TEXT NOT NULL DEFAULT , created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) ) conn.commit() conn.close()这里的关键点是通过conn.row_factory sqlite3.Row让查询结果可以按字段名访问前端拿到 JSON 时结构更自然。每次操作都新建连接是为了避免多线程下的连接共享问题演示代码保持简单。5.2 主服务与任务接口文件路径backend/main.pyimport sqlite3 import time from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from database import get_connection, init_db class TaskCreate(BaseModel): agent_name: str title: str description: str action_type: str target_resource: str class TaskAction(BaseModel): human_note: str asynccontextmanager async def lifespan(app: FastAPI): init_db() yield app FastAPI(lifespanlifespan) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) def now(): return time.strftime(%Y-%m-%d %H:%M:%S, time.localtime()) def row_to_dict(row): return dict(row) if row else None def get_task_or_404(task_id: int): conn get_connection() row conn.execute(SELECT * FROM tasks WHERE id ?, (task_id,)).fetchone() conn.close() if row is None: raise HTTPException(status_code404, detailTask not found) return row_to_dict(row) def update_status(task_id: int, new_status: str, human_note: str ): conn get_connection() if human_note: conn.execute( UPDATE tasks SET status ?, human_note ?, updated_at ? WHERE id ?, (new_status, human_note, now(), task_id), ) else: conn.execute( UPDATE tasks SET status ?, updated_at ? WHERE id ?, (new_status, now(), task_id), ) conn.commit() conn.close() app.post(/api/tasks) def create_task(task: TaskCreate): Agent 提交新任务默认状态为 pending。 conn get_connection() cursor conn.execute( INSERT INTO tasks (agent_name, title, description, action_type, target_resource, status, created_at, updated_at) VALUES (?, ?, ?, ?, ?, pending, ?, ?) , (task.agent_name, task.title, task.description, task.action_type, task.target_resource, now(), now()), ) conn.commit() task_id cursor.lastrowid conn.close() return get_task_or_404(task_id) app.get(/api/tasks) def list_tasks(status: str | None None): 查询任务列表可按状态过滤。 conn get_connection() if status: rows conn.execute( SELECT * FROM tasks WHERE status ? ORDER BY id DESC, (status,) ).fetchall() else: rows conn.execute(SELECT * FROM tasks ORDER BY id DESC).fetchall() conn.close() return [dict(row) for row in rows] app.get(/api/tasks/{task_id}) def get_task(task_id: int): return get_task_or_404(task_id) app.post(/api/tasks/{task_id}/approve) def approve_task(task_id: int, action: TaskAction None): 人工审批通过。 task get_task_or_404(task_id) if task[status] ! pending: raise HTTPException(status_code400, detailOnly pending tasks can be approved) note action.human_note if action else update_status(task_id, approved, note) return get_task_or_404(task_id) app.post(/api/tasks/{task_id}/reject) def reject_task(task_id: int, action: TaskAction None): 人工拒绝Agent 应放弃或调整方案。 task get_task_or_404(task_id) if task[status] ! pending: raise HTTPException(status_code400, detailOnly pending tasks can be rejected) note action.human_note if action else update_status(task_id, rejected, note) return get_task_or_404(task_id) app.post(/api/tasks/{task_id}/start) def start_task(task_id: int): Agent 收到审批结果后开始执行。 task get_task_or_404(task_id) if task[status] ! approved: raise HTTPException(status_code400, detailTask must be approved before start) update_status(task_id, running) return get_task_or_404(task_id) app.post(/api/tasks/{task_id}/complete) def complete_task(task_id: int): task get_task_or_404(task_id) if task[status] ! running: raise HTTPException(status_code400, detailOnly running tasks can be completed) update_status(task_id, completed) return get_task_or_404(task_id) app.post(/api/tasks/{task_id}/fail) def fail_task(task_id: int, action: TaskAction None): task get_task_or_404(task_id) if task[status] not in (running, approved): raise HTTPException(status_code400, detailInvalid status for fail) note action.human_note if action else update_status(task_id, failed, note) return get_task_or_404(task_id) app.post(/api/tasks/{task_id}/interrupt) def interrupt_task(task_id: int, action: TaskAction None): 人工中断运行中的任务。 task get_task_or_404(task_id) if task[status] ! running: raise HTTPException(status_code400, detailOnly running tasks can be interrupted) note action.human_note if action else update_status(task_id, interrupted, note) return get_task_or_404(task_id)这段代码有几个值得说的设计。状态校验放在接口层。每个接口开头都检查当前状态是否允许迁移这是状态机落地的关键。如果跳过这一步客户端随意调用complete很容易造成pending - completed这样的非法迁移审计数据就不可信了。Agent 执行与任务状态解耦。看板只负责记录状态不真正替 Agent 执行动作。真正的执行逻辑发生在 Agent 自己的环境里。这是出于安全考虑——如果看板服务本身被攻破至少它没有预置执行任意命令的能力。中断接口不做强约束。Agent 在轮询时发现自己被中断需要自行决定如何收尾。这个设计背后是分布式系统的一个常识中断请求可以很快下发但 Agent 的清理工作快不了也没有必要在看板里等到它清理完。5.3 测试接口是否可用后端代码完成后先用 uvicorn 启动验证cd backend uvicorn main:app --host 0.0.0.0 --port 8000启动成功后浏览器访问http://localhost:8000/docs可以看到 FastAPI 自动生成的 Swagger 接口文档。到这一步任务服务的骨架已经通了。6. Agent 接入与前端看板6.1 Agent 侧接入封装真实场景里Agent 会通过函数调用或工具调用来操作看板。这里先给一个 Python 封装文件路径agent/agent_client.pyimport requests class TaskBoardClient: def __init__(self, base_urlhttp://localhost:8000): self.base_url base_url def submit(self, agent_name, title, description, action_type, target_resource): resp requests.post(f{self.base_url}/api/tasks, json{ agent_name: agent_name, title: title, description: description, action_type: action_type, target_resource: target_resource, }) resp.raise_for_status() return resp.json()[id] def wait_for_approval(self, task_id, interval2, timeout120): 阻塞等待人工审批结果。 import time deadline time.time() timeout while time.time() deadline: task requests.get(f{self.base_url}/api/tasks/{task_id}).json() if task[status] approved: return approved if task[status] rejected: return rejected time.sleep(interval) return timeout def start(self, task_id): requests.post(f{self.base_url}/api/tasks/{task_id}/start) def complete(self, task_id): requests.post(f{self.base_url}/api/tasks/{task_id}/complete) def fail(self, task_id, note): requests.post(f{self.base_url}/api/tasks/{task_id}/fail, json{human_note: note})这里最值得关注的是wait_for_approval方法。它通过轮询任务状态来感知人工审批结果而不是让 Agent 直接监听数据库。这样做的好处是 Agent 和看板之间只有 HTTP 依赖Agent 换语言、换框架都不影响看板侧逻辑。6.2 模拟 Agent 示例有了客户端封装写一个模拟 Agent 就很直接。文件路径agent/simulate_agent.pyimport time import random from agent_client import TaskBoardClient def run(): client TaskBoardClient() tasks [ { title: 清理临时目录文件, description: 检测到 /tmp/agent_workspace 下有 12 个中间产物建议删除。, action_type: delete_file, target_resource: /tmp/agent_workspace, }, { title: 发送项目周报邮件, description: 将本周代码变更摘要发送给 project-teamexample.com。, action_type: send_email, target_resource: project-teamexample.com, }, { title: 调用外部天气 API, description: 获取上海未来三天天气用于行程安排模块。, action_type: call_api, target_resource: https://api.example.com/weather, }, ] for idx, task_info in enumerate(tasks, 1): print(f[Agent] 提交任务 {idx}: {task_info[title]}) task_id client.submit( agent_namedemo-agent-01, titletask_info[title], descriptiontask_info[description], action_typetask_info[action_type], target_resourcetask_info[target_resource], ) print(f[Agent] 等待人工审批task_id{task_id} ...) result client.wait_for_approval(task_id) if result rejected: print(f[Agent] 任务被拒绝切换到备选方案。) continue if result timeout: print(f[Agent] 审批超时放弃该任务。) continue print(f[Agent] 审批通过开始执行 ...) client.start(task_id) time.sleep(random.uniform(1, 3)) if random.random() 0.15: print(f[Agent] 执行过程中发生异常标记为失败。) client.fail(task_id, note模拟执行异常) else: print(f[Agent] 执行完成。) client.complete(task_id) # 间隔几秒再提交下一条 time.sleep(2) if __name__ __main__: run()模拟 Agent 告诉我们一个重要的实践Agent 提交任务后不应当无条件阻塞等待审批而应当有超时和降级逻辑。代码里通过timeout和result rejected分支处理了两种常见异常场景生产环境还会增加“审批超时后撤回任务”“拒绝后按人类反馈重写请求”等策略。6.3 前端看板实现前端部分我用单个 HTML 文件承载样式和结构JS 逻辑放在单独文件里。先看结构文件frontend/index.html!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 titleAgent Task Board/title style * { box-sizing: border-box; margin: 0; padding: 0; } body { font-family: -apple-system, PingFang SC, Microsoft YaHei, sans-serif; background: #f5f6f8; padding: 24px; } .header { max-width: 1200px; margin: 0 auto 20px; display: flex; justify-content: space-between; align-items: center; } .header h1 { font-size: 22px; } .header .status { color: #888; font-size: 14px; } .board { max-width: 1200px; margin: 0 auto; display: grid; grid-template-columns: repeat(3, 1fr); gap: 16px; } .column { background: #fff; border-radius: 12px; padding: 16px; box-shadow: 0 1px 4px rgba(0,0,0,0.06); } .column h2 { font-size: 15px; margin-bottom: 12px; padding-bottom: 8px; border-bottom: 1px solid #eee; } .task-card { border: 1px solid #eee; border-radius: 8px; padding: 12px; margin-bottom: 12px; background: #fafbfc; } .task-card h3 { font-size: 14px; margin-bottom: 6px; } .task-card .meta { font-size: 12px; color: #999; margin-bottom: 8px; } .task-card .desc { font-size: 13px; color: #444; margin-bottom: 10px; } .task-card .actions { display: flex; gap: 8px; flex-wrap: wrap; } button { border: none; border-radius: 6px; padding: 6px 12px; font-size: 13px; cursor: pointer; } .btn-approve { background: #22c55e; color: #fff; } .btn-reject { background: #ef4444; color: #fff; } .btn-interrupt { background: #f59e0b; color: #fff; } .empty { color: #bbb; font-size: 13px; text-align: center; padding: 20px 0; } /style /head body div classheader h1Agent 任务看板/h1 div classstatus人工审批工作台/div /div div classboard div classcolumn h2待审批/h2 div idpending-list/div /div div classcolumn h2执行中/h2 div idrunning-list/div /div div classcolumn h2已完成 / 其他/h2 div iddone-list/div /div /div script srcapp.js/script /body /html再看逻辑文件frontend/app.jsconst API_BASE http://localhost:8000; async function fetchTasks() { const resp await fetch(${API_BASE}/api/tasks); return resp.json(); } async function action(taskId, actionName) { const note prompt(填写备注可选); await fetch(${API_BASE}/api/tasks/${taskId}/${actionName}, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ human_note: note || }), }); refreshBoard(); } function renderTaskCard(task) { const card document.createElement(div); card.className task-card; const title document.createElement(h3); title.textContent #${task.id} ${task.title}; const meta document.createElement(div); meta.className meta; meta.textContent ${task.agent_name} | ${task.action_type} | ${task.created_at}; const desc document.createElement(div); desc.className desc; desc.textContent task.description; if (task.human_note) { const note document.createElement(div); note.className meta; note.textContent 备注: ${task.human_note}; card.appendChild(note); } card.appendChild(title); card.appendChild(meta); card.appendChild(desc); if (task.status pending) { const actions document.createElement(div); actions.className actions; const approveBtn document.createElement(button); approveBtn.className btn-approve; approveBtn.textContent 通过; approveBtn.onclick () action(task.id, approve); const rejectBtn document.createElement(button); rejectBtn.className btn-reject; rejectBtn.textContent 拒绝; rejectBtn.onclick () action(task.id, reject); actions.appendChild(approveBtn); actions.appendChild(rejectBtn); card.appendChild(actions); } if (task.status running) { const actions document.createElement(div); actions.className actions; const interruptBtn document.createElement(button); interruptBtn.className btn-interrupt; interruptBtn.textContent 中断; interruptBtn.onclick () action(task.id, interrupt); actions.appendChild(interruptBtn); card.appendChild(actions); } return card; } function clearList(element) { while (element.firstChild) { element.removeChild(element.firstChild); } } async function refreshBoard() { const tasks await fetchTasks(); const pendingList document.getElementById(pending-list); const runningList document.getElementById(running-list); const doneList document.getElementById(done-list); clearList(pendingList); clearList(runningList); clearList(doneList); const pending tasks.filter(t t.status pending); const running tasks.filter(t t.status running); const done tasks.filter(t ![pending, running].includes(t.status)); if (pending.length 0) { pendingList.innerHTML div classempty暂无待审批任务/div; } else { pending.forEach(t pendingList.appendChild(renderTaskCard(t))); } if (running.length 0) { runningList.innerHTML div classempty暂无执行中任务/div; } else { running.forEach(t runningList.appendChild(renderTaskCard(t))); } if (done.length 0) { doneList.innerHTML div classempty暂无已结束任务/div; } else { done.forEach(t doneList.appendChild(renderTaskCard(t))); } } // 每 3 秒轮询一次任务列表 refreshBoard(); setInterval(refreshBoard, 3000);前端逻辑的核心是一个refreshBoard函数它把后端返回的任务按状态分到三列里渲染。轮询间隔设了 3 秒这是权衡实时性和服务端压力的结果。如果只是看板展示3 秒完全够用如果关乎审批后 Agent 立即执行也可以在审批接口返回后主动触发一次前端刷新而不是干等下一次轮询。这里的用户交互我做了简化用prompt输入审批备注。真实项目这一块要换成弹窗表单甚至支持预设审批意见和快捷键但最小示例里已经足够演示完整流程。7. 运行与效果验证7.1 启动服务端打开终端启动 FastAPI 服务cd agent-task-board/backend uvicorn main:app --host 0.0.0.0 --port 8000 --reload看到类似输出说明启动成功INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Application startup complete.7.2 启动前端看板前端不需要构建工具直接在浏览器打开frontend/index.html即可。如果浏览器因为安全策略不允许跨域加载可以通过任意静态服务器托管cd agent-task-board python -m http.server 8080 --directory frontend然后访问http://localhost:8080/index.html。7.3 启动模拟 Agent新开一个终端运行模拟 Agentcd agent-task-board/agent python simulate_agent.py此时看板“待审批”列会出现任务卡片。你会看到类似这样的输出[Agent] 提交任务 1: 清理临时目录文件 [Agent] 等待人工审批task_id1 ...在看板上找到这张卡片点击“通过”返回 Agent 终端会看到它开始执行并在几秒后完成[Agent] 审批通过开始执行 ... [Agent] 执行完成。此时看板上任务从“待审批”移到“执行中”最终出现在“已完成 / 其他”列。走完这个循环等于把整套人机协作闭环验证通过。7.4 验证中断机制中断是这套系统里最容易出问题的一环建议单独测一次。把模拟 Agent 代码改成在start之后增加一个较长的睡眠时间比如 10 秒然后再次运行。趁任务处于running状态在看板上点击“中断”。Agent 终端不会立刻退出而是“执行完成”或“执行失败”因为我们的模拟 Agent 没有在每次轮询时检查中断标记。真实项目里Agent 必须在多次动作之间定期查询任务状态。如果把 Agent 的长时间任务拆成多个小步骤每步之间检查一次看板中断才能真正生效。这也是“deep agents interrupt”在工程上最常见的落地点中断是一种协作协议不是强杀进程。7.5 验证失败重试Agent 代码里留了 15% 概率执行失败。失败后任务出现在“已完成 / 其他”列为failed状态。目前后端接口里没有实现failed - pending的重试逻辑如果你希望人工点击“允许重试”可以仿照approve增加一个retry接口把failed状态的记录改回pending。这一步可以作为扩展练习。8. 常见问题与排查思路问题现象可能原因排查方式解决方案启动后访问 /docs 报 404uvicorn 没在 backend 目录下启动找不到 main检查启动时的工作目录先cd backend再启动前端看板一直无数据跨域被拦截或 API 地址错误打开浏览器控制台看 Network 和 Console确认 FastAPI 加了 CORS 中间件用静态服务器托管前端而非 file:// 直接打开Agent 提交任务后一直 pending忘记在服务端开放审批接口或前端按钮没触发检查 Agent 终端是否有报错浏览器点“通过”后看 Network 响应码确认后端接口正常返回前端跨域配置无误审批时返回 400 “Only pending tasks can be approved”任务状态不是 pending比如已经被其他终端处理查询当前任务状态刷新看板确认任务状态避免多个终端重复审批同一任务Agent 一直等到 timeout人工没有审批或审批后任务状态未正确更新看板刷新后任务是否从待审批消失检查数据库更新逻辑确认approve后状态为approved中断后 Agent 仍继续运行Agent 侧没有定期查询状态只是做了一次性等待查看 Agent 日志执行时间线在 Agent 长任务内部增加状态轮询检查点SQLite 文件多进程读写报 database is locked演示代码每次操作新建连接并发写时会锁观察高频并发下的报错日志生产环境换 PostgreSQL或启用 WAL 模式、合理管理连接池9. 最佳实践与工程建议9.1 状态机必须收敛在服务端前端只是渲染器后端才是状态权威。前端可以自由点击任何按钮但后端必须拒绝非法状态迁移。这个原则保证了即使有人绕过前端直接调 API也无法制造一条非法状态记录。生产环境建议把状态机定义成一个显式表可以用枚举类或者状态机库来管理不要在业务代码里到处散落if status ...。9.2 审批策略要集中配置不要把“哪种任务需要审批”写在各个 Agent 里。推荐放在看板的配置中心例如使用一个policies.json{ risk_levels: { delete_file: high, send_email: medium, call_api: low }, approval_required: [high, medium], auto_pass: [low] }Agent 提交任务时把action_type带上看板根据策略自动决定进入pending还是直接approved。这样当你发现某一类操作风险变高只需要改配置不需要重新部署所有 Agent。9.3 为 Agent 提供“审批中”的回退策略模拟 Agent 里已经展示了超时和拒绝两个分支。更完整的策略还包括审批超时后任务自动挂起Agent 转去处理其他低风险任务避免整条工作流阻塞。审批被拒后Agent 根据human_note修改请求内容换一种更安全的方案重新提交。同一类任务连续被拒多次Agent 停止自动重试进入人工复核队列。9.4 日志和审计必须保留原始请求任务被修改可以但必须有迹可循。建议在任务表之外增加task_events表记录每个任务的完整生命周期CREATE TABLE task_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, event_type TEXT NOT NULL, operator TEXT NOT NULL, note TEXT NOT NULL DEFAULT , created_at TEXT NOT NULL );每次状态变更都插入一条事件记录。这不仅能满足审计要求将来做“Agent 行为分析”时也是宝贵的数据来源。9.5 权限与最小权限原则看板本身会成为系统的安全网关所以看板的权限必须比普通业务系统更严格。建议审批操作必须登录支持角色控制比如管理员能审批高风险操作普通成员只能查看。Agent 提交任务用的 token 和人工审批用的 token 分离Agent token 不能调用审批接口。生产环境的看板服务单独部署与其他核心业务服务隔离避免因为看板被攻破导致连锁风险。9.6 从看板走向全流程可观测这套最小实现已经具备了“任务提交-审批-执行-完成”的链路。往长远看可以把看板升级为真正的控制平面接入指标监控统计每个 Agent 的审批通过率、平均等待时长、中断次数接入告警某个任务长时间 pending 时提醒审批人接入策略引擎根据历史数据动态调整审批阈值让低风险任务自动流转高风险任务严格把关。这也正好回应了社区里 “toward efficient agents” 的核心诉求让 Agent 高效不能靠放开所有限制而要靠建立精准的授权和干预机制。9.7 不要跳过本地测试最后一条建议最朴素也最容易被忽略。任何对状态机、审批策略、中断逻辑的改动都至少要在本地跑一遍完整链路包括通过、拒绝、中断、失败、重试五种情况。这套系统是给人机协作兜底的它出问题Agent 就失去约束后果通常会放大到业务层面。小结整套“Human task board for my agents”的最小实现核心不是前端界面而是三个设计决策用任务状态机管理 Agent 动作的生命周期。用服务端接口强制状态迁移保证数据可信。用审批策略控制哪些动作需要人介入在效率和风险之间做显式权衡。你可以在本文代码的基础上进一步加入审批策略、事件日志、权限控制、中断检查点甚至把看板扩展成多 Agent 共用的调度中心。跑通最小闭环之后这部分工程能力会直接提高你手上 Agent 项目的可靠性和可审计性也让你在跟团队解释“Agent 到底干了什么”的时候不再只能拿日志截图说话。建议先把这版代码保存好当作后续扩展的基线。