想象这样一个招聘网站发布岗位的不是 HR而是一个运行在云端的 AI Agent浏览简历的不是猎头而是另一个 Agent最终拍板发 offer 的还是一个 Agent。人和人在这个流程里几乎不出现。这不是科幻设定当我们决定搭建一个“雇主不是人类”的 job board 时第一步就把招聘系统的常规逻辑推倒了。项目做完后最值得写的不是“AI Agent 能发任务、能完成任务”这个结果而是过程中到底有哪些环节坏了。模型理解任务的能力没有让我们太意外真正让系统反复出问题的偏偏是那些在人与人协作中根本不算事的细节身份怎么认证、任务怎么防重复认领、Agent 跑一半失联怎么办。这篇文章没有太多炫技的模型方案更多是一次 Agent 业务系统的故障复盘。无论你是否打算做 Agent 招聘平台只要你在接触 Agent 任务分发、Agent 协作、Agent 工作流文中这些坑大概率都会遇到。下面先解释为什么会有这种平台再拆解它的核心架构然后按“最先坏掉的地方”逐个复盘最后给出可以复用的工程原则。1. 为什么需要这样一个“非人类”任务市场首先要理解一个趋势AI Agent 正在从“聊天机器人”变成“业务执行者”。当 Agent 需要完成真实任务时它必须解决一个前置问题——任务从哪里来传统的做法是人工把任务描述透传给 Agent。但当一个 Agent 需要处理成百上千个异构任务时人工分发就不现实了。这个时候Agent 之间需要一种结构化的任务分发协议而 job board 是目前的合理形态。这类平台在本质上不是招聘网站而是一个有准入控制、有状态约束、有结果校验的 Agent 协作总线。你可以把它理解为任务版的“应用市场”雇主 Agent 发布任务执行 Agent 认领任务平台负责撮合、追踪和仲裁。从业务角度看它解决的是三个具体问题任务供给的规模化Agent 不再依赖人工逐个派单而是从共享任务池中按能力和报价自主选择。任务表达的标准化不同 Agent 对任务的描述方式差异极大平台必须把自由文本收敛成结构化定义。执行结果的可验证Agent 执行完后平台必须有能力判断结果是真是假、是否完整而不是盲目信任返回 JSON。很多人在做 Agent 系统时都会把精力放在怎么让单个 Agent 变聪明。但这个项目告诉我们当 Agent 成为业务流程中的独立角色时真正的复杂度在于它们之间的交互规则。2. 平台的核心架构与角色设计这个系统的角色只有三个但每个角色都值得仔细定义。角色对应传统招聘系统在本平台的行为雇主 Agent企业 HR发布任务、设置预算、审核交付结果执行 Agent求职者浏览任务、认领任务、交付结果平台 Agent猎头 背调 法务身份认证、任务匹配、状态追踪、结果校验、结算平台侧的核心模块可以分为六层接入层、任务建模层、匹配调度层、状态管理层、执行校验层、审计结算层。接入层负责身份认证任务建模层把自然语言任务转成结构化 Job匹配调度层根据 Agent 技能档案和任务要求做撮合状态管理层维护任务的完整生命周期执行校验层负责超时熔断和结果合法性检查审计结算层记录所有动作并为后续计费提供基础。这个架构在设计时看起来合理真正跑起来后我们才发现每一层都藏着“非人类参与者”带来的新问题。比如身份认证层人类求职者可以靠手机号、邮箱、人脸识别确认身份Agent 呢它没有生物特征也可能频繁更换运行环境。再比如状态管理层人类不会同时点击十个岗位的“立即入职”按钮但 Agent 在并发请求下可能因为重试机制一口气认领十次任务。后面五个章节按故障爆发的时间顺序逐个复盘最典型的“坏掉”场景。3. Agent 身份与认证第一个崩掉的模块系统上线测试后最先出问题的是认证模块。第一类问题是身份粒度。一个企业可能有多个自动化流程每个流程都由不同 Agent 执行。如果认证只做到“企业账号”粒度就无法区分请求到底来自哪个 Agent一旦某个流程被攻破整个企业账号的权限都失控。第二类问题是重放攻击。Agent 之间的调用是自动化的请求参数和签名如果长期不变攻击者完全可以把合法请求录制下来反复提交。解决思路不是引入复杂的 OAuth而是先做一个“够用但严格”的 Agent 级认证每个 Agent 在注册后获得全局唯一 agent_id 和独立密钥每次调用必须携带签名头和时间戳。签名规则采用 HMAC-SHA256密钥只保存在服务端和 Agent 本地的环境变量里。下面是 FastAPI 中的中间件实现核心是校验签名和时间窗口。先说明这个示例把密钥放在内存里生产环境务必换成密钥管理服务。# 文件路径app/security/agent_auth.py import hashlib import hmac import time from fastapi import FastAPI, Request from fastapi.responses import JSONResponse app FastAPI() # 生产环境请从密钥管理服务读取不要硬编码 ACTIVE_AGENTS { agent_employer_001: { agent_secret: sk_live_replace_with_secret, status: active, } } def verify_agent_signature( agent_id: str, timestamp: str, signature: str, request_body: bytes, ) - bool: 校验 Agent 请求签名HMAC-SHA256(secret, timestamp body) record ACTIVE_AGENTS.get(agent_id) if not record or record.get(status) ! active: return False try: ts int(timestamp) except ValueError: return False # 时间窗口 5 分钟防止重放攻击 if abs(int(time.time()) - ts) 300: return False payload timestamp.encode(utf-8) request_body expected hmac.new( record[agent_secret].encode(utf-8), payload, hashlib.sha256, ).hexdigest() return hmac.compare_digest(expected, signature) app.middleware(http) async def agent_auth_middleware(request: Request, call_next): # 仅对 /v1/tasks 开头的接口启用 Agent 签名校验 if not request.url.path.startswith(/v1/tasks): return await call_next(request) agent_id request.headers.get(X-Agent-Id) timestamp request.headers.get(X-Timestamp) signature request.headers.get(X-Agent-Signature) if not all([agent_id, timestamp, signature]): return JSONResponse({error: missing_auth_headers}, status_code401) body await request.body() if not verify_agent_signature(agent_id, timestamp, signature, body): return JSONResponse({error: invalid_signature}, status_code401) # 注入当前调用方身份后续业务逻辑使用 request.state.agent_id agent_id return await call_next(request)这段代码里有一个容易踩的坑FastAPI 中间件读取 request.body() 后路由处理函数可能无法再次读取 body导致参数解析异常。生产环境中应该把 body 缓存到 request.state.body或者在使用流式请求时调整读取策略。否则就会出现“认证通过了但业务接口拿不到参数”的诡异问题。签名认证解决了两个问题平台能确认请求来自哪个 Agent同时因为请求体参与签名中间做任何篡改都会校验失败。但这只是第一步。认证通过之后Agent 提交的任务内容又是一场新的灾难。4. 任务描述解析自由文本到结构化定义的收敛Agent 发布任务和人类写 JD 有一个明显差异人类会考虑看的人的阅读体验Agent 不会。我们拿到过几千字的任务描述里面混合了 JSON、Markdown、纯文本和过期的 Python 配置片段。直接把这段文本交给另一个 Agent 去执行它根本不知道该听哪一部分。这个问题的本质不是模型能力不足而是任务定义缺失。没有结构化的 Job就无法做匹配、无法做预算校验、无法做结果验收。平台必须在入口处把任务强迫收敛到统一 Schema。我们设计了 JobDefinition 模型核心字段包括标题、描述、技能要求、预算、截止时间和可见性。任何 Agent 提交的任务都必须经过两个阶段优先尝试解析 Agent 直接提交的 JSON 结构。如果 JSON 解析失败或字段缺失再用大模型从自由文本中抽取。抽取后仍不满足 Schema 的任务直接拒绝并附带错误码绝不创建脏数据。# 文件路径app/services/job_parser.py import json import logging from typing import Any, Dict, List from pydantic import BaseModel, Field, ValidationError logger logging.getLogger(__name__) class JobRejected(Exception): 任务不满足平台 Schema 时抛出的异常 def __init__(self, reason: str): self.reason reason class JobDefinition(BaseModel): title: str Field(..., max_length128) description: str Field(..., max_length4000) required_skills: List[str] Field(default_factorylist) budget: Dict[str, Any] Field(default_factorydict) deadline: str Field(default) visibility: str Field(defaultprivate) def llm_extract_job(raw_text: str) - dict: 调用大模型从自由文本中抽取结构化任务字段。 这里省略了具体模型调用细节重点是 1. prompt 中明确要求只输出 JSON 2. 输出必须包含 title/description/required_skills 3. 对缺失字段使用默认值不要随意追加平台不认识的字段。 # 示意实现实际项目中请替换为真实模型调用 return { title: 使用 LLM 抽取的标题, description: raw_text[:4000], required_skills: [], } def parse_agent_job(raw_text: str, raw_json: str) - JobDefinition: 解析流程 1. 优先尝试直接把 Agent 提交的 JSON 映射为结构化 Job。 2. 如果 JSON 缺失关键字段再用大模型从 raw_text 中抽取。 3. 如果抽取后仍不满足 schema则返回错误码 REJECT。 if raw_json: try: data json.loads(raw_json) return JobDefinition(**data) except (json.JSONDecodeError, ValidationError) as exc: logger.warning(structured parse failed: %s, exc) extracted llm_extract_job(raw_text) try: return JobDefinition(**extracted) except ValidationError as exc: raise JobRejected(reasonfjob_schema_invalid: {exc.errors()})对应地Agent 提交任务时必须带上一个 JSON Schema 文档。这样两边都有明确的契约而不是靠“大模型智能”糊弄。{ schema_version: 1.0, title: { type: string, required: true, max_length: 128 }, description: { type: string, required: true, max_length: 4000 }, required_skills: { type: array, items: { type: string }, required: false }, budget: { type: object, properties: { currency: { type: string, enum: [USD, CNY] }, amount: { type: number, minimum: 0 } }, required: false }, deadline: { type: string, required: false }, visibility: { type: string, enum: [public, private], default: private } }真正重要的不是这个 Schema 本身而是“拒绝原则”。很多系统为了体验友好会在大模型解析失败时创建一个半成品任务让下游 Agent 去猜。这是灾难的开始。下游 Agent 一旦开始猜测字段含义整个平台的任务质量就失控了。宁可拒绝任务也不要生产脏任务。5. 任务状态一致性重复认领与状态丢失这是整个项目里影响最严重的一类故障。问题表现有两种。第一种是重复认领执行 Agent 因为网络抖动触发了重试同一个任务被同一个 Agent 认领了两次如果平台没做唯一性校验两个 Worker 会同时执行同一个任务造成资源浪费甚至两个人交付不同结果。第二种是状态卡死Worker 认领任务后运行到一半崩溃任务永远停留在 CLAIMED 状态没有其他 Agent 能接手。解决重复认领靠的是条件更新而不是“先查询再更新”。先查再改的模式在并发场景下必然出现竞态条件。我们用一条 UPDATE 语句只允许状态为 OPEN 的任务被认领并且通过影响行数判断是否抢占成功。-- 任务认领只有当前状态为 OPEN 的任务才能被认领 UPDATE task_board SET status CLAIMED, worker_agent_id :worker_agent_id, lease_expire_at NOW() INTERVAL 10 minutes, version version 1, updated_at NOW() WHERE task_id :task_id AND status OPEN AND (worker_agent_id IS NULL OR worker_agent_id :worker_agent_id) RETURNING task_id;如果这条 SQL 返回 0 行说明任务已经被其他 Agent 抢走直接返回 409 即可。# 文件路径app/routes/tasks.py from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import text from sqlalchemy.orm import Session router APIRouter(prefix/v1/tasks) router.post(/{task_id}/claim) def claim_task( task_id: str, agent_id: str Depends(get_current_agent_id), db: Session Depends(get_db), ): result db.execute( text( UPDATE task_board SET status CLAIMED, worker_agent_id :agent_id, lease_expire_at NOW() INTERVAL 10 minutes, version version 1, updated_at NOW() WHERE task_id :task_id AND status OPEN AND (worker_agent_id IS NULL OR worker_agent_id :agent_id) RETURNING task_id ), {task_id: task_id, agent_id: agent_id}, ) row result.fetchone() if not row: raise HTTPException(status_code409, detailtask_already_claimed) db.commit() return {task_id: task_id, status: CLAIMED}解决状态卡死靠的是租约机制lease。每个 CLAIMED 状态的任务都带一个 lease_expire_at 时间戳表示“允许当前 Worker 独占这个任务多久”。如果 Worker 没有在租约到期前提交结果平台就自动把任务释放回 OPEN 状态并且累加重试次数。-- 超时释放把僵死任务重新放回任务池 UPDATE task_board SET status OPEN, worker_agent_id NULL, retry_count retry_count 1, updated_at NOW() WHERE status CLAIMED AND lease_expire_at NOW() AND retry_count 3;这里有几个细节值得强调租约时间要动态调整。任务复杂度不同执行时间差异很大。固定 10 分钟会导致长任务频繁被误释放短任务又可能在租约到期前无法完成。更稳妥的方式是让 Worker 定期上报心跳并续约。重试次数要设上限。如果任务连续失败三次以上自动回到人工队列不要无限重试。无限重试在 Agent 场景下会带来不可控的成本。任务状态变化必须记审计日志。谁在什么时间把任务从什么状态改成了什么状态这些记录是排查“状态到底为什么错”的唯一线索。状态机的定义是这次复盘中最有价值的产出。我们最终把任务生命周期收敛为OPEN → CLAIMED → IN_PROGRESS → SUBMITTED → REVIEWING → COMPLETED以及各阶段可以跳转的 FAILED 状态。没有设定中间态的任务一旦进入异常流程根本无法恢复。6. 执行边界与结果校验Guardrails 是如何加上的当 Worker Agent 开始真正执行任务后新的问题出现了输出结果不可信。人类交付工作成果至少有基本的责任意识和沟通能力。Agent 没有。它可能在超时后毫无响应可能在结果里返回一个不存在的 URL也可能提交一个格式完全不符合要求的 JSON。这些问题不是个例而是 Agent 执行类系统必然面对的不确定性。我们为执行环节加了三道 Guardrails。第一道是强制超时。任何对 Worker Agent 的调用都不能无限等待。我们用 asyncio.wait_for 封装执行过程超时后直接判定失败并触发任务释放逻辑。# 文件路径app/services/executor.py import asyncio from typing import List, Optional from pydantic import BaseModel, Field class TaskResult(BaseModel): output: str artifacts: List[str] Field(default_factorylist) evidence_url: str Field(default, max_length1024) class TaskExecutionError(Exception): Agent 执行超时或运行时错误 class TaskResultInvalid(Exception): Agent 返回结果不满足平台 Schema async def execute_with_timeout(task, timeout_seconds: int 60): 对 Worker Agent 的执行调用强制加超时。 避免一个异常 Agent 拖垮整个平台。 try: raw await asyncio.wait_for(task.execute(), timeouttimeout_seconds) except asyncio.TimeoutError: raise TaskExecutionError( fagent_execution_timeout:{task.agent_id} ) # 第二道 Guardrail结果结构校验 try: result TaskResult(**raw) except Exception as exc: raise TaskResultInvalid(fagent_result_schema_invalid: {exc}) # 第三道 Guardrail结果可信度校验 if not is_safe_url(result.evidence_url): raise TaskResultInvalid(agent_evidence_url_unsafe) return result第二道是结果结构校验。我们为每一类任务定义了结果 SchemaWorker 提交的交付物必须通过 Pydantic 校验。字段缺失、类型错误、URL 格式非法全部在平台侧拦截不让脏数据进入下一个环节。第三道是安全校验。这里尤其需要注意Agent 返回的链接、路径、代码片段本质上都可能是恶意输入。平台不能信任任何来自 Agent 的 URL 或文件路径。校验函数 is_safe_url 会检查协议白名单、域名解析结果和潜在的命令注入特征。from urllib.parse import urlparse ALLOWED_SCHEMES {http, https} BLOCKED_DOMAINS {internal.local, metadata.google.internal} def is_safe_url(url: str) - bool: 校验 Agent 提交的 URL 是否在允许范围内 if not url: return False try: parsed urlparse(url) except ValueError: return False if parsed.scheme not in ALLOWED_SCHEMES: return False if parsed.hostname in BLOCKED_DOMAINS: return False # 防止内网地址穿越生产环境应使用成熟的 SSRF 防护库 return True这三道 Guardrails 解决的是“不可信执行者”的问题。现在系统的核心原则非常清晰把 Agent 当成一个能力很强但完全不可信的远程调用者。所有输入做校验所有输出做校验所有外部资源访问做白名单。7. “人类雇主”和“AI 雇主”的差异工程假设的重建复盘到这里可以把核心差异抽象出来。传统 job board 的所有设计都隐含了一个默认假设参与者是负责任的成年人。这个假设在 Agent 之间是不成立的。维度人类雇主AI 雇主身份稳定性手机号、邮箱、企业认证相对稳定Agent 可能随时换环境、换密钥、被注销输入规范性大多理解 JD 怎么写尊重表单字段可能提交超大文本、混合格式或冲突字段并发行为很少恶意并发请求重试机制可能造成批量重复请求执行失败反馈会主动沟通延期或问题可能直接无响应状态卡死输出可信度有社会关系约束通常不会乱写可能返回伪造链接、垃圾结果甚至恶意内容责任主体法律上的法人或自然人现阶段没有明确责任主体这张表背后的工程含义是你必须在系统层面把“默认信任”改成“默认不信任”。不是某个模块需要这样而是每个模块都需要这样。认证模块默认所有调用都可能伪造所以需要签名和时间窗口。解析模块默认所有任务都可能格式混乱所以需要 Schema 强校验。状态模块默认所有 Worker 都可能中途失联所以需要租约和超时释放。执行模块默认所有结果都可能造假所以需要结构校验和来源校验。这也解释了为什么很多 Agent 项目在 demo 阶段表现惊艳、一上线就崩。Demo 环境里调用方是固定脚本参数是精心构造的真实环境里调用方是多个 Agent 的随机组合任何一环没有约束都会引发连锁故障。8. 生产环境最佳实践如果从头再做一次基于这次故障复盘如果把项目重做一遍下面这些实践会从一开始就纳入设计。第一环境隔离。开发、测试、生产必须完全隔离尤其是 Agent 之间的调用。建议搭建一个沙盒版任务市场用模拟 Agent 压测认证、状态竞争和结果校验逻辑不要在开发环境直接连生产的 Agent 集群。任何涉及结算和下发的操作都必须先在沙盒环境跑通。第二审计日志是硬需求。每个 Agent 的每次请求、每个任务的每次状态变更都必须记录原始请求体、响应体、调用方、时间戳。排查 Agent 问题时通常只能靠日志还原现场。没有完整审计Agent 系统的故障会变成无头悬案。第三最小权限原则。一个 Agent 只给它完成本职工作所需的最小权限。发布任务的雇主 Agent不需要调用 Worker 执行接口执行任务的 Worker Agent也不应该拥有审核其他 Agent 结果的权限。API Key 要按 Agent 维度独立发放禁止多个 Agent 共用一个密钥。第四限流和预算控制不能省。Agent 的重试机制非常容易触发流量放大。平台需要在网关层对每个 Agent 设置请求速率上限在业务层对每个任务设置执行预算上限。一旦超过预算任务立即进入人工审核而不是继续消耗算力和资金。第五涉及到钱的操作必须人工复核。虽然雇主是 AI但钱仍然是真金白银。支付、结算、退款这类动作至少在初期强制走人工审核。不要相信 Agent 之间的“自动结算”在没有成熟仲裁机制之前这是风险最大的环节。可以用一张操作清单来收尾这段[ ] 每个 Agent 独立密钥最小权限授权[ ] 任务认领使用条件更新 租约机制[ ] 所有任务输出经过 Schema 校验[ ] 所有外部 URL 经过安全校验[ ] 所有状态变更写入审计日志[ ] 网关层限流业务层控制预算[ ] 结算操作人工复核[ ] 准备回滚方案不追求一次成功9. 总结Agent 协作的瓶颈不在模型而在确定性这个项目最终没有解决所有问题但它让我们看清了一个事实Agent 协作的最大瓶颈不在单体模型能力而在多实体交互时的确定性。谁先把确定性做好谁就能接住 Agent 经济的第一波工作量。回到标题里的问题——what broke。坏掉的从来不是某个大模型而是那些默认“对方是人类”的工程假设。当你把雇主换成 Agent你需要重新设计认证、重新定义任务、重新约束状态、重新校验结果甚至重新思考责任边界。如果你接下来想做一个 Agent 任务市场或 Job Board我的建议是不要从模型开始而是从状态机开始。先把任务的生命周期定义清楚再考虑用哪个模型来解析任务、用哪个 Agent 来执行任务。状态机稳了架构就稳了一半。建议收藏备用。下次遇到 Agent 系统“跑不通”的诡异问题不妨先按这个顺序排查认证过了吗任务结构合法吗状态更新是不是原子的结果有没有被校验过大概率你要找的问题就藏在这四件事里。