企业官网建设流程全解析

做AI应用这一年多，我最大的体会不是模型进步太快，而是“底层执行链先崩了”。去年我们把AI客服灰度上线，顺手打通了订单侧的数据链路，结果有一天凌晨三点，任务在没有人工确认的情况下被连续执行了几十次，内部代号AX。产品在群里问“为什么一条指令会被派生这么多次”，我盯着日志半天说不出话。事后复盘，我们缺的既不是prompt也不是模型能力，而是一层能把“规划、执行、复核、兜底”全部串起来的执行基础设施。这个东西，我们内部叫AX调度，现在团队和几个做Agent的朋友群里，“ax调度”几乎成了Agent Execution调度的通用缩写，谁家要是没个调度层，开会都不好意思提。

这篇文章不聊大模型原理，就聊怎么把模型产出的“意图”变成可控、可观测、可回滚的任务流。内容包括ax调度的整体设计、状态机、超时与幂等、并发控制、一段能直接改的Python执行器骨架，以及我调生产问题时遇到的高频坑。适合正在做多Agent编排、工具调用、或者准备把Agent接入业务系统的朋友，尤其是那些已经被“工具重复执行”“任务卡死”“上下文丢失”折磨过的人。

1. 起因：为什么我会自己动手写一套ax调度

先说一个容易被忽视的事实：大模型本身不是不可控的，真正不可控的是“模型调用工具之后，工具到底怎么执行”。LLM输出一段JSON，里面写“调用库存服务，扣减10件”，这一步只是生成了一句话。从这句话到数据库里的UPDATE真正生效，中间隔着HTTP超时、权限校验、并发冲突、重复请求。任何一个环节出问题，后果都直接落在真实业务上。

我当时面临的问题主要有三个。第一个是工具失控。模型有时候会重复生成同一动作，比如用户说“帮我把订单状态改成已退款”，模型可能规划了两次“修改订单状态”的动作，结果退款被提交两次。这不是模型故意使坏，而是多轮上下文里它已经忘了上一步做过什么。第二个是执行时间不可控。模型思考很快，但工具调用链路很长，如果同步等待整条链跑完再返回，用户早就不耐烦了。第三个是流程不可观测。一旦出问题，日志是散的，排查靠猜，谁也不知道当前Agent到底执行到哪一步、还有几个子动作没做。

这三个问题指向同一个答案：不能把模型的“决策”直接当成“执行”，必须把执行动作从主流程里拆出去，交给一个专门组件来管。这个组件就是ax调度。它的职责可以概括成一句话：模型只负责决定“做什么”，ax调度负责决定“什么时候做、怎么做、做失败怎么办”。

1.1 AI工具调用真正的问题不是“模型笨”，而是“没人管”

我见过团队在模型层面较劲，反复调temperature、调few-shot示例，就为了让模型不要重复调用工具。说实话，效果有限。因为重复调用很多时候不是模型推理错，而是执行层没有做幂等。模型生成两次相同参数的工具调用，在它的“认知”里是两次独立的尝试，在业务系统里却是两次扣款。

这种时候，靠prompt是压不住的。你得有一个执行层能在运行时识别“这个工具调用和上一次是同一个语义意图”，直接把第二次请求拦下来，或者返回上一次的结果。这就是幂等控制，属于调度层的基本功，而不是模型能力问题。

还有一个更微妙的点：模型调用工具是“大概率正确”，也就是它有可能调用一个并不存在的工具，或者传一个格式错误的参数。如果主流程是同步的，只要有一环出错，整个请求就失败，用户只能重新发起。更麻烦的是，某些工具有副作用，失败了但系统部分状态已经改掉了，你又不能简单重发。这些都需要调度层通过状态、重试、补偿机制来处理，不是一个try-catch能解决的。

1.2 ax调度的定位和普通任务调度有什么区别

不少人第一次听到“ax调度”会想：这不是咱公司那套定时任务平台能干的事吗？确实像，但细看差很多。普通的任务调度，比如crontab或者xxl-job这类，核心模型是“时间触发、执行任务、失败重试”。任务是预先定义好的，参数是确定的，执行完就结束，不太需要考虑上下文。

