AI Agent工程化实战：分层交付架构设计与五层实现指南

1. 为什么“分层交付”是 AI Agent 工程化的第一道生死线

我见过太多团队在 Demo 阶段惊艳全场，一进生产环境就原形毕露。问题往往不出在模型能力上，而是出在架构层面——他们把提示词、工具调用、业务逻辑、状态管理、错误处理全部塞进一个函数或者一个类里，美其名曰“快速迭代”，实际上是给自己埋了一颗定时炸弹。AI Agent 工程化的核心命题之一，就是分层交付。这个词听起来像是架构师的口头禅，但它的实际含义非常具体：把 Agent 的不同职责切分到独立的层，每层有明确的输入输出契约，可以独立开发、独立测试、独立部署、独立替换。

你可能会问，为什么不能像写一个普通脚本那样，把所有逻辑写在一起？答案很简单：因为 Agent 的行为是不确定的。大语言模型的输出具有随机性，工具调用的结果具有不可靠性，多轮对话的状态具有复杂性。当这三重不确定性叠加在一起时，如果你没有一个清晰的分层结构，排查问题就像在一锅粥里找一粒特定的米。分层交付的本质，是把不确定性关进笼子里，让每一层的不确定性被隔离、被观测、被控制。

这一篇是“AI Agent 工程化实战”系列的第二篇，聚焦的就是这个分层问题。我会从整体设计思路讲起，然后逐层拆解每一层的职责、实现要点和常见坑，最后给出一个可以直接参考的分层方案。无论你用的是 Python 还是 Java，无论你基于哪个模型服务，这套分层思路都是通用的。

2. 分层交付的整体设计思路与方案选型

2.1 从“一锅糊”到“分灶吃饭”：核心设计哲学

先说说“一锅糊”的典型症状。我见过一个项目，整个 Agent 就是一个agent.py文件，里面有一个run函数，大概长这样：接收用户输入，拼接系统提示词，调用模型，解析模型返回，如果返回里有工具调用就执行工具，把工具结果再拼回提示词，再次调用模型，循环直到没有工具调用，最后返回结果。所有逻辑都在一个 while 循环里，状态用一个字典维护，错误处理就是 try-except 包住整个循环。

这种写法在 Demo 阶段没问题，但一旦上线，你会遇到以下问题：第一，模型换了，提示词要改，但提示词散落在代码各处，改一处漏一处；第二，工具增加了，工具调用的解析逻辑和业务逻辑耦合在一起，加一个工具要动核心循环；第三，需要加日志和监控，发现根本没有合适的埋点位置；第四，需要支持多轮对话，状态管理混乱，上下文窗口经常溢出；第五，需要做 A/B 测试，两个版本的提示词无法并行运行。

分层交付的设计哲学就是“分灶吃饭”。每一层只关心自己的事，层与层之间通过明确的接口通信。这样做的好处是：替换模型时只动模型层，增加工具时只动工具层，调整业务逻辑时只动编排层，加监控时在层间插桩即可。每一层都可以独立测试，比如模型层可以用 mock 数据测试，工具层可以用单元测试覆盖，编排层可以用集成测试验证。

2.2 分层方案选型：四层还是五层？

业界常见的分层方案有三层、四层、五层不等。三层通常是“接入层-编排层-模型层”，四层会增加“工具层”，五层会增加“状态层”或“记忆层”。我的建议是采用五层结构，因为状态管理在 Agent 场景中太重要了，单独抽出来会让整个架构清晰很多。

具体来说，我推荐的分层是：接入层、编排层、模型层、工具层、状态层。接入层负责接收请求和返回响应，处理协议转换和鉴权；编排层是 Agent 的“大脑”，负责决策流程控制；模型层封装大语言模型的调用，屏蔽不同模型服务的差异；工具层封装所有外部能力，包括 API 调用、数据库查询、代码执行等；状态层负责对话历史、会话状态、长期记忆的存储和检索。

