1. 从一张架构图说起AI应用到底该怎么搭这两年跟不少团队聊过AI应用落地的事发现一个特别普遍的现象大家Demo跑得飞快一到生产环境就各种问题。模型调用超时、上下文爆炸、工具调用乱套、多轮对话状态丢失这些坑几乎每个团队都会踩一遍。说到底不是模型不行而是架构设计没跟上。我画过不下几十张AI应用的架构图从最简单的单模型调用到带RAG的问答系统再到多Agent协作的复杂编排每一层该放什么、边界怎么划、数据怎么流这些想不清楚后面写代码就是给自己挖坑。这篇内容就是把我自己从零搭AI应用时积累的架构思路完整拆一遍从最基础的LLM调用层到Agent编排层再到MCP工具协议层每一层为什么这么设计、怎么落地、容易踩什么坑都会讲到。不管你是刚接触AI应用开发的程序员还是已经在做Agent项目但总觉得架构别扭的工程师或者是从运维转过来想搞明白AI应用怎么扛并发的朋友这篇内容应该都能给你一些可以直接抄作业的思路。我不会只讲概念每个架构决策背后我都会说清楚为什么这么选以及实际写代码时怎么落地。2. AI应用架构的分层逻辑与核心选型2.1 为什么AI应用需要分层架构传统Web应用的分层大家都很熟了Controller-Service-DAO三层走天下。但AI应用完全不是这个逻辑。你如果拿传统MVC的思路去套AI应用很快就会发现不对劲——因为AI应用里有一个不确定性的核心LLM。传统业务代码的输入输出是确定的你传个userId进去返回一个User对象逻辑路径清晰。但LLM不一样同样的prompt两次调用可能返回完全不同的结果。这就导致AI应用的架构必须围绕“如何管理不确定性”来设计。我习惯把AI应用分成四层来看模型接入层负责跟各种LLM打交道处理API调用、重试、降级、限流能力增强层RAG检索、记忆管理、上下文压缩让模型能“知道更多”编排调度层Agent逻辑、工具调用、多步推理链让模型能“做更多”应用交互层对话管理、流式输出、前端渲染让用户能“用得舒服”这四层不是必须全部都有但每多一层你的应用能力就上一个台阶同时复杂度也上一个台阶。关键是根据你的实际场景做取舍。2.2 LLM选型别一上来就盯着最大的模型选LLM模型这件事我见过太多团队一上来就说“我们要用最强的模型”。结果一算成本发现根本扛不住。LLM选型的核心不是选最强的而是选最合适的。我一般从三个维度来评估维度关键问题典型考量能力任务需要多强的推理能力简单分类用小模型复杂推理用大模型成本每次调用的token成本高频调用场景必须算清楚单次成本延迟用户能接受多长的等待实时对话要求首token延迟低于1秒实际项目中我经常采用的是混合策略简单意图识别用7B级别的小模型复杂推理用大模型这样整体成本能降下来不少。比如一个客服场景80%的问题都是简单查询只有20%需要复杂推理那混合策略就能省掉大量成本。还有一个容易被忽略的点是模型降级。你的主模型如果挂了或者超时了有没有备用模型顶上这个在架构设计阶段就要考虑不能等出事了再补。2.3 Agent架构从ReAct到多Agent协作Agent是这两年AI应用最热的方向但很多人对Agent的理解还停留在“让模型调用工具”这个层面。实际上Agent架构有好几种模式选错了模式后面越写越别扭。最基础的是ReAct模式就是Reasoning Acting的循环模型先思考需要做什么然后调用工具拿到结果后再思考下一步。这个模式适合任务步骤不太确定的场景比如“帮我查一下最近的订单并总结”。再往上是Plan-and-Execute模式先让模型制定一个完整的计划然后按计划逐步执行。这个适合步骤明确但比较复杂的任务比如“帮我分析这份财报并生成报告”。最复杂的是多Agent协作模式多个Agent各司其职有的负责规划有的负责执行有的负责审核。这个模式适合大型复杂任务但调试难度也最大。我的经验是能用单Agent解决的绝不上多Agent。多Agent的通信开销、状态同步、错误传播都是大问题。很多团队一上来就搞多Agent结果发现调试成本远超预期。2.4 MCP协议工具调用的标准化尝试MCPModel Context Protocol是最近很火的一个话题本质上它想解决的是工具调用的标准化问题。在没有MCP之前每个AI应用接入工具都是自己定义一套接口。你想让模型调用数据库得自己写一套想让它调用文件系统又得写一套。不同应用之间的工具没法复用换个框架就得重写。MCP的思路是定义一套标准协议工具提供方按这个协议暴露能力AI应用按这个协议调用工具。这样理论上工具就能跨应用复用了。但实际落地中MCP还有一些不成熟的地方。比如授权机制、流式输出的处理、错误码的标准化这些在不同实现里差异还挺大的。我试过用MCP接入一些工具有些场景确实方便但有些场景还不如自己写个函数调用来得直接。我的建议是MCP适合工具生态比较丰富的场景如果你就接两三个工具自己写函数调用反而更简单可控。但如果你要接入大量第三方工具MCP的标准化价值就体现出来了。3. 核心模块的架构细节与实操要点3.1 模型接入层重试、降级、限流一个都不能少模型接入层看起来简单不就是调个API吗但实际生产环境中这一层要处理的问题非常多。重试策略是第一个要设计的。LLM API调用失败是常态网络抖动、服务限流、模型过载都会导致失败。但重试不能无脑重试要区分错误类型网络超时可以重试但要有退避策略限流错误要等一段时间再重试不能立即重试参数错误重试没用直接报错我一般用指数退避加抖动的策略第一次等1秒第二次等2秒第三次等4秒加上随机抖动避免多个请求同时重试。降级策略是第二个要设计的。主模型不可用时要有备用方案。最简单的降级是切换到备用模型复杂一点的降级是切换到缓存结果或者规则引擎。限流是第三个要设计的。不是限制用户而是限制你自己的调用频率。很多LLM API都有QPS限制你如果不管控很容易触发限流导致大面积失败。我一般用令牌桶算法做限流根据API的配额来设置速率。# 一个简化的模型调用封装示例 class LLMClient: def __init__(self, primary_model, fallback_model, qps_limit): self.primary primary_model self.fallback fallback_model self.rate_limiter TokenBucket(qps_limit) async def call(self, prompt, max_retries3): for attempt in range(max_retries): try: await self.rate_limiter.acquire() return await self.primary.invoke(prompt) except RateLimitError: await asyncio.sleep(2 ** attempt random.random()) except TimeoutError: if attempt max_retries - 1: return await self.fallback.invoke(prompt) await asyncio.sleep(1)注意重试次数不是越多越好。我见过有团队设置重试10次结果一个请求卡了30秒用户体验极差。一般3次重试就够了超过3次还没成功说明问题不是重试能解决的。3.2 RAG检索层切分策略决定检索质量RAG是AI应用里最常用的能力增强手段但很多团队做出来的RAG效果很差核心问题往往出在文档切分上。文档切分看起来简单实际上有很多讲究。切得太碎语义不完整检索出来的片段没有上下文切得太大噪声太多模型抓不住重点。我一般用递归切分策略先按段落切如果段落太长再按句子切如果句子还太长再按字符切。同时设置一个重叠窗口保证相邻片段之间有上下文衔接。切分粒度怎么定我的经验是技术文档500-800字符因为技术概念通常需要一段话才能说清楚对话记录按轮次切每轮对话作为一个单元法律合同按条款切每个条款独立还有一个容易被忽略的点是元数据。每个片段除了文本内容还应该带上来源、章节、时间等元数据。这样检索的时候可以按元数据过滤提高准确率。检索策略上我一般用混合检索向量检索加关键词检索两路结果合并后重排序。纯向量检索对精确匹配不敏感比如你搜一个具体的错误码向量检索可能找不到但关键词检索能精确命中。3.3 记忆管理短期记忆和长期记忆要分开设计AI应用如果没有记忆每次对话都是从头开始用户体验很差。但记忆管理不是简单地把历史对话都塞进上下文那样很快就会超出token限制。我的做法是把记忆分成两层短期记忆是当前会话的上下文通常保留最近N轮对话。N的取值取决于你的token预算和对话特点。我一般设置最近10轮超过的做摘要压缩。长期记忆是跨会话的用户信息比如用户的偏好、历史问题、重要事实。这些信息存在数据库里每次对话开始时检索相关部分注入上下文。摘要压缩是短期记忆管理的关键。当对话轮次超过阈值时让模型对早期对话做摘要用摘要替代原始对话。这样既保留了关键信息又控制了token消耗。# 记忆管理的简化逻辑 class MemoryManager: def __init__(self, max_turns10, summary_threshold15): self.short_term [] self.long_term {} self.max_turns max_turns self.summary_threshold summary_threshold def add_turn(self, user_msg, assistant_msg): self.short_term.append({user: user_msg, assistant: assistant_msg}) if len(self.short_term) self.summary_threshold: self._compress() def _compress(self): old_turns self.short_term[:-self.max_turns] summary llm.summarize(old_turns) self.short_term [{summary: summary}] self.short_term[-self.max_turns:]提示摘要压缩会丢失细节所以重要信息要及时写入长期记忆。我一般让模型在对话过程中主动识别“值得记住的信息”比如用户说“我下周要去北京出差”这个信息就应该写入长期记忆。3.4 Agent编排状态机比自由循环更可控Agent编排最怕的就是模型陷入死循环反复调用同一个工具或者在一个步骤上卡住。我见过有Agent调了20次搜索工具还没给出答案token烧了一大堆用户等得想砸键盘。解决这个问题的关键是用状态机约束Agent的行为。不要让模型完全自由地决定下一步做什么而是预定义好状态和转移条件。比如一个客服Agent状态可以定义为意图识别 - 信息收集 - 方案生成 - 确认执行 - 完成。每个状态下模型能做什么是有限的状态之间的转移也有明确条件。这样即使模型输出不稳定整体流程还是可控的。状态机还有一个好处是可观测性。你能清楚地知道当前在哪个状态卡了多久哪个状态出错率最高。这对排查问题非常重要。多Agent协作的场景下状态机就更重要了。每个Agent有自己的状态机Agent之间的通信通过消息队列或者共享状态来协调。我一般用黑板模式所有Agent共享一个状态板每个Agent读取自己关心的部分写入自己的产出。3.5 流式输出用户体验的关键细节流式输出是AI应用用户体验的关键。用户等10秒才看到完整回复和每秒都在看到文字冒出来感受完全不一样。但流式输出在架构上会带来一些复杂性。首先是错误处理流式输出到一半出错了怎么办我的做法是先把完整结果生成好再流式返回这样出错可以在开始流式之前就发现。但这会增加首token延迟需要权衡。其次是工具调用和流式输出的配合。Agent调用工具的时候工具执行可能需要几秒钟这段时间用户看到什么我一般会输出一个“正在查询...”的提示让用户知道系统在工作。还有一个细节是流式输出的分块策略。不是越小越好太小的块会导致前端渲染频繁重排反而卡顿。我一般按句子或者按固定字符数分块兼顾流畅度和性能。4. 完整实操从零搭建一个AI应用骨架4.1 项目结构设计说了这么多架构原则接下来我带你走一遍完整的搭建过程。以一个知识库问答Agent为例从项目结构开始。ai-app/ ├── config/ │ ├── models.yaml # 模型配置 │ └── agents.yaml # Agent配置 ├── core/ │ ├── llm_client.py # 模型接入层 │ ├── memory.py # 记忆管理 │ └── rate_limiter.py # 限流器 ├── rag/ │ ├── splitter.py # 文档切分 │ ├── embedder.py # 向量化 │ └── retriever.py # 检索器 ├── agent/ │ ├── state_machine.py # 状态机 │ ├── tools.py # 工具定义 │ └── orchestrator.py # 编排器 ├── api/ │ ├── routes.py # API路由 │ └── streaming.py # 流式输出 └── main.py这个结构的好处是职责清晰。模型接入的问题在core里解决检索的问题在rag里解决Agent逻辑在agent里解决。每层可以独立测试和替换。4.2 模型接入层的完整实现模型接入层我一般会定义一个统一的接口屏蔽不同模型提供商的差异。from abc import ABC, abstractmethod class BaseLLM(ABC): abstractmethod async def invoke(self, messages, **kwargs): pass abstractmethod async def stream(self, messages, **kwargs): pass class OpenAIClient(BaseLLM): async def invoke(self, messages, **kwargs): # 具体实现 pass class LocalLLMClient(BaseLLM): async def invoke(self, messages, **kwargs): # 本地模型实现 pass这样上层代码不关心底层用的是哪个模型切换模型只需要改配置。配置我一般用YAML管理方便不同环境切换models: primary: provider: openai model: gpt-4 timeout: 30 max_retries: 3 fallback: provider: local model: qwen-7b timeout: 604.3 RAG检索层的落地细节RAG检索层我踩过最大的坑是向量数据库选型。一开始用FAISS单机跑没问题但数据量大了之后内存扛不住。后来换成Milvus支持分布式但运维复杂度上来了。我的建议是数据量小于100万条用FAISS就够了简单省事。超过100万条再考虑Milvus或者Qdrant这些专业向量数据库。嵌入模型的选择也很关键。OpenAI的embedding模型效果好但需要调API有网络延迟和成本。本地模型比如BGE系列效果也不错而且没有网络依赖。我一般根据场景选对延迟敏感用本地模型对效果要求高用API模型。检索的时候我一般会检索Top-20然后重排序取Top-5注入上下文。为什么要多检索再重排因为向量检索的召回率有限多检索一些能提高命中率重排序能把最相关的排到前面。4.4 Agent状态机的实现Agent状态机我用Python的transitions库来实现比手写状态管理清晰很多。from transitions import Machine class QAAgent: states [idle, retrieving, generating, tool_calling, done] def __init__(self): self.machine Machine( modelself, statesQAAgent.states, initialidle, transitions[ {trigger: start, source: idle, dest: retrieving}, {trigger: retrieve_done, source: retrieving, dest: generating}, {trigger: need_tool, source: generating, dest: tool_calling}, {trigger: tool_done, source: tool_calling, dest: generating}, {trigger: finish, source: generating, dest: done}, ] )每个状态对应一个处理函数状态转移由明确的触发条件驱动。这样即使模型输出不稳定Agent也不会跑飞。4.5 流式输出的实现流式输出我用SSEServer-Sent Events来实现比WebSocket简单而且天然支持断线重连。from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() app.post(/chat) async def chat(request: ChatRequest): async def generate(): async for chunk in agent.stream(request.message): yield fdata: {json.dumps({content: chunk})}\n\n yield data: [DONE]\n\n return StreamingResponse(generate(), media_typetext/event-stream)前端用EventSource接收每收到一个chunk就追加到界面上。这样用户就能看到文字逐字冒出来的效果。注意流式输出的时候要处理好异常。如果生成过程中出错了要发送一个错误事件给前端而不是直接断开连接。前端收到错误事件后可以显示友好的错误提示。5. 生产环境中的典型问题与排查实录5.1 模型调用超时和重试风暴问题现象某个时间段大量请求超时重试后更加拥堵形成雪崩。排查思路先看监控确认是模型API的问题还是自己服务的问题。如果是API的问题看错误码是限流还是服务不可用。如果是自己服务的问题看是不是线程池满了或者连接池不够。解决方案我一般用熔断器来解决这个问题。当错误率超过阈值时直接熔断不再调用模型快速返回降级结果。等一段时间后再半开试探性放几个请求过去如果成功就恢复。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.failure_count 0 self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.state closed self.last_failure_time None def call(self, func, *args, **kwargs): if self.state open: if time.time() - self.last_failure_time self.recovery_timeout: self.state half-open else: raise CircuitOpenError() try: result func(*args, **kwargs) if self.state half-open: self.state closed self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state open raise5.2 上下文超长导致的截断问题问题现象多轮对话到后面模型开始“失忆”忘记前面说过的内容。排查思路检查token计数看是不是超过了模型的上下文窗口。很多人只计算了用户输入忘了把系统提示、历史对话、检索结果都算进去。解决方案我一般用滑动窗口加摘要的策略。保留最近N轮完整对话更早的对话做摘要。同时监控token使用量接近上限时主动触发压缩。还有一个技巧是动态调整检索结果数量。如果当前上下文已经很长了就少注入一些检索结果如果上下文很短就多注入一些。这样能更充分地利用token预算。5.3 Agent工具调用死循环问题现象Agent反复调用同一个工具或者在不同工具之间来回跳转始终不给出最终答案。排查思路看Agent的调用日志确认是模型的问题还是工具的问题。有时候是工具返回的结果格式不对模型解析不了就反复重试。有时候是模型陷入了思维定势需要外部干预。解决方案我一般设置最大步数限制比如最多10步超过就强制结束并返回当前结果。同时设置重复调用检测如果同一个工具用相同参数调用了3次就跳过这个工具强制模型换一种方式。class ToolCallTracker: def __init__(self, max_repeats3): self.calls {} self.max_repeats max_repeats def should_skip(self, tool_name, params): key f{tool_name}:{hash(str(params))} self.calls[key] self.calls.get(key, 0) 1 return self.calls[key] self.max_repeats5.4 并发场景下的状态混乱问题现象多个用户同时使用时对话状态串了A用户看到B用户的回复。排查思路检查会话ID的生成和传递逻辑看是不是有全局变量或者单例对象被多个请求共享了。解决方案这是典型的并发安全问题。所有跟会话相关的状态必须绑定到会话ID上不能用全局变量。我一般用会话ID作为key把状态存在Redis里每个请求根据自己的会话ID读写状态。class SessionManager: def __init__(self, redis_client): self.redis redis_client async def get_state(self, session_id): data await self.redis.get(fsession:{session_id}) return json.loads(data) if data else {} async def set_state(self, session_id, state): await self.redis.setex( fsession:{session_id}, 3600, json.dumps(state) )提示会话状态的过期时间要设置合理。太短了用户聊到一半状态丢了太长了Redis内存扛不住。我一般设置1小时同时用LRU策略淘汰不活跃的会话。5.5 常见问题速查表问题可能原因排查方法解决方案模型响应慢API限流/网络延迟/模型过载看监控指标区分是API问题还是本地问题限流、降级、切换模型上下文丢失token超限/状态未持久化检查token计数和会话状态摘要压缩、Redis持久化工具调用失败参数格式错误/工具不可用看工具调用日志和返回结果参数校验、工具降级输出格式不对prompt不清晰/模型能力不足检查prompt和模型选择优化prompt、换模型并发状态混乱全局变量/单例共享检查状态存储方式会话隔离、Redis存储6. 架构演进从单机到分布式的思考6.1 什么时候需要分布式架构一开始不要想分布式单机跑得好好的就别折腾。我见过太多团队一上来就搞微服务结果运维成本比开发成本还高。什么时候需要考虑分布式我的判断标准是QPS超过单机处理能力比如单机只能扛50 QPS但业务需要200 QPS需要高可用单点故障不可接受需要多实例部署模型推理需要GPU集群本地模型推理需要多卡多机数据量超过单机存储向量数据、对话历史超过单机容量如果没到这些阈值单机加垂直升级就够了。6.2 分布式架构的关键设计分布式架构下有几个关键问题要解决会话亲和性同一个用户的请求最好路由到同一个实例这样本地缓存能命中。但如果实例挂了会话要能迁移到其他实例。我一般用一致性哈希做路由同时把会话状态存在Redis里这样实例挂了也不影响。模型服务的负载均衡多个模型实例之间怎么分配请求我一般用加权轮询根据每个实例的GPU利用率和响应延迟动态调整权重。消息队列解耦Agent的异步任务通过消息队列来分发比如文档索引、批量推理这些耗时操作。这样API层可以快速返回后台慢慢处理。6.3 可观测性建设分布式架构下可观测性是生命线。我一般从三个维度来建设指标Metrics每个环节的耗时、成功率、token消耗量。用Prometheus采集Grafana展示。日志Logs每个请求的完整链路日志包括prompt、模型输出、工具调用、检索结果。用ELK或者Loki存储。链路追踪Tracing一个请求经过哪些服务、每个服务耗时多少。用OpenTelemetry做埋点Jaeger展示。这三个维度缺一不可。只有指标没有日志出了问题不知道原因只有日志没有链路不知道瓶颈在哪。提示AI应用的可观测性比传统应用更重要因为LLM的不确定性导致问题很难复现。完整的链路日志能帮你回溯每一次调用的完整上下文这对排查问题至关重要。7. 一些踩坑之后的真心话做AI应用架构这两年最大的体会是不要为了架构而架构。我见过太多团队把架构画得特别漂亮各种分层、各种抽象结果代码写出来跑都跑不起来。架构的目的是解决问题不是展示技术。你的应用如果就一个简单的问答场景那就一个FastAPI加一个模型调用就够了不需要什么Agent编排、状态机、多级缓存。等业务复杂了再逐步演进。另一个体会是prompt工程和架构设计同样重要。很多人把精力都花在架构上prompt随便写写结果效果很差。实际上在LLM应用里prompt就是业务逻辑prompt写不好架构再好也没用。最后一个建议多动手少看文章。包括这篇内容你看完了觉得有道理但不动手搭一遍永远不知道坑在哪。找个周末从最简单的模型调用开始一步步加上RAG、加上Agent、加上流式输出每加一个功能就压测一下看看瓶颈在哪。这个过程比看十篇文章都有用。我自己到现在还在不断调整架构每次业务变化都会带来新的挑战。没有一劳永逸的架构只有持续演进的架构。