AI应用架构设计图解：从接入层到模型层的四层架构与Agent编排实战

1. 从一张架构图说起：AI应用到底在搭什么

很多人第一次接触AI应用开发，脑子里冒出来的第一个问题不是“怎么写代码”，而是“这东西到底长什么样”。你去看市面上的技术分享，满屏都是Agent、LLM、MCP、RAG、Tool Calling这些词，单独拎出来每个都认识，拼在一起就不知道从哪下手了。我刚开始做AI应用的时候也有这个困惑，后来画了几十张架构图、拆了十几个开源项目之后才慢慢理清楚——AI应用的架构设计，本质上就是在回答一个问题：一个用户请求进来，经过哪些环节，最终变成一个靠谱的答案返回去。

这件事听起来简单，但真正落地的时候你会发现，难点根本不在“调模型”这一步。调API谁都会，三行代码就能跑通。难的是当你的应用要面对真实用户、真实数据、真实并发的时候，整个链路的每一个环节都可能出问题。模型返回格式不对怎么办？用户问的问题需要查数据库怎么接？多个Agent之间怎么协作？上下文太长超了token限制怎么处理？这些问题才是架构设计真正要解决的东西。

这篇内容我打算用图解的思路，把AI应用架构从最外层到最里层拆一遍。不是那种画个框框写个“LLM”就完事的示意图，而是把每一层的职责、每一层之间的数据流、每一层容易踩的坑都讲清楚。适合正在做AI应用开发的程序员、正在从传统后端往AI方向转的工程师，以及想搞清楚Agent和LLM到底怎么配合的产品同学。看完之后你至少能做到两件事：第一，拿到一个AI应用需求，能画出它的架构分层；第二，知道每一层该用什么技术方案，以及为什么这么选。

2. 拆开AI应用的骨架：四层架构与数据流向

2.1 为什么AI应用不能只有“一个模型调用”

先说什么叫“只有一个模型调用”的架构。最原始的做法是：前端发一个请求过来，后端直接把用户输入拼到prompt里，调一次LLM的API，拿到结果返回给前端。这个架构在demo阶段完全够用，我早期做内部工具的时候就是这么干的，一天就能上线。

但这个东西一旦要产品化，问题就全冒出来了。用户问“帮我查一下上个月的销售数据”，模型不知道你的数据库里有什么，它只能瞎编。用户连续问了五个问题，每个问题都依赖前面的上下文，但你的API调用是无状态的，每次都是全新的对话。用户上传了一个PDF让你总结，PDF有五十页，直接塞进prompt里token直接爆了。更别说多个用户同时用的时候，你怎么管理会话、怎么做限流、怎么做缓存。

所以真实的AI应用架构，一定是一个分层的结构。我把它拆成四层：接入层、编排层、能力层、模型层。这四层不是我拍脑袋分的，而是从实际项目中反复验证出来的——每一层解决一类特定问题，层与层之间通过明确定义的接口通信，任何一层换实现都不影响其他层。

2.2 四层架构各自的职责边界

接入层负责的是和用户打交道。它处理的是HTTP请求、WebSocket连接、会话管理、鉴权、限流这些传统后端的事情。这一层不需要懂AI，它只需要知道“有一个请求进来了，我要把它转成内部格式，然后交给下一层”。很多从传统后端转过来的同学在这一层有天然优势，因为这就是你们平时写的东西。

编排层是整个AI应用的大脑。它决定了一个请求进来之后，要经过哪些步骤、调用哪些工具、是否需要多轮推理。Agent的逻辑主要就活在这一层。比如用户问“帮我分析一下这份财报”，编排层会决定：先调用文档解析工具把PDF转成文本，然后调用LLM做摘要，再调用LLM做关键指标提取，最后把结果组装成结构化输出。这一层是AI应用和传统应用最大的区别所在。