为什么这样分？因为每一层的变更频率不同。接入层变更频率最低，一旦协议确定很少改动；编排层变更频率中等，业务逻辑调整时会改；模型层变更频率取决于模型迭代速度，可能几个月换一次；工具层变更频率较高，业务需求变化时会频繁增删工具；状态层变更频率也较高，存储方案和检索策略会不断优化。把变更频率不同的东西放在不同的层，可以避免“牵一发而动全身”。

2.3 层间通信契约：接口设计的关键考量

分层之后，层与层之间怎么通信？这是最容易出问题的地方。我的经验是：层间通信必须使用明确的数据结构，禁止传递裸字典或裸字符串。比如编排层调用模型层时，不应该传一个prompt字符串，而应该传一个ModelRequest对象，里面包含messages列表、temperature参数、max_tokens限制等。模型层返回的也不应该是一个字符串，而是一个ModelResponse对象，里面包含content、tool_calls、usage等信息。

这样做的好处是：第一，类型安全，编译期或运行期就能发现字段缺失；第二，可扩展，加字段不影响已有代码；第三，可测试，构造测试数据方便；第四，可观测，日志里打印对象比打印字典清晰。我见过太多项目因为层间传字典，导致字段名拼写错误、类型不一致、默认值缺失等问题，排查起来非常痛苦。

另外，层间通信应该是单向依赖的。接入层依赖编排层，编排层依赖模型层、工具层、状态层，模型层和工具层不依赖编排层，状态层不依赖任何其他层。这种单向依赖保证了每层可以独立替换和测试。如果出现循环依赖，说明分层设计有问题，需要重新审视职责划分。

3. 核心层级的职责拆解与实操要点

3.1 接入层：不只是收发包那么简单

接入层看起来最简单，就是接收 HTTP 请求，解析参数，调用编排层，返回响应。但实际上，接入层承担着很多容易被忽视的职责。首先是协议适配，你的 Agent 可能同时需要支持 REST API、WebSocket、gRPC 等多种协议，接入层要负责把这些协议统一转换成内部调用格式。其次是鉴权和限流，不同用户有不同的权限和配额，接入层要负责校验和拦截。再次是请求预处理，比如参数校验、格式转换、敏感词过滤等。

我在实际项目中踩过一个坑：接入层直接把用户输入透传给编排层，没有做长度限制。结果有用户输入了一篇几万字的文章，导致模型调用超时，整个服务被拖垮。后来在接入层加了输入长度校验，超过阈值直接返回错误提示，问题才解决。所以接入层一定要做输入校验和防御，不能假设上游传来的数据是合法的。

还有一个容易忽视的点是超时控制。Agent 的执行时间通常比普通 API 长，因为涉及多次模型调用和工具调用。接入层要设置合理的超时时间，并且要区分“连接超时”和“读取超时”。如果超时时间设置太短，正常请求会被中断；如果设置太长，异常请求会占用资源。我的经验是：根据业务场景的 P99 耗时来设置，通常留 2-3 倍余量。

3.2 编排层：Agent 的决策中枢

编排层是整个 Agent 的核心，它决定了“什么时候调用模型、什么时候调用工具、什么时候结束”。这一层的设计直接影响到 Agent 的智能程度和稳定性。常见的编排模式有三种：ReAct 模式、Plan-and-Execute 模式、Workflow 模式。

ReAct 模式是最常见的，就是“思考-行动-观察”循环。模型先输出思考过程，然后决定调用哪个工具，工具返回结果后，模型再根据结果决定下一步。这种模式灵活性强，但容易陷入死循环或者偏离目标。Plan-and-Execute 模式是先让模型制定一个计划，然后按计划逐步执行，执行过程中可以根据情况调整计划。这种模式适合复杂任务，但计划本身可能不合理。Workflow 模式是预定义好流程，模型只在特定节点做决策。这种模式可控性最强，但灵活性最差。

