ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

从代码问答到任务执行:羲和Agent的架构设计与工程实践

从代码问答到任务执行:羲和Agent的架构设计与工程实践 1. 为什么多数AI编码助手止步于问答先讲个真实场景。上个月我给团队搭了一套代码知识库问答机器人效果相当能打仓库里几千个文件问订单超时重试的逻辑在哪个模块、支付回调幂等怎么做的模型能把文件路径、方法名、关键代码片段都给你翻出来。团队小伙伴用得挺开心直到有人问了一句那你能不能直接帮我把这个定时任务跑一下看看是不是真的有问题我愣住了。问答可以动手不行。这个感受不是个例。过去一年我用过不少号称AI编码助手的工具大多数本质上是增强版的代码搜索加代码生成。你问它问题它给你答案你让它改代码它给你 diff但你要它去把测试跑一遍、把任务调度起来、把构建结果拉回来看看它就无能为力了。原因倒也不复杂问答链路只需要理解并生成文本而任务执行链路需要理解、决策、调用外部系统、感知结果、再决策这条链路一旦走起来要补的工程细节就多出一个量级。我设计羲和XiheAgent的初衷就是想把它从一个很懂代码的聊天机器人往前推一步变成一个能自己动手干活的编码助手。这里的干活不是指自动补全几行代码而是指它能够理解你的诉求把诉求拆解成具体动作调用真实的工具去执行这些动作然后根据执行结果决定下一步怎么做。代码问答是它的基本功任务执行才是它真正值钱的地方。这篇文章就围绕这个设计展开。我会先把问答和执行之间的鸿沟讲清楚然后拆解羲和的整体架构再重点讲两个核心跨越——工具调用协议的落地和执行任务时的调度与告警链路最后聊聊我在真实环境中踩过的坑。如果你也在做类似的Agent或者想把现有的AI编码助手往能干活的方向推一把这篇应该能给你一些直接能用的思路。2. 鸿沟到底在哪里问答与执行的三层差异很多人觉得让AI从回答代码问题到执行任务不就是在后面加一个调用函数的步骤吗真不是。我把它拆成三层每一层都是一道坎。2.1 第一层上下文从静态快照变成动态状态问答模式下模型看到的是仓库的快照——代码文件、文档、索引。这些东西在回答过程中不会变。你可以把整个仓库读进上下文里慢慢分析模型不需要关心现在系统处于什么状态反正它只是解释代码。执行任务就不一样了。比如让羲和把这个定时任务跑起来然后告诉我结果它面对的是一个动态系统任务现在是什么状态是等待中、运行中还是上次跑失败了运行环境里有没有权限任务跑的这台机器资源够不够这些信息模型在静态代码里看不到必须实时去问系统才能拿到。这意味着架构里不能只有知识库还得有状态感知这一层。2.2 第二层输出从文本变成动作问答链路里模型的输出是给人看的自然语言。你可以接受它有模糊表达比如这可能有问题、建议你检查一下XX。但执行链路里模型输出的是一系列明确的动作指令——调哪个接口、传什么参数、按什么顺序调。这里的容错率极低参数传错一个字符轻则任务跑不起来重则把生产环境搞出问题。所以你必须给模型设计一套严格的工具调用协议而不是让它自由发挥输出JSON片段然后硬解析。协议里要有明确的约束工具名、参数类型、必填项、取值范围、幂等性要求、超时时间。这些约束不是用来限制模型的而是用来保证模型随便一调系统也能给出明确反馈。2.3 第三层流程从一次生成变成多轮闭环问答基本是一锤子买卖问题进来答案出去结束。执行任务则是一个循环发起动作 → 等待结果 → 分析结果 → 决定下一个动作 → 再发起。模型不再是生成一句话而是驾驶一个流程。这个循环里最难的不是让模型想清楚下一步而是工程上怎么让这个循环稳定、可控地跑起来。包括动作执行结果怎么返回给模型失败时怎么重试重试多少次执行时间太长怎么处理模型在这个循环里会不会陷入死循环所有这些都要求你在模型外面架设一套执行引擎而不是把希望全部寄托在模型本身的推理能力上。我建羲和的时候把这三点当成顶层设计约束来对待。也因此初版架构就没有走一个模型包打天下的路线而是拆成了三层。3. 羲和XiheAgent的整体架构三层链路的取舍先看一张总览图这是我最终定下来的结构简化处理后是这样用户输入自然语言 / 代码片段 / 任务指令 ↓ 第1层 理解层意图识别 上下文检索代码问答的基础 ↓ 第2层 规划层任务拆解 工具选择 参数生成 ↓ 第3层 执行层工具网关 任务调度 结果回传 告警 ↓ 用户收到执行结果 / 代码建议 / 状态报告这里我要解释一下每层存在的必要性因为很多同类项目栽就栽在只有第3层或者只有第1层没有中间这个规划层。3.1 理解层代码问答的地基不能丢羲和的底座还是代码问答。原因很简单没有对代码库的深度理解任务执行就是盲人摸象。你让AI去跑一个定时任务它至少得先知道这个任务的代码逻辑是什么、依赖哪些配置、以往的执行记录长什么样这些信息都来自理解层。理解层我用了标准的RAG方案检索增强生成核心是这三件事代码索引把仓库里所有源文件解析成语义块chunk按函数、类、文件层级建立索引。这里我强烈建议按函数粒度切块而不是按固定token数量硬切。原因很实际模型回答问题时要引用具体函数粒度太粗容易把不相关的代码混进上下文粒度太细又会丢失函数间调用关系。混合检索关键词检索BM25加向量检索并行做然后做Rerank。单纯用向量检索在代码场景下不太够用因为代码里的符号、类名、方法名往往是精确匹配才有效比如搜retryStrategy这个字段名向量检索可能给你捞一堆语义相近但完全无关的东西。调用链上下文这个比较关键。检索到某个函数后我会额外把它的上游调用方和下游被调函数也带进上下文。问答这个任务为什么失败时模型能看到的不只是孤立函数而是整条调用链。这一层的产出是模型对代码库的认知快照是后续规划层的输入。3.2 规划层把去执行翻译成执行什么、怎么执行规划层是羲和区别于普通代码问答助手的核心。它的职责是把用户的模糊指令翻译成一组结构化、可执行、可验证的工具调用序列。举个例子。用户说帮我把订单回流任务重新跑一遍如果失败就通知我。规划层要做的事包括判断订单回流任务对应系统里的哪个任务要去配置中心或调度平台查而不是靠猜。判断重新跑一遍是触发一次新执行还是重跑上次失败的执行。判断失败通知我对应哪个渠道企业微信、邮件、短信并生成对应的通知参数。把这整串逻辑输出为一个行动计划交给执行层。这一层最难的其实是让模型学会在不确定的时候停下来问人而不是硬猜。我在规划提示词里明确写了三条规则任务名称匹配不到时必须给用户候选列表参数无法从上下文推断时必须向用户确认涉及删除、覆盖、回滚类操作时必须二次确认。这看起来很笨但实测下来能省掉大量事故。3.3 执行层让规划真正落地执行层是羲和的手它负责把规划层输出的工具调用序列真正执行掉。这一层要处理的问题特别工程化包括工具注册中心有哪些工具可用每个工具的入参出参是什么。执行网关统一处理鉴权、限流、超时、重试。结果回传执行结果要结构化回传给规划层供模型判断下一步。状态记录每次执行都要落库方便追溯和审计。执行层的设计直接决定了Agent闯祸的概率。我见过不少人做Agent时把执行层简化成了模型直接调API结果模型调用一个删除接口时把整套测试数据清空了。所以执行层一定要做能力边界控制——哪些工具AI可以自主调用哪些必须人工审批这个红线必须在执行层卡死。4. 工具协议设计让模型从给建议变成发指令的关键规划层说了要输出工具调用序列那模型输出的格式到底长什么样我前后迭代了三版协议踩了不少坑这里分享一个目前用着最顺手的版本。参考OpenAI的Function Calling和Anthropic的Tool Use的思路我给羲和定义了一套统一的工具调用协议核心是一个JSON-RPC风格的请求体{ tool: workflow.trigger, request_id: req_8f3a2b9c, params: { task_name: order_reconcile_daily, trigger_mode: manual, timeout_seconds: 3600 }, on_success: [ { tool: notification.send, params: { channel: wecom, message_template: task_{task_name}_success_at_{time} } } ], on_failure: [ { tool: notification.send, params: { channel: wecom, message_template: task_{task_name}_failed_reason_{error} } } ] }4.1 为什么协议里要有成功/失败回调分支第一版协议我写得很简单就是调什么工具、传什么参数。但实际跑下来发现一个严重问题模型在工具调用失败后不知道怎么处理。比如它让调度平台跑任务任务因为权限不足被拒绝正常情况下应该换一个高权限的token重试或者通知用户我没有权限做这件事但模型只会把错误信息原样返回给用户跟个报错复读机一样。后来我加了on_success和on_failure分支含义是模型在发起一个动作时就要把成功和失败两个分支都规划好。这逼着模型提前思考任务失败了怎么办而不是事后补救。实测下来加了回调分支后羲和在执行任务时一次跑通的比例明显提高因为很多低级错误在执行前就被模型自己预判到了。4.2 工具注册表每个工具都要有人话描述工具调用协议里有一个容易被低估的点模型要怎么知道有哪些工具可用、每个工具是干啥的我的做法是维护一张工具注册表每个工具包含以下字段字段说明示例name工具名必须全局唯一workflow.triggerdescription人话描述说明工具的用途、适用场景触发一个工作流任务执行适用于定时任务/ETL任务的手动触发、重跑parameters参数JSON Schema包含类型、必填、枚举、默认值task_name(string, required)timeout_seconds(int, default3600)permissions调用该工具所需的最小权限admin / developer / readonlyidempotent是否幂等能否安全重试true / falsetimeout默认超时时间30s这里我想特别强调description和idempotent两个字段。description写得越有人味模型选对工具的概率越高。比如你不写用于执行DolphinScheduler工作流而是写触发DolphinScheduler工作流执行用于手动重跑昨天的失败任务、补数据等场景注意该操作会真实启动一个分布式任务耗时可能较长模型就知道补数据这种诉求该用它而不是去用另一个只查询状态的工具。idempotent字段决定执行层能否自动重试。如果工具是幂等的比如发送通知失败后可以直接重试如果不幂等比如创建订单失败后绝不能盲目重试否则会造成重复数据。这一点在后面的真实事故里会讲到。4.3 模型输出校验不能信任模型的自由发挥模型不是严格按照协议输出的有时会漏参数有时会编一个不存在的工具名有时参数类型给错。所以工具协议解析后还必须要过一道严格校验包括三件事Schema校验参数是否合法、类型是否正确、必填项是否齐全。工具存在性校验工具名是否在注册表里。权限校验当前用户/会话是否有权调用该工具。这三道校验全部通过后执行层才会真正发起工具调用。不通过时会直接把错误信息反馈给规划层让模型去改而不是硬着头皮执行。这一步很关键因为LLM生成的JSON经常会有格式问题你在解析层多花一点功夫能省掉后面大量故障排查时间。5. 任务执行链路从DolphinScheduler调度到企业微信失败告警工具协议设计好之后真正能体现任务执行价值的是执行链路本身。这一章我拿一个典型场景来讲透用户让羲和去触发一个DolphinScheduler调度任务如果执行失败自动往企业微信群里发告警。这个场景非常真实数据开发的同学应该都懂。传统做法是你自己打开DolphinScheduler控制台找到工作流点击运行然后每隔几分钟刷一次日志。现在这个操作可以让羲和来干。5.1 场景需求拆解用户到底想要什么用户的原话可能是把昨天跑失败的XX日报数据任务重新跑一遍跑挂了记得在群里喊一声。这句话翻译成系统操作其实是四件事在DolphinScheduler里找到名为XX日报数据的工作流。查看它的最近一次执行记录确认它确实是失败状态。重新触发一次执行注意这里不能盲目触发得先确认工作流是否会做幂等处理否则重跑可能导致数据重复。实时监控这次执行的运行状态如果最终还是失败发送企业微信告警到指定群。你看自然语言一句话背后是查询—校验—触发—监控—告警五个环节每个环节都可能出幺蛾子。羲和要做的就是把这五个环节串起来做成一个稳定的执行链路。5.2 DolphinScheduler工具封装不只是调一个API我封装了两个DolphinScheduler相关工具ds.query_workflow和ds.trigger_workflow。前者的作用是查询工作流列表、获取工作流定义ID、查最近执行记录后者是真正触发执行。封装过程中有两个细节容易被忽略我写出来DolphinScheduler的API是分版本的。V1.3和V2.0的接口路径和鉴权方式差不少。我见过不少踩坑帖子直接把网上搜到的API代码拿来用结果因为版本不匹配返回一堆莫名其妙的错误。封装的时候一定要先确认目标环境的DS版本再选对接方案。触发执行分直接触发和指定调度时间触发两种。重跑昨天的失败任务一般用调度时间参数指定schedule_time为昨天那个计划时间点手动随即跑一次才是直接点击运行按钮的效果。这两个语义如果搞混跑出来的任务结果可能完全不对。5.3 执行监控与状态机流转任务被触发后羲和不能干等也不能问一次就完事。它需要进入一个状态机循环TRIGGERED → RUNNING → SUCCESS链路结束通知成功 ↓ FAILURE/TIMEOUT 重试判断是否重试剩余次数 ↓ 重试 重新触发回到 RUNNING ↓ 不重试 发送企微告警链路结束这个状态机我用Python写了一个简单的执行器核心逻辑如下class WorkflowExecutor: def __init__(self, workflow_id, max_retry2, poll_interval30): self.workflow_id workflow_id self.max_retry max_retry self.retry_count 0 self.poll_interval poll_interval def run(self): while True: status self._poll_status() if status SUCCESS: return self._notify_success() if status FAILURE: self.retry_count 1 if self.retry_count self.max_retry: self._trigger_again() continue return self._notify_failure() if status TIMEOUT: self._notify_manual_review() return time.sleep(self.poll_interval)这里有个很实用的判断重试之前一定要先看失败原因再做决定。如果是代码逻辑错误比如SQL写错了重试多少次都白搭应该直接告警让人来处理如果是资源不足、并发冲突这类临时性问题重试才有意义。这个判断逻辑在初版我没做导致一个SQL错误的任务傻乎乎重试了两轮才告警白白浪费了十几分钟。后来我让模型每次失败时先抓日志、分析失败原因、判断是否值得重试再决定是否进入重试分支效果好了很多。5.4 企业微信告警告警内容的模板设计任务最终还是失败且不打算继续重试时就要发企业微信告警了。我封装了notification.send工具支持企微机器人消息、企微应用消息两种通道。这里重点讲讲告警内容的设计——很多人以为告警就是把错误堆栈贴上去就行其实不是。有效的告警内容需要包含五件事任务名、运行ID、失败阶段、失败原因摘要、处理建议。比如【羲和告警】数据日报任务执行失败 - 任务名order_reconcile_daily - 运行IDRUN-20241107-001 - 失败阶段SQL执行step3/5hive sql - 失败原因表 dwd_order_detail 分区数据未就绪job timeout after 600s - 建议检查上游调度dwd_order_detail是否正常产出或等待分区可用后重跑 触发人张三来自羲和Agent手动触发原样贴堆栈的告警在群里没人愿意看而这种告警一眼就知道是谁的事、该干什么。这个模板我是在被运维同事吐槽了两次之后改出来的。5.5 风险控制执行任务必须先过这一关任务执行和代码问答最大的不同是执行会有真实影响。我在执行链路里强制加了三个关卡缺一不可第一次执行必确认任何工具调用只要属于会改变系统状态的类别触发任务、写数据库、发消息执行前都要向用户确认一次确认执行吗只有读操作查询状态可以不确认。这个规则听从了很多老运维的建议宁可多一步不可少一步。脏数据保护涉及重跑、补数这类操作时羲和必须先查询目标分区的数据现状如果已有数据且不是空表它会提示该分区已有数据重跑会覆盖是否继续。操作审计每一次工具调用、谁触发的、什么时候、参数是什么、结果如何全部写入审计日志。这个日志在排查这个任务是谁跑的这种问题上能救你的命。6. 实测中的翻车现场与兜底设计理论讲了不少但真实的Agent开发从来不缺意外。这一章我写几个印象深刻的翻车现场以及我针对性的兜底设计。每一个都是用真金白银换来的教训。6.1 事故一模型在企微告警内容里插入了一大段免责声明第一次把企微告警接进线上群时模型发送的消息末尾多了一行本告警由AI自动生成可能存在错误请谨慎参考最终解释权归AI所有。群里瞬间就炸了。运维大哥直接私聊我你这是什么意思让我别信这告警后来我查了原因模型在生成消息时受到了提示词里注意告知用户这是AI生成内容的安全指令影响自作主张加进了告警正文。这提醒了我一个问题——提示词里的安全约束一定要区分该给用户看的和不该给用户看的。告警内容是面向用户的产品文案不是说教现场。后来我把所有面向用户的输出模板都从模型自由发挥改成了模型填参数模板渲染输出告警语气的稳定度一下子就上来了。6.2 事故二重试机制差点重复跑了两遍任务有段时间一个上游数据任务经常因为资源竞争失败。执行器第一次失败后自动重试居然成功了。但用户后来发现任务对应的数据表生成了两份数据。排查后发现DolphinScheduler的工作流配了对目标表的覆盖写策略先删后写但任务本身没有做幂等控制。当任务第一次实际执行了一部分、只是状态上报失败时重试会导致数据写两遍。这个事故让我彻底接受了幂等性必须前置判断的原则不确认幂等的任务AI绝不自动重试宁可告警给人看。后来我把DolphinScheduler工作流按是否幂等分成了红绿两类名单。绿名单明确幂等、可安全重试里的任务AI可以自动重试红名单里的任务执行失败直接转人工。这个红绿名单机制看起来土但比任何花哨的提示词都管用。6.3 事故三上下文爆炸导致Agent失忆有一次羲和在执行一个很耗时的任务轮询状态轮了很多次每一轮结果都往上下文里塞。跑了大概二十分钟后模型突然开始胡言乱语把之前确认过的「任务ID」和「企微群ID」都记串了差点把告警消息发到另一个群里。这不是玄学是标准的上下文超载问题。长对话场景下早期关键信息会被淹没。兜底方案三件事一是执行关键信息提取每轮只把大象限的状态当前状态、累计耗时、最近错误塞进上下文而不是原始JSON二是关键参数锚定到上下文开头——任务ID、目标群ID、用户ID只写一遍但放在系统提示词最前面确保模型始终能看到三是超长任务用压缩摘要把中间N轮轮询压缩成一行已轮询N次前N-1次均为RUNNING无异常。这套兜底做完之后再也没出现告警发错群的乌龙。6.4 事故四企微告警风暴失败任务一口气刷屏某次DolphinScheduler整个集群出问题几十个任务集体失败。羲和按流程每个失败任务发一条企微告警一分钟内群里被刷了一百多条同格式的消息。运维同事差点顺着网线来打我。从那之后我给告警模块加了聚合与去重策略相同时间段内同类型同根因的失败任务合并成一条聚合告警单条告警里列任务清单和失败原因五分钟内相同任务只上报一次不再重复刷屏。另外还加了一个简单的自适应降噪如果连续多次告警后任务仍然失败就自动降低告警频率从每次失败都告警降为每十分钟汇总一次。这个改造上线后企微群的存活率显著上升。6.5 兜底设计的核心思路追了这么多事故我总结出一个核心兜底思路Agent的自主权和责任成反比——AI能自主做的动作越多兜底设计就必须越严密。具体来说可以从六个维度兜底维度兜底手段权限最小权限原则AI默认只有只读权限写操作必须临时授权操作确认非幂等/破坏性操作执行前必须人工确认重试策略幂等任务自动重试非幂等任务转人工上下文管理关键信息锚定、进程压缩、防止超载失忆告警聚合、去重、降噪避免风暴审计全量操作留痕支持一键回查这六个维度不一定做得多重但每个都不能缺。缺哪一个对应的风险就会在某一天以事故的形式找上你。7. 从设计能跑到跑得可靠几个关键工程判断羲和目前的状态是代码问答稳定可用任务执行在限定场景里已经能承担一部分值班工作。但这一路下来我最大的感受是——Agent能不能落地七分在工程三分在模型。我认为最重要的工程判断有三个一是模型选型要克制。不是所有任务都需要最强的模型。羲和里意图识别用轻量模型就够了便宜又快规划层才需要上强推理模型最终答复话术整理又可以用轻量模型。一套任务链路里多个模型各司其职成本能降一半响应速度还能快一截。这比追求一个模型干所有事靠谱得多。二是别让Agent自己决定边界。模型天然会给自己加戏。你以为它只会执行工具调用它能在告警消息里给你写免责声明你以为它会按计划执行它会因为上下文一长就把关键参数记串。所以边界规则必须全部写在工程代码里——能不能重试、要不要确认、发不发告警全部由代码和配置决定而不是由模型的临场发挥决定。三是从小切口开始落地。如果你问我做AI编码助手最先该接哪个任务执行能力我会毫不犹豫地推荐查询类任务——查任务状态、查日志、查配置。这类操作是只读的出不了大事用户又高频需要能让Agent快速积累信任感。跑通了只读任务再慢慢加上触发重跑、告警通知这类有副作用的操作。步子迈大了容易把整个项目做成事故现场。最后分享一个小技巧关于模型提示词和工具调用最后分享一个我的个人技巧价值不低但说的人不多。我给羲和写工具提示词的时候不追求告诉模型所有可能的情况而是刻意写一个反直觉的指令如果不能确定用哪个工具就直接问用户不要猜。一开始团队有人担心这不就变笨了吗实际上这一条指令让执行成功率反而提高了。因为模型发现自己可以承认不确定之后就不硬编了很多低级错误在源头就被拦下来。AI编码助手这个方向的想象空间很大但竞争也激烈。谁能从会说话走到会干活谁才能真正解决开发者的痛点。希望这篇分享能给你一些实打实的参考。有些细节如果没写透欢迎一起讨论。
返回列表