ax调度不一样。它面对的任务是模型动态生成的，任务之间还有先后依赖关系。一个根任务可以拆出多个子任务，某些子任务必须等前一个成功之后才能执行，某些子任务可以并行。子任务失败后，不是简单重跑，而是可能要根据失败原因决定是重试、跳过、还是回滚整个根任务。

我用两张表对比一下普通任务调度和ax调度的差异。

我常说一句话：普通调度的核心是“准时”，ax调度的核心是“靠谱”。前者关心任务有没有在正确时间跑，后者关心任务跑完之后系统是不是还在一个确定的状态里。

2. 整体设计逻辑：把一次Agent执行变成可推进的流程

我们设计ax调度时，最先讨论的不是技术选型，而是“一次Agent执行到底应该被抽象成什么”。最后定下来的模型非常朴素：一次执行就是一棵树。

树的根节点是用户发起的原始意图，比如“帮我调一下A仓库和B仓库的库存差异”。模型接收到这个意图后，可能拆出几个子节点：“查询A仓库存”“查询B仓库存”“对比计算”“生成调拨单”“推送审核”。其中前三个可以并行，后面两个必须按顺序执行。每一个节点，我们称之为一个执行单元，对应一次具体的工具调用。

所有执行单元都挂在同一个执行上下文里。这个上下文记录了父任务是谁、子任务有哪些、当前节点状态、执行输入输出、重试次数、取消标志。说白了，就是把一个Agent的“脑子”里想过的事，完整落成一张可查询的表。这样用户问“刚才那个任务到底跑完没有”，你不需要看日志，直接查这张表就行。

2.1 核心抽象：根任务、子任务与执行树

在ax调度的数据模型里，一张任务表是核心。字段粗略包括：

• 任务ID：全局唯一ID，用UUID或者雪花ID都行。
• 根任务ID：标记这棵树从哪来。
• 父任务ID：标记当前节点属于谁。
• 任务类型：对应具体工具名。
• 任务状态：见下面的状态机。
• 输入参数：模型生成的JSON参数。
• 执行结果：成功后的返回值，或者失败后的错误码。
• 重试次数：当前已经重试了几次，最大允许几次。
• 创建时间、执行时间、完成时间。

这张表不仅是存储，还是整个系统的“真相来源”。调度器推进状态时读它，执行器写它，人工后台也看它。所有操作都以任务ID为线索，能把一棵树的执行历史完整捞出来。

有了执行树这个抽象，很多麻烦事就变得清晰了。比如取消操作：用户说“等一下，我不要调拨了”，那就不需要遍历代码去停进程，只需要把根任务状态置为CANCELLED，所有还在排队、尚未执行的子任务看到这个标志后直接跳过。已经在执行的子任务，通过协程或上下文取消信号来终止，具体机制下面会讲。

2.2 状态机怎么设计，才不会跑飞

状态机是整个ax调度里我最重视的部分，因为状态一旦乱了，整个系统就变成“薛定谔的执行中”。我的经验是，状态不能太少，也不能太多。太少了区分不了“超时”和“失败”，太多了团队自己都记不住。

我们最终使用的状态集如下：

这几个状态之间有一条铁律：只有RUNNING状态下的执行器才允许产生业务副作用。其他状态都只是计划或结果。这点很重要，如果没有这条约束，调度器可能会在RETRY_WAIT状态下重复下发，导致业务重复执行。

另外，我特别加了一个DEAD状态。很多人不喜欢死信，觉得意味着系统失败。但实际上，在AI工具调用场景里，有些错误是你预设的重试规则处理不了的，比如参数完全不符合业务规则。这种情况下最危险的不是报错，而是“静默失败”或者“盲目重试”。所以我宁可把这类任务丢进DEAD，让运营人员人工看一眼，也不想让它自己无脑跑到天荒地老。

2.3 为什么用异步事件驱动而不是同步顺序执行

最开始我们做过一版同步编排，逻辑很简单：Agent调调度器，调度器直接调工具，拿到结果再让模型判断下一步。这套方案在demo里跑得很流畅，一上生产就傻了。原因倒不复杂：同步方式下，整个链路的总计时间等于模型思考时间加所有工具调用时间相加。模型思考3秒，工具调用2秒，再来几个分支，用户那边就是长达十几秒的无响应白屏。