我的建议是：根据任务复杂度选择合适的编排模式，并且支持混合使用。比如一个客服 Agent，简单问题用 Workflow 模式直接走预设流程，复杂问题用 ReAct 模式让模型自主决策。编排层要提供统一的接口，让上层可以根据场景选择不同的编排策略。

编排层还有一个重要职责是循环控制。ReAct 模式本质上是一个循环，必须有终止条件。常见的终止条件包括：模型输出中没有工具调用、达到最大循环次数、达到超时时间、检测到重复调用同一个工具等。我见过一个项目因为没有设置最大循环次数，模型陷入死循环，一夜之间烧掉了几千块的 API 费用。所以循环控制是编排层的生命线，必须设置多重保险。

3.3 模型层：屏蔽差异，统一接口

模型层的核心职责是封装大语言模型的调用，向上提供统一的接口。为什么要单独抽一层？因为模型服务可能随时更换。今天用这个模型，明天可能换另一个；今天用云端 API，明天可能部署私有化模型。如果模型调用逻辑散落在编排层各处，更换模型时就要改很多地方。

模型层要封装的内容包括：请求构造、响应解析、错误处理、重试策略、限流控制、成本统计。请求构造要把统一的ModelRequest转换成具体模型服务的 API 格式；响应解析要把模型服务的返回转换成统一的ModelResponse；错误处理要区分可重试错误和不可重试错误；重试策略要设置合理的重试次数和退避算法；限流控制要防止超过模型服务的 QPS 限制；成本统计要记录每次调用的 token 消耗和费用。

这里有一个关键设计决策：是否支持多模型路由。有些场景下，简单问题用便宜的小模型，复杂问题用昂贵的大模型，可以显著降低成本。模型层可以提供路由能力，根据请求的复杂度或者配置的策略，选择不同的模型。但要注意，不同模型的输出格式可能不同，模型层要做好归一化处理。

3.4 工具层：让 Agent 长出手脚

工具层封装了 Agent 可以调用的所有外部能力。一个设计良好的工具层应该具备以下特征：工具注册机制、参数校验、执行隔离、结果标准化、错误处理。

工具注册机制让新增工具变得简单。我推荐使用装饰器或者配置文件来注册工具，每个工具声明自己的名称、描述、参数 schema 和执行函数。这样编排层只需要知道工具的名称和参数格式，不需要关心工具的具体实现。参数校验要在工具执行前进行，防止非法参数导致工具崩溃。执行隔离要保证一个工具的失败不会影响其他工具，通常用 try-except 包住工具执行，把异常转换成标准化的错误结果。

结果标准化很重要。不同工具返回的数据格式不同，有的返回 JSON，有的返回字符串，有的返回二进制。工具层要把所有结果转换成统一的格式，比如ToolResult对象，包含success、data、error三个字段。这样编排层处理工具结果时就不需要针对每个工具写不同的解析逻辑。

我踩过的一个坑是：工具执行没有设置超时。有一个工具调用外部 API，对方服务挂了，请求一直挂起，导致整个 Agent 卡死。后来给每个工具都加了超时控制，超时后返回错误结果，让模型决定下一步。所以工具层必须设置超时，这是血的教训。

3.5 状态层：记忆的存储与检索

状态层负责管理 Agent 的“记忆”。这包括短期记忆（当前对话的历史消息）和长期记忆（跨对话的用户偏好、历史事实等）。短期记忆通常直接放在上下文窗口里，但要注意 token 限制。长期记忆需要持久化存储，并且要支持检索。

短期记忆的管理策略有几种：全量保留、滑动窗口、摘要压缩。全量保留适合对话轮次少的场景；滑动窗口保留最近 N 轮对话，简单但可能丢失重要信息；摘要压缩用模型对历史对话做摘要，保留关键信息，但会增加模型调用成本。我的建议是：根据业务场景选择合适的策略，并且支持动态调整。比如对话初期全量保留，超过阈值后自动切换到摘要压缩。

