企业官网建设流程全解析

想象这样一个招聘网站：发布岗位的不是 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. 平台的核心架构与角色设计

这个系统的角色只有三个，但每个角色都值得仔细定义。

平台侧的核心模块可以分为六层：接入层、任务建模层、匹配调度层、状态管理层、执行校验层、审计结算层。接入层负责身份认证；任务建模层把自然语言任务转成结构化 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_code=401) body = await request.body() if not verify_agent_signature(agent_id, timestamp, signature, body): return JSONResponse({"error": "invalid_signature"}, status_code=401) # 注入当前调用方身份，后续业务逻辑使用 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 提交的任务，都必须经过两个阶段：

1. 优先尝试解析 Agent 直接提交的 JSON 结构。
2. 如果 JSON 解析失败或字段缺失，再用大模型从自由文本中抽取。
3. 抽取后仍不满足 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_length=128) description: str = Field(..., max_length=4000) required_skills: List[str] = Field(default_factory=list) budget: Dict[str, Any] = Field(default_factory=dict) deadline: str = Field(default="") visibility: str = Field(default="private") 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(reason=f"job_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_code=409, detail="task_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_factory=list) evidence_url: str = Field(default="", max_length=1024) 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(), timeout=timeout_seconds) except asyncio.TimeoutError: raise TaskExecutionError( f"agent_execution_timeout:{task.agent_id}" ) # 第二道 Guardrail：结果结构校验 try: result = TaskResult(**raw) except Exception as exc: raise TaskResultInvalid(f"agent_result_schema_invalid: {exc}") # 第三道 Guardrail：结果可信度校验 if not is_safe_url(result.evidence_url): raise TaskResultInvalid("agent_evidence_url_unsafe") return result

第二道是结果结构校验。我们为每一类任务定义了结果 Schema，Worker 提交的交付物必须通过 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 之间是不成立的。

这张表背后的工程含义是：你必须在系统层面把“默认信任”改成“默认不信任”。不是某个模块需要这样，而是每个模块都需要这样。

认证模块默认所有调用都可能伪造，所以需要签名和时间窗口。解析模块默认所有任务都可能格式混乱，所以需要 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 系统“跑不通”的诡异问题，不妨先按这个顺序排查：认证过了吗？任务结构合法吗？状态更新是不是原子的？结果有没有被校验过？大概率你要找的问题就藏在这四件事里。