更致命的是，同步模式下任何一环超时，整个请求就挂在那边，既不能取消，也不能重试中间断点。有一次某个下游接口slow了，客户端直接断开，调度器这边还傻傻等着返回值，整个执行链就悬在半空。

后来我们改成异步事件驱动。核心思想是：调度器只负责把任务推入队列并更新状态，执行器从队列里取任务、执行、把结果作为事件发回。调度器收到事件后，再决定是推进下一个子任务、重试当前节点，还是标记整个根任务失败。这样一个模型会话的“思考”和“执行”不再强绑定，用户体验提升一大截。

异步化带来的额外好处是天然削峰。比如早上九点大量用户提问，每个问题拆出三五个子任务，队列会自然堆积，执行器按自己的消费速度慢慢跑。不会因为瞬时并发太高直接把下游系统打挂。

3. 关键细节：超时、幂等、锁与单飞

这一部分是ax调度能不能在生产环境活下去的分水岭。前两章讲的是“架子”，这章讲的是“里子”。

3.1 超时到底设多少，才算既有体验又防失控

给工具调用设超时，是所有人都知道要做的事，但设多少合适，很多人拍脑袋。我的建议是分三层来看。

第一层是模型调用超时。模型不参与工具执行，但参与工具链的下一步规划。这一层我一般留30秒到60秒，因为模型偶尔会在长上下文下思考很久。第二层是单次工具调用超时。如果是内部API，一般10秒到15秒足够；如果是外部第三方接口，我建议不要超过20秒，因为外部依赖更容易拖尾。第三层是根任务整体超时。这个取决于业务场景，比如库存同步这种重操作，我设30分钟；如果是查天气这种轻量查询，2分钟就顶天了。

这里有个容易踩坑的点：只在HTTP客户端加超时是不够的。如果工具内部发起了一条数据库查询或者调用另一个内部服务，HTTP层超时只能断掉当前连接的等待，但下游可能还在继续执行。更安全的方式是使用带超时传递的调用链，比如在Python里用asyncio.timeout包住整个业务逻辑，同时要求下游数据库或服务自身也设置语句超时。

给个具体数字参考吧，最近我在一个Go项目里习惯用context.WithTimeout，DB层设置statement_timeout=5s，外部HTTP调用timeout=15s，模型服务timeout=45s。这样每一层都有限制，不会出现“HTTP超时了，但数据库还在跑”的尴尬局面。

3.2 幂等性不是“加个唯一索引”就完事

先讲一个真实事故。我们用唯一索引防重复，给任务表加了“请求ID”唯一键，以为万事大吉。结果线上还是出现了两次扣款，排查到最后发现，扣款服务是在事务提交之前先发了消息通知，而通知里的业务操作没走幂等判断，等于同一个请求在系统里走了两条不同的通道。

在ax调度里，幂等性至少要做好三张表。

第一张是请求ID表。每次模型决定调用一个工具时，调度器就生成一个全局唯一的request_id，比如toolname-uuid。这个ID会在整个调用链中透传。

第二张是幂等键表。request_id是唯一键，表的字段包括：幂等键、工具名、参数哈希、状态、首次执行时间、最后执行时间、返回结果。执行器在真正调用工具之前，先往这张表写一条“PROCESSING”的记录；工具调用成功后，把结果写进这条记录并置为“SUCCESS”；工具调用失败，置为“FAILED”。如果同一个request_id再次闯进来，直接查这张表，状态是SUCCESS就返回老结果，状态是FAILED就按失败逻辑处理。

第三张是变更审计表。每次工具执行产生的业务变更、状态流转、返回值，都追加记录。这张表不是直接用于防止重复，而是让事后排查有据可查。特别是模型可能出现幻觉，你总要知道上一次它到底是拿什么参数去调用工具的。

核心原则是：幂等控制必须发生在真正产生业务副作用之前的“最后一道门”，而不是在入口处草草判断一下就觉得安全了。