长期记忆的存储方案选择很多，可以用关系型数据库、向量数据库、键值存储等。关键是要设计好记忆的写入和检索机制。写入时，要决定哪些信息值得长期保存；检索时，要根据当前对话内容找到相关的记忆。向量数据库适合语义检索，但要注意 embedding 的成本和延迟。我的经验是：长期记忆不要贪多，只存真正有价值的信息，否则检索噪声会很大。

4. 实操过程与核心环节实现

4.1 项目结构搭建：从零开始的分层骨架

先给出一个推荐的项目结构，以 Python 为例：

agent_project/ ├── config/ │ ├── settings.py │ └── prompts/ │ ├── system.yaml │ └── tools.yaml ├── gateway/ │ ├── api.py │ ├── middleware.py │ └── schemas.py ├── orchestrator/ │ ├── engine.py │ ├── strategies/ │ │ ├── react.py │ │ ├── plan_execute.py │ │ └── workflow.py │ └── loop_control.py ├── model/ │ ├── client.py │ ├── adapters/ │ │ ├── openai_adapter.py │ │ └── local_adapter.py │ └── router.py ├── tools/ │ ├── registry.py │ ├── base.py │ └── implementations/ │ ├── search.py │ ├── calculator.py │ └── database.py ├── state/ │ ├── short_term.py │ ├── long_term.py │ └── storage/ │ ├── redis_store.py │ └── vector_store.py └── main.py

这个结构清晰地划分了五层，每层有独立的目录。config目录存放配置和提示词模板，提示词用 YAML 文件管理，方便非开发人员修改。gateway是接入层，orchestrator是编排层，model是模型层，tools是工具层，state是状态层。

4.2 模型层实现：统一接口与适配器模式

模型层的核心是定义一个统一的接口，然后用适配器模式适配不同的模型服务。先定义请求和响应的数据结构：

from dataclasses import dataclass, field from typing import List, Optional, Dict, Any @dataclass class Message: role: str # system, user, assistant, tool content: str tool_call_id: Optional[str] = None tool_calls: Optional[List[Dict]] = None @dataclass class ModelRequest: messages: List[Message] temperature: float = 0.7 max_tokens: int = 2048 tools: Optional[List[Dict]] = None stream: bool = False @dataclass class ModelResponse: content: str tool_calls: List[Dict] = field(default_factory=list) usage: Dict[str, int] = field(default_factory=dict) finish_reason: str = "stop"

然后定义模型客户端的抽象基类：

from abc import ABC, abstractmethod class BaseModelClient(ABC): @abstractmethod async def chat(self, request: ModelRequest) -> ModelResponse: pass @abstractmethod async def stream_chat(self, request: ModelRequest): pass

接着实现具体的适配器。以 OpenAI 兼容接口为例：