能力层是编排层可以调用的“工具箱”。这里面包括：向量检索（RAG）、数据库查询、外部API调用、代码执行、文件处理等等。编排层说“我需要查一下知识库”，能力层就去执行向量检索；编排层说“我需要调一下天气API”，能力层就去发HTTP请求。这一层的关键是标准化——每个工具都有统一的输入输出格式，编排层不需要知道工具内部怎么实现的。

模型层就是LLM本身，以及围绕LLM的一些基础设施，比如模型路由（根据任务类型选择不同模型）、token计数、缓存、重试等等。这一层要解决的是“怎么稳定、高效地调用模型”这个问题。

2.3 一次完整请求的数据流拆解

光说分层太抽象，我们跟着一个具体请求走一遍。假设用户在前端输入：“帮我对比一下我们公司和竞争对手上季度的营收情况。”

请求到达接入层，首先做鉴权，确认这个用户有权限访问财务数据。然后从会话存储中取出这个用户的历史对话上下文。接着做限流检查，确认没有超过配额。最后把请求包装成一个内部消息格式，扔给编排层。

编排层收到消息后，开始做任务规划。它先判断这是一个需要多步推理的任务，然后拆解成子任务：第一步，查询本公司上季度营收；第二步，查询竞争对手上季度营收；第三步，对比分析。编排层发现第一步和第二步都需要访问数据库，于是向能力层发起工具调用请求。

能力层收到“查询本公司上季度营收”的请求，把它翻译成SQL语句，去数据库执行，拿到结果后返回给编排层。同样的流程处理竞争对手的数据。

编排层拿到两份数据后，构造一个prompt，把数据塞进去，调用模型层的LLM接口。模型层负责选择合适的模型（这种分析任务用大一点的模型效果更好），管理token预算，执行调用，返回结果。

最后结果沿着原路返回：模型层→编排层→接入层→前端。接入层把这次对话存入会话历史，供下一轮使用。

这个流程看起来简单，但每一层都有大量的细节要处理。下面我逐层拆开讲。

3. 编排层：Agent逻辑到底怎么写才不乱

3.1 从“if-else”到“任务规划”的思维转变

很多人写Agent的第一反应是写一堆if-else。用户问天气就调天气API，用户问股票就调股票API。这种做法在意图识别阶段确实有用，但一旦任务复杂起来就完全不够用了。因为真实用户的问题往往不是单一意图，而是多个意图的组合，甚至包含模型才能理解的隐含意图。

编排层的核心思路是任务规划：把用户的自然语言请求，转化成一个可执行的任务序列。这个转化过程可以基于规则，也可以基于LLM。我实测下来，纯规则的方式适合意图非常明确的场景（比如客服机器人的固定问答），而基于LLM的规划方式适合开放式任务。

基于LLM的规划怎么做？简单说就是给LLM一个“工具清单”，让它输出一个JSON格式的任务计划。比如：

{ "plan": [ {"step": 1, "action": "query_database", "params": {"table": "revenue", "company": "self", "quarter": "last"}}, {"step": 2, "action": "query_database", "params": {"table": "revenue", "company": "competitor", "quarter": "last"}}, {"step": 3, "action": "llm_analyze", "params": {"template": "comparison", "inputs": ["step1_result", "step2_result"]}} ] }

编排层拿到这个计划后，按顺序执行，把每一步的结果传给下一步。这种做法比if-else灵活得多，因为新增一个工具只需要在工具清单里加一行描述，不需要改代码逻辑。

3.2 ReAct模式与Plan-Execute模式的选择

目前主流的Agent编排模式有两种：ReAct和Plan-Execute。

ReAct是“推理-行动”交替进行。模型先想一步（Thought），然后做一个动作（Action），观察结果（Observation），再想下一步，再行动。这种方式的好处是灵活，模型可以根据中间结果动态调整策略。坏处是调用次数多，延迟高，而且容易陷入循环。