3.3 并发锁与请求合并：同一会话不允许重复执行链

有了幂等表，能挡住“同一次重试”造成的重复。但还有另一种情况：用户刷新页面或连点两次按钮，同一个Agent会话被创建了两个执行链，它们可能拿着不同的request_id，但业务意图是同一个“改订单地址”。这时候幂等表就拦不住了，因为参数可能一样但request_id不同。

解决这个问题的办法是会话级锁。我习惯在Redis里建一把锁，键名为agent_lock:{session_id}，值存当前根任务ID，加过期时间，并带上一个随机token。当新任务准备执行时，先尝试加锁。如果锁不存在，说明当前没有执行链，新任务获得执行权；如果锁已存在，后到的任务直接返回“当前会话已有任务在处理中”，而不是排队等待。

可能有人问：为什么不直接让后到的任务排队，等前一个跑完再跑？我之前也这么设计过，上线后发现体验很怪：用户明明在追问一个新问题，系统却在排队等上一个任务跑完，给人一种“机器人卡住了”的感觉。后来改成“后到的任务直接拒绝并提示重试”，反而更符合用户预期。

加锁之后还要处理锁过期问题。模型工具链条一旦很长，执行时间可能超过锁的过期时间，锁自动消失，新的执行链趁虚而入。我的做法是给锁加一个看门狗协程，在任务运行期间自动续期，续期到根任务最大执行时间后结束。另外锁的值要带一个随机token，释放锁时校验token，避免误删别人的锁。

4. 实操：从队列到执行器的最小闭环

讲完设计，来点能直接落地的。下面的方案我用了很久，组件都是常见的东西，不挑框架。

4.1 最小闭环需要哪些零件

一个能跑起来的ax调度，至少需要五个零件：任务存储、消息队列、调度器、执行器、管理后台。

任务存储我用MySQL，因为事务支持好，任务状态流转需要强的数据一致性。消息队列我用Redis Stream，它比普通List更适合这个场景，因为支持消费者组和Ack机制。执行器是独立的Python服务，也可以是多语言混合的，这取决于工具生态。调度器是一个比较轻的服务，它负责把根任务拆成子任务、把子任务推入队列、监听队列结果事件、更新任务状态。管理后台可以先用一套简单的Web界面，用来查任务状态、手动取消、重新执行死信。

我的建议是千万别一上来就上Kafka加多个微服务，复杂度会把你的精力耗尽。先用一个进程把所有逻辑跑通，等任务量真的上来了再拆分也不迟。

4.2 一个可复现的执行器代码骨架

这是我最想分享的部分。执行器的核心逻辑不复杂，难的是把“超时、重试、心跳、结果上报”这些边界问题处理干净。我用Python写了一个最小骨架，读起来更直观。