import httpx from tenacity import retry, stop_after_attempt, wait_exponential class OpenAICompatibleClient(BaseModelClient): def __init__(self, base_url: str, api_key: str, model: str): self.base_url = base_url self.api_key = api_key self.model = model self.client = httpx.AsyncClient(timeout=60.0) @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) async def chat(self, request: ModelRequest) -> ModelResponse: payload = { "model": self.model, "messages": [self._convert_message(m) for m in request.messages], "temperature": request.temperature, "max_tokens": request.max_tokens, } if request.tools: payload["tools"] = request.tools response = await self.client.post( f"{self.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json=payload ) response.raise_for_status() data = response.json() choice = data["choices"][0] return ModelResponse( content=choice["message"].get("content", ""), tool_calls=choice["message"].get("tool_calls", []), usage=data.get("usage", {}), finish_reason=choice.get("finish_reason", "stop") ) def _convert_message(self, msg: Message) -> Dict: result = {"role": msg.role, "content": msg.content} if msg.tool_call_id: result["tool_call_id"] = msg.tool_call_id if msg.tool_calls: result["tool_calls"] = msg.tool_calls return result

这里用了tenacity库做重试，设置了指数退避。注意重试只针对网络错误和 5xx 错误，4xx 错误不应该重试。实际项目中要细化异常处理逻辑。

4.3 工具层实现：注册机制与执行隔离

工具层的核心是注册机制。我用装饰器来实现：

from typing import Callable, Dict, Any from pydantic import BaseModel, ValidationError import asyncio class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict] = {} def register(self, name: str, description: str, params_schema: type[BaseModel]): def decorator(func: Callable): self._tools[name] = { "name": name, "description": description, "params_schema": params_schema, "func": func } return func return decorator def get_tool_definitions(self) -> list: return [ { "type": "function", "function": { "name": t["name"], "description": t["description"], "parameters": t["params_schema"].model_json_schema() } } for t in self._tools.values() ] async def execute(self, name: str, arguments: dict, timeout: float = 30.0) -> dict: if name not in self._tools: return {"success": False, "error": f"Tool {name} not found"} tool = self._tools[name] try: params = tool["params_schema"](**arguments) except ValidationError as e: return {"success": False, "error": f"Invalid params: {e}"} try: result = await asyncio.wait_for( tool["func"](params), timeout=timeout ) return {"success": True, "data": result} except asyncio.TimeoutError: return {"success": False, "error": f"Tool {name} timeout after {timeout}s"} except Exception as e: return {"success": False, "error": f"Tool {name} failed: {str(e)}"}

使用示例：

registry = ToolRegistry() class SearchParams(BaseModel): query: str max_results: int = 5 @registry.register( name="web_search", description="搜索互联网获取最新信息", params_schema=SearchParams ) async def web_search(params: SearchParams): # 实际搜索逻辑 return {"results": [...]}

这个设计的好处是：工具定义和实现在一起，新增工具只需要加一个装饰器；参数校验自动完成；执行超时和异常被统一处理，不会影响编排层。

4.4 编排层实现：ReAct 循环与终止条件

编排层的 ReAct 循环实现：

class ReActOrchestrator: def __init__(self, model_client, tool_registry, state_manager, max_iterations=10): self.model = model_client self.tools = tool_registry self.state = state_manager self.max_iterations = max_iterations async def run(self, user_input: str, session_id: str) -> str: messages = await self.state.get_messages(session_id) messages.append(Message(role="user", content=user_input)) tool_definitions = self.tools.get_tool_definitions() for iteration in range(self.max_iterations): request = ModelRequest( messages=messages, tools=tool_definitions if tool_definitions else None ) response = await self.model.chat(request) messages.append(Message( role="assistant", content=response.content, tool_calls=response.tool_calls )) if not response.tool_calls: await self.state.save_messages(session_id, messages) return response.content for tool_call in response.tool_calls: tool_name = tool_call["function"]["name"] arguments = json.loads(tool_call["function"]["arguments"]) result = await self.tools.execute(tool_name, arguments) messages.append(Message( role="tool", content=json.dumps(result, ensure_ascii=False), tool_call_id=tool_call["id"] )) await self.state.save_messages(session_id, messages) return "抱歉，我无法在限定步骤内完成这个任务。"

这里有几个关键点：第一，max_iterations限制了最大循环次数，防止死循环；第二，每次循环都把工具结果追加到消息列表，让模型看到执行结果；第三，如果没有工具调用，说明模型认为任务完成，直接返回；第四，达到最大迭代次数后返回兜底回复。

4.5 状态层实现：短期记忆与长期记忆

短期记忆用 Redis 存储，设置过期时间：