Plan-Execute是先制定完整计划，再一次性执行。好处是效率高，调用次数少。坏处是如果中间某一步的结果和预期不符，整个计划可能就废了。

我的经验是：任务步骤少于5步、步骤之间依赖关系明确的，用Plan-Execute；任务复杂、需要根据中间结果动态调整的，用ReAct。实际项目中我经常混用——先用Plan-Execute做粗粒度规划，每个步骤内部再用ReAct做细粒度执行。

3.3 上下文管理：别让对话历史撑爆你的token

编排层还有一个容易被忽视但极其重要的职责：上下文管理。多轮对话的场景下，如果把所有历史消息都塞进prompt，token消耗会线性增长，成本扛不住，而且模型对超长上下文的注意力也会下降。

常见的做法有三种。第一种是滑动窗口，只保留最近N轮对话。简单粗暴，但会丢失早期的重要信息。第二种是摘要压缩，把早期对话用LLM总结成一段摘要，和最近几轮对话一起塞进去。这种方式保留了关键信息，但增加了一次LLM调用。第三种是向量检索，把所有历史对话存入向量库，每次根据当前问题检索最相关的几条历史记录。这种方式最灵活，但实现复杂度也最高。

我一般用组合方案：最近3轮对话保留原文，3轮之前的对话做摘要压缩，同时把关键实体（人名、数字、日期）单独提取出来放在prompt的固定位置。这样既控制了token量，又不会丢失关键信息。

提示：上下文管理策略一定要在项目早期就设计好，后期再改会牵涉到会话存储结构、prompt模板、缓存策略的全面调整，成本极高。

4. 能力层：工具调用与MCP协议的实际落地

4.1 工具调用的本质是“结构化输出”

能力层最核心的事情就是工具调用（Tool Calling）。很多人觉得工具调用很神秘，其实它的本质就是让LLM输出一段结构化的JSON，你的代码解析这个JSON然后执行对应的函数。

比如你定义了一个查询天气的工具：