# ax_worker.py import asyncio import json import time import uuid import redis.asyncio as redis STATUS_WAITING = "WAITING" STATUS_RUNNING = "RUNNING" STATUS_SUCCESS = "SUCCESS" STATUS_FAILED = "FAILED" STATUS_RETRY_WAIT = "RETRY_WAIT" class AXWorker: def __init__(self, stream_name="ax:queue"): self.redis_client = redis.from_url("redis://localhost:6379") self.stream_name = stream_name async def heartbeat(self, task_id: str): # 每5秒上报一次心跳，避免调度器误判为卡死 while True: await self.redis_client.hset( f"ax:heartbeat:{task_id}", "last_beat", time.time() ) await asyncio.sleep(5) async def process_task(self, task): task_id = task["task_id"] tool_name = task["tool_name"] payload = json.loads(task["payload"]) await self.redis_client.hset( f"ax:task:{task_id}", "status", STATUS_RUNNING ) # 启动心跳 hb_task = asyncio.create_task(self.heartbeat(task_id)) try: # 这里替换成真实的工具调用，注意超时 result = await asyncio.wait_for( self.call_tool(tool_name, payload), timeout=15 ) await self.redis_client.hset( f"ax:task:{task_id}", "status", STATUS_SUCCESS ) await self.redis_client.hset( f"ax:task:{task_id}", "result", json.dumps(result) ) # 上报结果事件，让调度器推进下一个节点 await self.redis_client.xadd( "ax:result", { "task_id": task_id, "status": STATUS_SUCCESS, "result": json.dumps(result), } ) return True except asyncio.TimeoutError: await self.redis_client.hset( f"ax:task:{task_id}", "status", STATUS_RETRY_WAIT ) await self.redis_client.xadd( "ax:result", { "task_id": task_id, "status": STATUS_RETRY_WAIT, "error": "timeout", } ) return False except Exception as exc: await self.redis_client.hset( f"ax:task:{task_id}", "status", STATUS_FAILED ) await self.redis_client.xadd( "ax:result", { "task_id": task_id, "status": STATUS_FAILED, "error": str(exc), } ) return False finally: hb_task.cancel() async def call_tool(self, tool_name: str, payload: dict): # 工具注册表，按需扩展 if tool_name == "order_status": # 模拟一次内部接口调用 await asyncio.sleep(2) return {"order_id": payload["order_id"], "status": "refunded"} raise ValueError(f"tool {tool_name} not found") async def consume(self): # 从Redis Stream读取任务 while True: res = await self.redis_client.xreadgroup( group="ax_workers", consumer=f"worker-{uuid.uuid4().hex[:6]}", streams={self.stream_name: ">"}, count=1, block=5000, ) if not res: continue for stream, messages in res: for msg_id, msg in messages: task = msg ok = await self.process_task(task) if ok: await self.redis_client.xack( self.stream_name, "ax_workers", msg_id ) else: # 失败的不ACK，让它进入Pending列表 pass

这个骨架里有一个容易被忽略的细节：失败的任务不ACK，让它留在Redis Stream的Pending队列里。这样调度器可以通过排查Pending列表找出那些“执行了但没成功确认”的任务。有些团队在这里图省事直接ACK然后重新入队，处理不好就会导致任务丢失。

再说一下为什么不直接用redis里的BLPOP这种简单列表。因为Redis Stream的消费者组支持Ack机制，任务被某个worker取走后，调度器能明确知道它有没有处理完。普通List拿一个消息就少一个，没有任何确认机制，万一worker处理到一半挂了，这条消息就彻底丢了，这在业务工具调用场景是不可接受的。

4.3 如何接进LangChain这类Agent框架

很多人的Agent已经基于LangChain或LlamaIndex这类框架开发了。要接入ax调度，不需要重写框架，只需要把工具调用那一层替换成异步下发。

简单说，就是在原来工具函数内部，不再直接执行业务代码，而是创建一个调度任务，把参数发给ax，返回一个task_id。框架继续往下走的时候，通过这个task_id去查状态。原来的同步返回值变成“轮询状态后的最终结果”。

我给你一个变形示意，原来的工具函数可能是这样：

def check_stock(warehouse_id: str): return stock_service.check(warehouse_id)

接入ax调度后变成：

def check_stock(warehouse_id: str): task_id = ax_scheduler.submit( tool_name="check_stock", payload={"warehouse_id": warehouse_id} ) return ax_scheduler.wait_for_result(task_id, timeout=30)

这段代码看起来变化不大，但背后本质变了：业务不再直接跟stock_service打交道，而是经过调度层。调度层可以做超时、重试、幂等、并发控制，而业务代码保持简单。

如果你是纯用LangChain的AgentExecutor，直接改写Tool的_run方法就好了。特别提醒一点：改造之后，工具调用耗时可能会有一定增加，因为多了一层队列和状态查询。对于耗时敏感的场景，不要把wait_for_result的等待时间设得太短，否则Agent会误判工具失败。

5. 问题清单：调试ax调度遇到的5个真实坑

这部分是我觉得全文最有价值的地方。以下问题全部来自生产环境，每一个都让我加过班。

5.1 任务一直停在RUNNING，积压不消费

有一天我看到任务表里几十个任务都停在RUNNING，但队列里明明没有积压。排查后确认是某个worker进程卡死了，但没有发心跳。当时的执行器没有做心跳上报，调度器只能看到任务一直没完成，却不知道worker是不是还活着。