import json import redis.asyncio as redis class ShortTermMemory: def __init__(self, redis_client: redis.Redis, ttl: int = 3600): self.redis = redis_client self.ttl = ttl async def get_messages(self, session_id: str) -> list: key = f"session:{session_id}:messages" data = await self.redis.get(key) if not data: return [] messages = json.loads(data) return [Message(**m) for m in messages] async def save_messages(self, session_id: str, messages: list): key = f"session:{session_id}:messages" data = json.dumps([m.__dict__ for m in messages], ensure_ascii=False) await self.redis.setex(key, self.ttl, data)

长期记忆用向量数据库存储，检索时根据语义相似度召回：

class LongTermMemory: def __init__(self, vector_store, embedding_client): self.vector_store = vector_store self.embedding = embedding_client async def save(self, user_id: str, content: str, metadata: dict): vector = await self.embedding.embed(content) await self.vector_store.upsert( id=f"{user_id}:{hash(content)}", vector=vector, metadata={"user_id": user_id, "content": content, **metadata} ) async def recall(self, user_id: str, query: str, top_k: int = 3) -> list: vector = await self.embedding.embed(query) results = await self.vector_store.search( vector=vector, filter={"user_id": user_id}, top_k=top_k ) return [r.metadata["content"] for r in results]

长期记忆的写入时机很关键。我的经验是：在对话结束后，用模型对整段对话做一次总结，提取值得长期保存的信息，然后写入。不要每轮对话都写入，否则噪声太大。

5. 常见问题与排查技巧实录

5.1 模型输出格式不稳定怎么办

这是最常见的问题。模型有时候返回 JSON，有时候返回 Markdown，有时候夹杂解释性文字。解决方案有三层：第一，在提示词中明确要求输出格式，并给出示例；第二，使用模型的结构化输出功能（如果支持）；第三，在模型层做后处理，用正则表达式提取关键信息，提取失败时触发重试。

我通常会在模型层加一个parse_response方法，尝试多种解析策略。如果都失败，就把原始输出返回给编排层，让编排层决定是重试还是报错。重试时可以在提示词中追加“请严格按照 JSON 格式输出”的强调。

5.2 工具调用参数错误怎么处理

模型生成的工具调用参数经常有误，比如字段名拼错、类型不对、缺少必填项。工具层的参数校验会捕获这些错误，返回标准化的错误结果。编排层把错误结果追加到消息列表，模型看到错误后通常会自行修正。如果连续多次参数错误，可以触发人工介入或者返回兜底回复。

这里有一个技巧：在工具描述中把参数 schema 写清楚，包括每个参数的类型、含义、是否必填、示例值。模型看到清晰的 schema，生成正确参数的概率会大大提高。

5.3 上下文窗口溢出怎么解决

对话轮次多了之后，消息列表会超出模型的上下文窗口限制。解决方案是滑动窗口 + 摘要压缩。保留最近 N 轮对话的完整内容，更早的对话用模型做摘要，把摘要作为一条 system 消息放在最前面。摘要的提示词要明确要求保留关键信息，比如用户偏好、已确认的事实、未完成的任务等。

另一个技巧是工具结果的截断。有些工具返回的结果很长，比如搜索返回了十篇文章的全文。可以在工具层做截断，只保留前 N 个字符，或者只保留摘要。这样能显著减少 token 消耗。

5.4 常见问题速查表

5.5 独家避坑技巧

第一个技巧：在编排层加一个“思考日志”。每次模型调用和工具调用都记录到日志里，包括输入、输出、耗时、token 消耗。这样排查问题时可以完整回放 Agent 的决策过程。我通常用结构化日志，方便后续做分析和监控。

第二个技巧：给工具调用加“幂等性”设计。有些工具调用是有副作用的，比如发送邮件、创建订单。如果因为重试导致重复调用，会产生严重后果。解决方案是给每个工具调用生成一个唯一 ID，工具实现方根据 ID 做幂等处理。

第三个技巧：模型层做“降级策略”。当主模型服务不可用时，自动切换到备用模型。备用模型可以是更便宜的、更稳定的，虽然效果差一点，但保证服务可用。降级策略要配置在模型层，对编排层透明。