tools = [ { "name": "get_weather", "description": "查询指定城市的天气", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"}, "date": {"type": "string", "description": "日期，格式YYYY-MM-DD"} }, "required": ["city"] } } ]

当用户问“北京明天天气怎么样”，模型会输出：

{"name": "get_weather", "arguments": {"city": "北京", "date": "2025-01-16"}}

你的代码拿到这个JSON，调用实际的天气API，把结果返回给模型，模型再生成自然语言回复。整个流程没有任何魔法，就是“模型输出JSON→代码执行→结果回传”这个循环。

4.2 MCP协议解决了什么实际问题

MCP（Model Context Protocol）是最近很火的一个概念。它要解决的问题是：工具定义和工具实现之间的耦合。

在没有MCP之前，每个AI应用都要自己定义工具清单，自己实现工具函数。你想换一个AI框架，工具部分要重写。你想让多个AI应用共享同一套工具，得复制粘贴。MCP的做法是把工具的定义和实现抽出来，做成一个独立的服务，通过标准协议对外暴露。任何支持MCP的AI应用都可以连接这个服务，自动发现可用的工具，直接调用。

你可以把MCP理解成“AI工具界的USB接口”。以前每个设备有自己的充电口，现在统一成USB-C，谁都能插。实际落地的时候，你可以把数据库查询、文件处理、API调用这些通用能力做成MCP Server，然后你的Agent作为MCP Client去连接。这样你的Agent代码里就不需要写任何工具实现，只需要写编排逻辑。

目前MCP的生态还在快速发展中，支持MCP的框架和工具越来越多。我的建议是：新项目可以直接上MCP，老项目如果工具不多可以先不动，等MCP生态更成熟了再迁移。

4.3 RAG不是“向量检索”四个字就完事了

RAG（检索增强生成）是能力层里最常用的工具，但也是最容易做砸的。很多人以为RAG就是“把文档切块→向量化→存向量库→检索→塞进prompt”，跑通demo觉得效果不错，一上真实数据就发现检索出来的东西驴唇不对马嘴。

问题出在切块策略上。固定长度切块是最简单的做法，但效果往往最差。因为一个完整的语义单元可能被切断了，检索出来的片段缺少上下文。我一般用语义切块：按段落、按标题、按句子边界来切，保证每个块是一个完整的语义单元。如果文档结构复杂，还会用LLM来做智能切块——让模型判断哪里是自然的切分点。

另一个坑是检索策略。纯向量检索适合语义相似度匹配，但对关键词精确匹配不擅长。用户搜“2024年Q3营收”，向量检索可能返回一堆“营收”相关的段落，但漏掉精确包含“2024年Q3”的那一段。所以实际项目中我一般用混合检索：向量检索+关键词检索，两路结果合并后做重排序（Rerank）。重排序用一个小的交叉编码器模型来做，效果比单纯按向量距离排序好很多。

5. 模型层：LLM选型、路由与成本控制

5.1 不同任务用不同模型，别拿大炮打蚊子

模型层最容易被忽视的优化点就是模型路由。很多项目从头到尾就用一个模型，要么全用最贵的，要么全用最便宜的。前者成本爆炸，后者效果拉胯。

正确的做法是根据任务类型选择模型。我一般把任务分成三档：

模型路由的实现方式有两种。一种是静态路由，在代码里写死“这个任务用哪个模型”。另一种是动态路由，用一个小的分类模型先判断任务难度，再决定用哪个模型。静态路由简单可靠，动态路由更灵活但增加了一次调用。我一般先用静态路由，等任务类型稳定了再考虑动态路由。

5.2 Token成本的精算与优化

Token成本是AI应用绕不开的话题。我见过太多项目上线第一个月账单出来才发现成本失控的。控制成本的核心是精算每一处token消耗。

先算一笔账。假设你的应用每天处理10000个请求，每个请求平均输入2000 token、输出500 token。用中等价位的模型，输入$0.5/百万token、输出$1.5/百万token来算：

• 每日输入成本：10000 × 2000 / 1000000 × 0.5 = $10
• 每日输出成本：10000 × 500 / 1000000 × 1.5 = $7.5
• 每日总成本：$17.5
• 每月成本：约$525

这还只是一个中等规模的应用。如果输入长度翻倍、请求量翻倍，成本直接四倍。所以优化token消耗是必须做的事。

优化手段有几个。Prompt压缩：把冗余的指令精简掉，把few-shot示例从5个减到2个。缓存：相同的输入直接返回缓存结果，不重复调用模型。分级处理：简单请求走小模型，复杂请求才走大模型。输出限制：在prompt里明确要求“简洁回答”，设置max_tokens上限。

注意：缓存策略要小心处理。如果缓存粒度太粗，不同用户的请求可能互相污染；如果缓存粒度太细，命中率又太低。我一般按“用户ID+请求类型+关键参数”来做缓存key。

5.3 模型调用的稳定性设计

LLM API不是100%可靠的。超时、限流、返回格式错误、服务不可用，这些情况在生产环境都会遇到。模型层必须做好稳定性设计。

重试机制是基础。但重试不能无脑重试，要区分错误类型。超时和限流可以重试，参数错误重试也没用。重试次数一般2-3次，每次间隔用指数退避。

降级策略是保险。主模型不可用时，自动切换到备用模型。备用模型可以选同级别的另一个模型，也可以降级到小模型保证基本可用。

格式校验是兜底。模型返回的JSON可能不合法，可能缺少必填字段。拿到结果后一定要做schema校验，校验不通过就触发重试或降级。

超时控制是必须。LLM调用可能卡住很久，必须设置合理的超时时间。一般简单任务10-30秒，复杂任务60-120秒。超时后要么重试，要么返回兜底话术。

6. 踩坑实录：那些架构图上看不到的坑

6.1 并发场景下的会话状态管理

单用户测试的时候一切正常，一上并发就出问题。最常见的问题是会话状态串了。用户A的对话历史被用户B看到了，或者同一个用户的多个请求互相覆盖了上下文。

根因在于会话存储的设计。如果你把会话状态存在内存里，多实例部署的时候就会出问题——请求被负载均衡到不同实例，每个实例的内存里只有部分会话。解决方案是把会话状态外置到Redis或数据库中，每个请求根据session_id去取对应的上下文。

另一个并发问题是同一会话的并发请求。用户快速发了三条消息，三个请求同时到达。如果每个请求都去读会话历史、追加新消息、写回，就会出现覆盖。解决方案是加分布式锁，按session_id加锁，保证同一会话的请求串行处理。

6.2 工具调用的“幻觉参数”

模型在调用工具的时候，经常会编造一些不存在的参数，或者参数格式不对。比如你定义的工具只接受city和date两个参数，模型偏偏传了一个province进来。或者你要求日期格式是YYYY-MM-DD，模型传了明天。

这个问题没有完美的解决方案，但可以缓解。第一，在工具描述里把参数约束写清楚，包括格式、枚举值、必填项。第二，在代码里做严格的参数校验，不合法的参数直接拒绝并返回错误信息给模型，让模型重新生成。第三，对于关键参数，可以在prompt里加few-shot示例，展示正确的参数格式。

我实测下来，参数校验+错误回传这个组合能解决80%以上的幻觉参数问题。模型拿到错误信息后，大部分情况下能自我纠正。

6.3 长对话中的“中间遗忘”

多轮对话超过一定轮数后，模型会开始“遗忘”中间的内容。你明明在第三轮告诉了它一个关键信息，到第十轮它就不记得了。这不是模型的问题，而是注意力机制的特性——上下文太长的时候，中间部分的信息容易被稀释。

解决方案除了前面说的上下文管理策略之外，还有一个技巧：关键信息前置。把最重要的系统指令、用户偏好、关键实体放在prompt的最前面，而不是埋在对话历史中间。另外，可以在每轮对话开始时，用一句话重申关键上下文，比如“继续之前关于XX项目的讨论”。

6.4 流式输出的边界情况

流式输出（Streaming）是提升用户体验的重要手段，但它的边界情况比非流式多得多。比如：流到一半网络断了怎么办？流式输出的内容需要做后处理怎么办？多个工具调用的结果需要合并后再流式输出怎么办？

我的经验是：流式输出只用于最终的自然语言回复，工具调用和中间推理不走流式。这样可以把复杂性控制在最小范围。对于流式中断的情况，前端要做好重连和状态恢复。对于需要后处理的内容，先缓冲完整结果，处理完再一次性输出，不要边流边处理。

7. 从架构图到代码：一个最小可运行示例

7.1 项目结构设计

说了这么多理论，最后给一个可以直接跑的最小示例。这个示例实现了“用户提问→Agent规划→调用工具→LLM生成回答”的完整链路。

项目结构如下：

ai-app/ ├── main.py # 入口，FastAPI应用 ├── orchestrator.py # 编排层：任务规划与执行 ├── tools/ │ ├── __init__.py │ ├── registry.py # 工具注册中心 │ ├── database.py # 数据库查询工具 │ └── search.py # 向量检索工具 ├── llm/ │ ├── __init__.py │ ├── client.py # LLM客户端封装 │ └── router.py # 模型路由 ├── session/ │ ├── __init__.py │ └── manager.py # 会话管理 └── config.py # 配置

7.2 核心模块的代码实现

先看工具注册中心，这是能力层的核心：

# tools/registry.py from typing import Callable, Any import json class ToolRegistry: def __init__(self): self._tools = {} def register(self, name: str, description: str, parameters: dict): def decorator(func: Callable): self._tools[name] = { "function": func, "schema": { "name": name, "description": description, "parameters": parameters } } return func return decorator def get_schemas(self) -> list: return [t["schema"] for t in self._tools.values()] def execute(self, name: str, arguments: dict) -> Any: if name not in self._tools: raise ValueError(f"未知工具: {name}") return self._tools[name]["function"](**arguments) registry = ToolRegistry() @registry.register( name="query_revenue", description="查询指定公司指定季度的营收数据", parameters={ "type": "object", "properties": { "company": {"type": "string", "description": "公司名称"}, "quarter": {"type": "string", "description": "季度，格式如2024Q3"} }, "required": ["company", "quarter"] } ) def query_revenue(company: str, quarter: str) -> dict: # 实际项目中这里查数据库 return {"company": company, "quarter": quarter, "revenue": 12500000}

再看编排层的核心逻辑：

# orchestrator.py import json from tools.registry import registry from llm.client import LLMClient class Orchestrator: def __init__(self, llm_client: LLMClient): self.llm = llm_client self.max_steps = 5 def run(self, user_input: str, history: list) -> str: # 第一步：任务规划 plan = self._plan(user_input, history) # 第二步：执行计划 results = [] for step in plan["steps"]: if step["action"] == "tool_call": result = registry.execute( step["tool_name"], step["arguments"] ) results.append(result) elif step["action"] == "llm_generate": result = self.llm.generate( prompt=step["prompt"], context=results ) results.append(result) # 第三步：生成最终回答 return self._final_answer(user_input, results) def _plan(self, user_input: str, history: list) -> dict: tools_desc = json.dumps(registry.get_schemas(), ensure_ascii=False) prompt = f"""你是一个任务规划器。根据用户输入，制定执行计划。 可用工具：{tools_desc} 用户输入：{user_input} 输出JSON格式的计划，包含steps数组。每个step有action字段， action为tool_call时包含tool_name和arguments， action为llm_generate时包含prompt。""" response = self.llm.generate(prompt) return json.loads(response) def _final_answer(self, user_input: str, results: list) -> str: prompt = f"""根据以下信息回答用户问题。 用户问题：{user_input} 执行结果：{json.dumps(results, ensure_ascii=False)} 请用简洁的自然语言回答。""" return self.llm.generate(prompt)

7.3 跑通之后的扩展方向

这个最小示例跑通之后，你可以按需扩展。加工具：在registry里注册新的工具函数就行，编排层不需要改。加模型路由：在LLMClient里根据任务类型选择不同模型。加缓存：在LLMClient的generate方法里加一层缓存查询。加监控：在编排层的每个步骤前后打点，记录耗时和token消耗。

架构设计的价值就在于：每个扩展点都是独立的，改一个地方不会牵一发而动全身。这也是为什么我一开始就强调分层——分层不是为了好看，而是为了在需求变化的时候，你只需要改一层，其他层不动。

8. 一些关于架构演进的个人体会

我做AI应用这两年多，最大的体会是：架构不是一开始就设计完美的，而是随着对业务理解的深入逐步演进的。第一个版本可能就是一个简单的prompt调用，第二个版本加了RAG，第三个版本引入了Agent编排，第四个版本才做了模型路由和成本优化。每一步演进都是被真实问题驱动的，而不是为了架构而架构。

另一个体会是：不要过度设计。我见过一些项目，一开始就上多Agent协作、上复杂的规划器、上全套的监控体系，结果开发周期拉长了好几倍，上线后发现用户根本用不到那些复杂功能。先用最简单的方案跑通核心链路，拿到真实用户反馈，再针对性地优化，这个节奏更稳。

最后一个建议：把架构图画出来，贴在团队所有人都能看到的地方。不是为了好看，而是为了让每个人都知道自己的代码在整个系统中的位置，知道自己的模块和谁交互、通过什么接口交互。很多协作问题，本质上都是因为大家对架构的理解不一致。一张清晰的架构图，能省掉很多沟通成本。