AI Agent工程化：分层交付架构设计与落地实践

1. 为什么“分层交付”是 AI Agent 工程化的第一道生死线做 AI Agent 项目最怕什么不是模型不够聪明而是你把所有逻辑——意图识别、工具调用、状态管理、结果渲染——全塞进一个巨大的提示词或者一个巨型函数里。我见过太多团队Demo 阶段跑得挺欢一旦要接入真实业务、要多人协作、要灰度发布整个系统就像被搅成一锅粥的八宝饭谁也不敢动其中任何一粒米。“分层交付”这个词听起来像架构师爱拽的黑话但说白了就一件事把 Agent 的不同关注点拆成独立的层每层有明确的输入输出契约可以单独开发、单独测试、单独替换。这跟前端工程化里把 UI、状态、路由、请求分层是一个道理只不过 Agent 多了一个“不确定性”的维度——大模型的输出天然是概率性的你更需要用工程手段把这种不确定性关进笼子里。这一篇我就拿一个真实的客服工单处理 Agent 作为贯穿案例把分层交付从设计思路到落地代码全部拆开讲。这个 Agent 要干的事很简单用户提交一段自然语言描述的问题Agent 判断问题类型、提取关键信息、调用内部 API 查询订单状态、生成回复话术、必要时转人工。麻雀虽小五脏俱全分层没做好后面加一个“多轮追问”功能就能让你重构三天。适合谁看如果你正在从零搭建 AI Agent或者手里的 Agent 已经超过 500 行代码开始变得难以维护又或者你是一个团队的技术负责人需要制定 Agent 开发规范这篇内容可以直接拿去当参考模板。我会给出具体的目录结构、接口定义、参数计算和踩坑记录不是泛泛而谈的架构图。2. 分层交付的整体设计思路与选型考量2.1 为什么不能“一锅端”从一次线上事故说起去年我参与过一个电商售后 Agent 的改造。原系统是一个 1200 行的 Python 文件里面用 if-else 判断用户意图用正则提取订单号用 f-string 拼接提示词然后直接调模型 API拿到结果后又在同一个函数里做敏感词过滤和格式化输出。上线第一周就出了事运营想调整回复话术的礼貌程度开发改了一行提示词结果意图识别的准确率从 92% 掉到了 78%。原因很简单——提示词里同时包含了“判断意图”和“生成话术”两个任务的指令改一个必然影响另一个。这就是典型的“智能糊进一锅”。大模型对提示词是全局敏感的你把多个任务的指令混在一起它们会互相干扰。分层交付的第一个价值就是隔离变化话术调整不应该影响意图识别工具调用逻辑变更不应该影响结果渲染。第二个价值是可测试性。一锅端的代码你没法写单元测试因为每次调用模型结果都不一样。分层之后意图识别层可以用固定输入做回归测试工具调用层可以用 mock 数据做契约测试只有生成层需要人工评估。测试成本直接降一个数量级。第三个价值是可替换性。今天你用某个模型做意图识别明天想换一个更便宜更快的只要接口契约不变上层代码一行不用改。这在模型快速迭代的当下是保命的能力。2.2 分几层怎么分——基于职责而非基于技术分层最容易犯的错误是按技术分层模型层、API 层、数据库层。这是后端开发的习惯但 Agent 的分层应该按职责来。我推荐的分法是四层加一个横切层层级职责输入输出变化频率接入层接收用户请求、鉴权、限流、会话管理HTTP/WebSocket 请求标准化请求对象低认知层意图识别、实体抽取、任务规划标准化请求对象结构化意图与槽位中执行层工具调用、API 编排、状态流转结构化意图与槽位执行结果对象高表达层话术生成、格式化、敏感词过滤执行结果对象最终回复文本高横切层日志、监控、缓存、降级各层埋点可观测数据低这个分法的核心逻辑是变化频率越高的层越靠上越稳定的层越靠下。接入层和横切层几乎不变认知层偶尔调整执行层和表达层天天在改。把天天改的东西和几乎不变的东西混在一起就是自找麻烦。还有一个关键决策认知层和执行层之间要不要加一个“规划层”对于简单 Agent意图识别完直接执行就行。但对于需要多步推理的任务比如“帮我查一下上周买的那个蓝色杯子到哪了如果还没发货就取消订单”这里包含查询和条件取消两个动作就需要一个规划层把任务拆成步骤。我的建议是初期不要加等遇到第一个多步任务再加。过早抽象比不抽象更可怕。2.3 层间通信用数据结构而不是自然语言这是分层交付里最容易被忽视但最致命的一点。很多团队分层了但层与层之间传的还是自然语言字符串。比如认知层输出“用户想查订单状态订单号是 12345”执行层再去解析这个字符串。这等于没分层只是把一锅粥分装到了几个碗里吃的时候还得倒回锅里。正确的做法是定义严格的数据结构。用 PydanticPython或者 ZodTypeScript定义每一层的输入输出模型层间传递的是序列化后的 JSON 对象。这样做的好处是类型检查能在编译期发现错误契约测试能自动生成日志能结构化查询。from pydantic import BaseModel, Field from enum import Enum from typing import Optional class IntentType(str, Enum): QUERY_ORDER query_order CANCEL_ORDER cancel_order REFUND_REQUEST refund_request HUMAN_HANDOFF human_handoff UNKNOWN unknown class CognitionResult(BaseModel): intent: IntentType confidence: float Field(ge0.0, le1.0) slots: dict Field(default_factorydict) raw_query: str trace_id: str class ExecutionResult(BaseModel): success: bool data: Optional[dict] None error_code: Optional[str] None error_message: Optional[str] None trace_id: str上面这段代码定义了认知层和执行层的输出契约。注意trace_id贯穿所有层这是可观测性的基础。confidence字段用于降级决策——置信度低于阈值时直接转人工不要硬着头皮往下走。3. 核心层级的细节拆解与实操要点3.1 认知层意图识别不是分类问题是决策问题很多人把意图识别当成一个文本分类任务拿一堆标注数据训一个 BERT 就完事。但在 Agent 场景里意图识别的本质是决策不仅要判断用户想干什么还要判断自己能不能干、该不该干、置信度够不够。我的实操做法是“规则兜底 模型主判 置信度门控”三段式。规则兜底处理高频确定场景比如用户消息里包含“转人工”三个字直接走人工通道不要浪费模型调用。模型主判用 Few-shot Prompting 而不是微调因为 Agent 的意图集合变化频繁微调跟不上迭代速度。置信度门控是安全阀低于 0.7 的置信度直接走澄清流程让用户确认意图。提示词的设计有个关键技巧把意图定义和判断标准分开写。不要写“判断用户是想查订单还是想退款”而要写“如果用户消息中包含订单号且询问物流状态则意图为 query_order如果用户明确表达退款意愿且提供了退款原因则意图为 refund_request”。前者是让模型猜后者是给模型判据。COGNITION_PROMPT 你是一个意图识别引擎。根据用户消息判断意图并抽取槽位。 意图定义 - query_order: 用户询问订单状态、物流信息。必须包含订单号或可识别的订单指代。 - cancel_order: 用户明确要求取消订单。必须包含订单号。 - refund_request: 用户要求退款。必须包含订单号和退款原因。 - human_handoff: 用户明确要求人工服务或表达强烈不满。 - unknown: 无法归入以上任何一类。 输出格式严格 JSON {intent: ..., confidence: 0.0-1.0, slots: {...}} 用户消息{user_message} 注意confidence是让模型自己给的虽然模型给的置信度不一定校准但作为相对排序是有效的。实测下来模型给 0.9 以上的意图准确率确实高于给 0.6 的这个信号可以用。3.2 执行层工具调用的幂等性与超时控制执行层是 Agent 真正“动手”的地方也是最容易出生产事故的地方。我踩过最惨的坑是用户点了两次提交Agent 调了两次取消订单 API结果第二笔取消报错整个会话状态错乱。所有工具调用必须幂等。具体做法是在执行层维护一个idempotency_key由trace_id intent slots_hash生成。同一个 key 的调用在 5 分钟内直接返回缓存结果不重复执行。这个逻辑不要放在工具内部要放在执行层的编排器里因为工具可能是第三方 API你改不了。超时控制同样关键。Agent 调用的工具可能是内部微服务、第三方 API、数据库查询响应时间差异巨大。我的经验值是查询类工具超时 3 秒写入类工具超时 5 秒超过就降级。降级策略不是简单报错而是返回一个“系统繁忙已为您记录稍后会有专员联系”的兜底话术同时把任务写入重试队列。import hashlib import time from functools import wraps class ToolExecutor: def __init__(self, cache_client, retry_queue): self.cache cache_client self.retry_queue retry_queue self.timeout_map { query_order: 3.0, cancel_order: 5.0, refund_request: 5.0, } def _make_idempotency_key(self, trace_id, intent, slots): raw f{trace_id}:{intent}:{sorted(slots.items())} return hashlib.sha256(raw.encode()).hexdigest()[:16] def execute(self, trace_id, intent, slots): key self._make_idempotency_key(trace_id, intent, slots) cached self.cache.get(key) if cached: return cached timeout self.timeout_map.get(intent, 3.0) start time.time() try: result self._dispatch(intent, slots, timeout) self.cache.setex(key, 300, result) return result except TimeoutError: self.retry_queue.push({trace_id: trace_id, intent: intent, slots: slots}) return ExecutionResult( successFalse, error_codeTIMEOUT, error_message工具调用超时已进入重试队列, trace_idtrace_id )这段代码里_dispatch方法根据 intent 路由到具体的工具函数每个工具函数内部用signal.alarm或者asyncio.wait_for实现超时。注意缓存时间设 300 秒太短起不到幂等作用太长会导致用户改了订单号还拿到旧结果。5 分钟是个经验值你可以根据业务调整。3.3 表达层话术生成与安全过滤的先后顺序表达层最容易犯的错是先过滤再生成或者边生成边过滤。正确的顺序是先生成完整话术再做安全过滤最后做格式化。为什么因为安全过滤可能会截断或替换部分内容如果先生成一半就过滤模型后续的生成会基于被污染的前文导致语义断裂。安全过滤我建议用“规则 小模型”双通道。规则通道处理明确的敏感词、竞品名称、承诺性话术如“保证退款”直接替换或删除。小模型通道处理隐晦的风险表达比如阴阳怪气、暗示性承诺这个用一个小尺寸的文本分类模型就够了不需要调大模型。还有一个细节话术模板和模型生成要混合使用。对于高频确定场景比如“您的订单已发货物流单号是 XXX”直接用模板填充不要调模型。模型只用于需要灵活表达的场景比如安抚情绪、解释复杂政策。这样既保证了一致性又控制了成本和延迟。class ExpressionLayer: def __init__(self, llm_client, safety_filter, template_engine): self.llm llm_client self.safety safety_filter self.templates template_engine def render(self, execution_result, cognition_result): if execution_result.success and cognition_result.intent IntentType.QUERY_ORDER: template self.templates.get(query_order_success) raw_text template.render( order_noexecution_result.data[order_no], statusexecution_result.data[status], logisticsexecution_result.data.get(logistics, 暂无物流信息) ) else: raw_text self.llm.generate( promptself._build_expression_prompt(execution_result, cognition_result) ) filtered_text self.safety.filter(raw_text) return self._format(filtered_text)模板引擎用 Jinja2 或者简单的str.format都行关键是模板要版本化管理每次话术调整都要记录变更原因和影响范围。我见过团队直接在生产环境改模板字符串结果一个{}占位符写错导致所有回复都带上了KeyError这种低级错误用版本管理完全可以避免。4. 完整实操流程从零搭建一个分层 Agent4.1 项目目录结构与依赖管理先看目录结构这是分层的物理体现。我推荐按层分目录而不是按功能分目录。按功能分目录比如order/、refund/会导致每加一个业务就要横跨所有层按层分目录则每层内部可以独立演进。agent/ ├── gateway/ # 接入层 │ ├── __init__.py │ ├── router.py # 路由与鉴权 │ ├── session.py # 会话管理 │ └── schemas.py # 请求/响应模型 ├── cognition/ # 认知层 │ ├── __init__.py │ ├── intent.py # 意图识别 │ ├── slot.py # 槽位抽取 │ └── prompts/ # 提示词模板 │ └── intent_v3.txt ├── execution/ # 执行层 │ ├── __init__.py │ ├── executor.py # 编排器 │ ├── tools/ # 具体工具 │ │ ├── order.py │ │ └── refund.py │ └── idempotency.py # 幂等控制 ├── expression/ # 表达层 │ ├── __init__.py │ ├── renderer.py # 话术生成 │ ├── safety.py # 安全过滤 │ └── templates/ # 话术模板 │ └── query_order_success.j2 ├── crosscutting/ # 横切层 │ ├── __init__.py │ ├── logging.py │ ├── metrics.py │ └── tracing.py └── config/ ├── settings.py └── layer_contracts.py # 层间契约定义依赖管理有个原则上层可以依赖下层下层不能依赖上层同层之间尽量不依赖。认知层不能 import 执行层的代码执行层不能 import 表达层的代码。层间通过layer_contracts.py里定义的数据模型通信。这个约束用 import-linter 或者简单的 CI 检查就能强制执行。4.2 层间契约的定义与版本管理layer_contracts.py是整个项目的核心文件它定义了所有层间传递的数据结构。这个文件应该由架构师或者技术负责人维护任何变更都需要评审。from pydantic import BaseModel, Field from typing import Optional, Literal from datetime import datetime class GatewayRequest(BaseModel): user_id: str session_id: str message: str channel: Literal[app, web, wechat] timestamp: datetime class CognitionResult(BaseModel): intent: str confidence: float slots: dict needs_clarification: bool False clarification_question: Optional[str] None trace_id: str class ExecutionResult(BaseModel): success: bool data: Optional[dict] None error_code: Optional[str] None error_message: Optional[str] None should_retry: bool False trace_id: str class ExpressionResult(BaseModel): text: str format: Literal[plain, markdown, card] plain quick_replies: list[str] Field(default_factorylist) trace_id: str契约版本管理用语义化版本号比如CognitionResultV2。当字段发生不兼容变更时新建一个类而不是改旧的旧类保留至少一个迭代周期。这样灰度发布时新旧版本可以共存不会出现认知层已经升级但执行层还没升级导致的解析失败。4.3 横切层的埋点与可观测性横切层不参与业务逻辑但决定了你出问题时能不能快速定位。我要求每个层的关键节点都必须打三类埋点耗时、结果状态、关键参数。耗时埋点用装饰器实现自动记录每个层方法的执行时间。结果状态记录成功/失败/降级。关键参数记录 trace_id、intent、confidence 这些用于关联分析的字段。所有埋点输出结构化 JSON直接进日志系统不要用 print。import time import json import logging from functools import wraps logger logging.getLogger(agent.observability) def observe(layer_name: str): def decorator(func): wraps(func) def wrapper(*args, **kwargs): trace_id kwargs.get(trace_id) or unknown start time.time() try: result func(*args, **kwargs) elapsed (time.time() - start) * 1000 logger.info(json.dumps({ layer: layer_name, method: func.__name__, trace_id: trace_id, elapsed_ms: round(elapsed, 2), status: success })) return result except Exception as e: elapsed (time.time() - start) * 1000 logger.error(json.dumps({ layer: layer_name, method: func.__name__, trace_id: trace_id, elapsed_ms: round(elapsed, 2), status: error, error: str(e) })) raise return wrapper return decorator这个装饰器挂在每个层的入口方法上比如cognition.intent.recognize、execution.executor.execute、expression.renderer.render。有了这些数据你可以快速回答“今天意图识别平均耗时多少”“哪个 intent 的失败率最高”“表达层有没有出现超时”这些问题。4.4 端到端串联与灰度发布策略所有层开发完之后用一个 orchestrator 把它们串起来。orchestrator 的代码应该极薄只做流程编排不包含任何业务逻辑。class AgentOrchestrator: def __init__(self, gateway, cognition, execution, expression): self.gateway gateway self.cognition cognition self.execution execution self.expression expression observe(orchestrator) def handle(self, request: GatewayRequest) - ExpressionResult: trace_id generate_trace_id() cognition_result self.cognition.recognize(request, trace_id) if cognition_result.needs_clarification: return self.expression.render_clarification(cognition_result) if cognition_result.confidence 0.7: return self.expression.render_handoff(cognition_result) execution_result self.execution.execute( trace_id, cognition_result.intent, cognition_result.slots ) return self.expression.render(execution_result, cognition_result)灰度发布时按user_id哈希取模分流。比如 10% 的用户走新版本认知层90% 走旧版本。新版本认知层的CognitionResult如果字段有变化执行层需要同时兼容新旧两个版本。这就是为什么契约要版本化——灰度期间两套契约并存是常态。5. 常见问题与排查技巧实录5.1 意图识别准确率突然下降怎么排查这是最高频的问题。排查顺序应该是先看输入分布有没有变化再看提示词有没有被改动最后看模型版本有没有升级。输入分布变化是最常见的原因。比如运营做了一次活动用户消息里突然大量出现“活动”“优惠”这些词而你的意图定义里没有覆盖模型就会把这些归到 unknown 或者错误归类。排查方法是拉取最近 24 小时的 unknown 意图样本人工看 50 条基本就能定位。提示词改动是第二常见原因。前面说过提示词是全局敏感的改一个词可能影响所有意图。排查方法是做 A/B 对比用旧提示词和新提示词跑同一批测试集看每个意图的准确率变化。测试集至少 200 条覆盖所有意图每个意图 40 条以上。模型版本升级是最隐蔽的。同一个模型的不同版本对同一个提示词的响应可能完全不同。排查方法是固定测试集每次模型升级前跑一遍回归准确率下降超过 3 个百分点就回滚。排查项检查方法典型表现处理方式输入分布拉取 unknown 样本人工看大量新词出现补充意图定义或加规则兜底提示词改动A/B 对比测试集某个意图准确率骤降回滚提示词或重新调优模型版本固定测试集回归所有意图均匀下降回滚模型版本槽位抽取失败检查 slots 为空的比例意图对但槽位缺失检查抽取提示词和正则5.2 工具调用超时和重试的坑工具调用超时不可怕可怕的是超时后的重试把下游系统打挂。我踩过的坑是某个查询接口超时后执行层自动重试了 3 次每次间隔 1 秒结果下游数据库连接池被打满整个服务雪崩。正确的重试策略是指数退避 抖动 熔断。第一次重试等 1 秒第二次等 2 秒第三次等 4 秒每次加一个 0 到 1 秒的随机抖动避免多个请求同时重试。连续失败 10 次后熔断 30 秒期间所有请求直接返回降级结果不再调用下游。还有一个坑是重试的幂等性。如果工具本身不幂等重试会导致重复写入。比如取消订单接口第一次调用超时但实际已经执行成功重试第二次会报“订单已取消”。这时候执行层要能识别这种“业务成功但 HTTP 报错”的情况不能简单当成失败。我的做法是在工具层返回一个business_success标志即使 HTTP 状态码是 500只要业务语义是成功的就当成成功处理。5.3 表达层话术不一致的治理多个开发同时改话术模板很容易出现同一个场景在不同入口话术不一致。治理方法是话术模板集中管理 变更评审 自动化对比测试。所有话术模板放在expression/templates/目录下用 Git 管理。每次变更必须提交 MR由产品经理和开发共同评审。评审通过后CI 自动跑对比测试用同一组 execution_result 渲染新旧模板输出差异报告确认差异符合预期才允许合并。还有一个细节话术里的变量占位符要有默认值。比如{{ logistics }}如果 execution_result 里没有这个字段模板渲染会报错。默认值写成{{ logistics | default(暂无物流信息) }}这样即使上游数据缺失也不会导致整个回复失败。5.4 层间契约变更的兼容性处理契约变更是最需要谨慎的操作。我的原则是新增字段永远兼容删除字段必须走废弃流程修改字段类型绝对禁止。新增字段时给默认值旧版本代码不传这个字段也能正常工作。删除字段时先标记为 deprecated保留至少两个迭代周期期间日志里记录所有还在使用该字段的调用方等调用方全部迁移后再删除。修改字段类型比如把confidence从 float 改成 string这是绝对禁止的只能新建一个字段名。灰度期间新旧契约并存执行层需要做版本适配。适配逻辑放在 orchestrator 里不要污染各层内部代码。def adapt_cognition_result(raw: dict) - CognitionResult: if confidence in raw: return CognitionResult(**raw) else: return CognitionResult( intentraw[intent], confidence1.0, slotsraw.get(slots, {}), trace_idraw[trace_id] )这个适配函数在灰度结束后就可以删掉不会留下技术债务。6. 分层交付的边界与团队协作规范分层交付不只是技术问题更是团队协作问题。如果团队没有共识再好的分层设计也会被慢慢侵蚀。我建议在项目初期就定下几条铁律写进开发规范里。第一条禁止跨层调用。表达层不能直接调执行层的工具认知层不能直接读数据库。所有跨层通信必须通过契约对象。这条用 CI 检查 import 关系就能强制执行。第二条每层必须有独立的测试。认知层用固定输入做回归测试执行层用 mock 工具做契约测试表达层用固定 execution_result 做渲染测试。测试覆盖率不要求 100%但核心路径必须覆盖。第三条提示词和模板必须版本化。每次变更记录变更人、变更原因、影响范围。生产环境使用的提示词版本号要打进日志出问题时能快速定位是哪个版本。第四条横切层埋点不可省略。新加一个功能必须同时加埋点。没有埋点的功能不允许上线。这条听起来严苛但出过一次线上问题找不到日志之后团队就会自觉遵守了。分层交付的最终目标不是架构好看而是让变化可控。当产品说“明天要加一个退款进度查询”你能在认知层加一个意图、执行层加一个工具、表达层加一个模板三个改动互不影响半天上线。当模型供应商说“下个月要下线旧模型”你能只改认知层的模型调用其他层一行不动。这才是工程化的价值。我个人在实际操作中的体会是分层的前期成本大概多花 20% 的时间但后期每次需求变更能省 50% 以上的时间。项目周期超过三个月分层就是稳赚不赔的买卖。唯一要注意的是不要过度分层四层加横切层对绝大多数 Agent 项目足够了再加层就是给自己找麻烦。