第四个技巧：状态层做“快照”。每隔几轮对话，把当前状态做一个快照存储。如果后续对话出现问题，可以回滚到快照点重新开始。这在调试复杂问题时非常有用。

6. 分层交付的部署与迭代策略

6.1 各层的独立部署方案

分层交付的一个核心优势是各层可以独立部署。接入层可以用 Nginx 或 API Gateway 做负载均衡；编排层可以水平扩展多个实例，用消息队列做异步任务；模型层可以独立部署，配置多个模型服务的连接；工具层可以拆分成微服务，每个工具独立部署；状态层用 Redis 集群或数据库集群保证高可用。

实际部署时，我建议先单体部署，再逐步拆分。一开始所有层打包在一个服务里，通过模块化保证分层清晰。当某一层成为瓶颈时，再把它拆出来独立部署。不要一开始就搞微服务，那样运维成本太高。

6.2 分层迭代的版本管理

每一层都应该有独立的版本号。接入层的 API 版本用 URL 路径区分，比如/v1/chat和/v2/chat。编排层的策略版本用配置管理，可以动态切换。模型层的适配器版本跟随模型服务版本。工具层的工具版本用注册时的元数据标记。状态层的存储 schema 版本用迁移脚本管理。

版本管理的核心原则是向后兼容。新增字段可以，删除字段要谨慎；新增工具可以，修改工具参数要评估影响；新增模型可以，切换模型要做好灰度。我通常会在接入层做版本路由，根据请求头或用户配置，把请求路由到不同版本的编排层。

6.3 监控与可观测性建设

分层之后，监控也要分层做。接入层监控 QPS、延迟、错误率；编排层监控循环次数、工具调用次数、任务完成率；模型层监控调用次数、token 消耗、响应时间、错误率；工具层监控调用次数、成功率、执行时间；状态层监控读写次数、命中率、存储容量。

除了这些常规指标，还要做链路追踪。给每个请求生成一个 trace_id，在层间传递，这样可以在日志系统里串联起整个调用链。我通常用 OpenTelemetry 做链路追踪，配合 Jaeger 或 Zipkin 做可视化。

6.4 成本控制的分层策略

成本控制也要分层做。接入层做限流，防止恶意请求；编排层做循环控制，防止死循环；模型层做模型路由，简单问题用小模型；工具层做结果缓存，相同查询不重复调用；状态层做数据清理，过期数据及时删除。

我算过一笔账：一个中等复杂度的 Agent 任务，如果不做任何优化，token 消耗可能在 10000 左右；做了模型路由和结果缓存后，可以降到 3000 左右；再做上下文压缩和工具结果截断，可以降到 1500 左右。成本降低了 85%，效果基本不变。

7. 从分层交付到工程化落地

分层交付不是目的，而是手段。它的最终目标是让 AI Agent 从“玩具”变成“产品”，从“Demo”变成“生产系统”。我见过太多团队在 Demo 阶段信心满满，一到生产环境就各种问题。分层交付是解决这些问题的第一步，也是最关键的一步。

这套分层方案我在多个项目中实践过，包括客服 Agent、数据分析 Agent、代码助手 Agent 等。每次实践都会根据具体场景做调整，但核心思路不变：职责分离、接口明确、独立测试、独立部署。如果你正在做 AI Agent 的工程化落地，我建议你先从分层开始，把架构搭好，后面的路会顺很多。

最后分享一个我在实际项目中的体会：分层不是越细越好。我见过一个项目分了十几层，结果层间调用比业务逻辑还复杂。分层的粒度要适中，通常五层左右就够了。关键是每层要有明确的职责边界，层间通信要简单直接。如果发现某一层特别薄，只有几行代码，那可能不需要单独分层，合并到相邻层即可。架构是演进来的，不是设计出来的，先跑起来，再优化。