后来加上心跳机制，调度器定期检查ax:heartbeat:{task_id}这个键。如果某个worker过期超过30秒没心跳，就把任务重新投回队列，并由调度器发告警。这里注意，不能简单把任务重新投回就完事，因为原本的worker可能还没死透，只是卡在某个慢调用上。如果重新投递的worker和旧worker同时执行，就会造成重复。所以重新投递之前，必须通过幂等表确认旧worker没有真正产生业务变更。

5.2 工具被调了两次，幂等没拦住的怪异现场

这是最让我头疼的一个问题。幂等表明明加了唯一索引，结果还是出现了重复执行。排查后定位到原因是：第一次请求已经写入幂等表但没有提交事务，第二次请求看到的是旧快照，判断“这个键不存在”，于是又插入了一条。最终两条都提交了。

解决办法是使用“条件插入 + 检查影响行数”的方式。先执行INSERT ... ON DUPLICATE KEY UPDATE，如果影响行数为0，说明这条幂等记录已经存在，就不再执行工具。如果涉及事务隔离级别问题，可以考虑把幂等表查询改成SELECT ... FOR UPDATE，但这对性能有影响。更务实的方案是：把幂等记录的写入和真正业务操作的执行放在同一个事务里，确保“先占坑、再执行、同生共死”。

5.3 模型幻觉式传参，调度层应该怎么办

LLM不是百分百按schema出参的，它可能传一个不存在的仓库编号，或者把单位“件”写成“箱”。如果工具直接执行，轻则返回错误，重则写错业务数据。在ax调度里，我在执行器前面加了一层“参数校验网关”。每个工具注册时附带一份JSON Schema，执行器拿到参数后先做schema校验，不合格的不会调用真实工具，而是把任务置为“QUARANTINE”状态，相当于一个等待人工处理的特殊状态池。

针对那些有破坏性的工具，比如批量删除、修改价格、退款，我额外加了“二次确认”规则。模型生成的调用会先进到确认队列，由运营或用户点确认后才真正执行。这个机制看起来“慢”了一点，但在真实业务里能救大命。

5.4 任务风暴与队列积压的常见原因

某个大促期间，我们突然发现队列积压了上万条任务，最夸张的一次延迟了20分钟。看了下罪魁祸首，是某条根任务拆子任务时写了一个“死循环”：模型没有判停，反复调用同一个查询工具。一个用户会话就能生成几十个工具调用，几十个用户就能把队列打满。

要防这种问题，只靠调度器限流是不够的。我给每个根任务设置了子任务数量上限，比如最多50个子任务；还给每个模型的“决策轮次”设置了上限。一旦超出，调度器会强制终止这条执行链，并且把根任务标记为DEAD。另一招是给执行器设置最大消费速率，比如每秒最多消费20个任务，防止下游系统被打爆。

5.5 上下文用完就丢，重试后恢复不了

最后说一个比较隐蔽的坑。一个根任务里如果有个子任务失败了，我们常规操作是重试这个子任务。但重试时需要重新组装模型上下文，而原上下文里包含了前面几个子任务的执行结果。如果这些结果没有被持久化，重试的时候就拿不到必要的上下文，模型只能瞎猜。

我的解决方案是把每个子任务的输入输出都持久化到任务表或者专门的trace表，重试时直接从数据库读回之前的执行结果，而不是指望内存里还留着。这个改动简单，但对重试成功率提升非常明显。如果团队用的是内存队列，一定得注意worker重启后任务状态还在不在，不然一重启，所有上下文都没了。

说实话，ax调度这层我踩过最大的坑，不是锁写得不严谨，也不是状态机设计失误，而是“半吊子调度”。任务发出去没人管，看着很轻量，一出事就得跪着道歉。后来我把每条执行链都当成一条要收口的线段来看待，必须有状态、有心跳、有兜底，才算真正把AI工具调用放进了生产环境。

我的个人体会是：别太指望模型提供商未来会在SDK里帮你把调度、幂等、重试都内置好。模型迭代速度再快，调度层的“可靠”依然是要自己动手垒的。调用一次模型很便宜，出一次数据事故很贵，这笔账越早算明白越好。希望这篇ax调度的拆解能帮正在做Agent工程化的你少走点弯路。