1. 为什么“图解”才是AI应用架构的正确打开方式
做AI应用开发这几年,我最大的感受是:架构图比代码更值钱。你去看任何一个能跑起来的AI应用,代码可能几千行,但真正决定它能不能稳定跑、能不能扩展、能不能换模型的,是那张架构图。我见过太多团队,模型选得挺好,Prompt调得也不错,但一上线就崩——问题不在模型,在架构。
“图解AI应用架构设计”这个主题,核心不是教你画图,而是教你用架构思维去拆解一个AI应用。它解决的是这样一个问题:当你拿到一个需求,比如“做一个能自动查资料、写报告、发邮件的Agent”,你脑子里应该浮现出什么样的分层结构?哪些模块是必须的?哪些可以省?数据怎么流?模型怎么换?工具怎么接?
这篇文章适合三类人:一是刚接触AI应用开发、想搞清楚整体框架的开发者;二是已经在做LLM应用、但总觉得系统不够稳的工程师;三是需要做技术选型和架构评审的技术负责人。我会从分层设计讲到核心模块,从Agent编排讲到MCP协议,从实操步骤讲到踩坑经验,尽量把每个“为什么”都说清楚。
先给一个我常用的AI应用五层架构总览,后面所有内容都围绕这张图展开:
| 层级 | 名称 | 核心职责 | 典型组件 |
|---|---|---|---|
| L5 | 交互层 | 用户输入输出、多端适配 | Web、App、API Gateway |
| L4 | 编排层 | 任务规划、Agent调度、流程控制 | Agent框架、Workflow引擎 |
| L3 | 能力层 | 工具调用、检索、记忆、多模态 | MCP、RAG、Memory、Tool |
| L2 | 模型层 | 推理、生成、嵌入 | LLM、Embedding、Reranker |
| L1 | 基础设施层 | 算力、存储、监控、安全 | GPU、向量库、日志、鉴权 |
这张图看着简单,但每一层都有讲究。下面我逐层拆。
2. 交互层与编排层:Agent到底该怎么编排
2.1 交互层不是“画个界面”那么简单
很多人觉得交互层就是前端的事,跟AI架构没关系。这是个误区。交互层决定了用户意图怎么被结构化地传给编排层,也决定了模型输出怎么被安全地展示给用户。
我做过一个企业知识助手,前端看起来就是个聊天框,但背后做了三件事:第一,把用户输入做意图预分类,判断是问答、检索还是操作指令;第二,对输入做敏感信息过滤,避免把内部数据直接塞给外部模型;第三,对输出做流式渲染和引用标注,让用户知道答案来自哪个文档。
提示:交互层一定要做输入预处理和输出后处理,不要直接把用户输入丢给LLM,也不要直接把LLM输出渲染到页面。
具体来说,输入预处理包括:长度截断、敏感词过滤、意图标签提取、会话上下文组装。输出后处理包括:流式分块、Markdown渲染、引用来源标注、幻觉检测标记。这些逻辑放在API Gateway或BFF层做,不要放到前端,也不要用模型做。
2.2 编排层的三种模式:Chain、Agent、Workflow
编排层是AI应用的大脑。我把它分成三种模式,对应不同复杂度:
第一种,Chain模式。就是固定流程,A步骤输出给B步骤,B给C。比如“翻译→摘要→格式化”。这种最简单,用LangChain的LCEL或者直接写代码串起来就行。优点是可控、可测、便宜;缺点是不灵活,用户换个问法就废了。
第二种,Agent模式。给LLM一堆工具,让它自己决定调哪个、调几次。比如用户说“帮我查一下北京明天天气,如果下雨就提醒我带伞”,Agent会先调天气工具,再根据结果决定是否调提醒工具。优点是灵活;缺点是贵、慢、不可控,容易陷入循环。
第三种,Workflow模式。介于两者之间,用状态机或DAG定义流程,但允许在某些节点做动态路由。比如客服场景:先分类问题,如果是退款走退款流程,如果是咨询走咨询流程,每个流程内部是固定的,但入口是动态的。
我的经验是:能用Workflow就别用Agent,能用Chain就别用Workflow。Agent适合探索性任务,不适合生产环境的高频请求。我见过一个团队用Agent做订单查询,结果每次查询要调三四次模型,成本是固定流程的十倍,延迟还高。
2.3 Agent编排的核心:状态、工具、记忆
如果你确实需要Agent,那编排层要重点设计三个东西:
状态管理。Agent执行过程中会产生很多中间状态:当前任务、已调用的工具、工具返回结果、剩余步数。这些状态要持久化,不然Agent一崩就全丢了。我一般用Redis存会话状态,用数据库存任务轨迹。
工具注册与发现。Agent怎么知道有哪些工具可用?传统做法是硬编码一个工具列表,但这样扩展性差。现在更流行的做法是用MCP(Model Context Protocol)来标准化工具接口。MCP本质上是一个协议,让模型和工具之间用统一的方式通信。你可以把它理解成“AI世界的USB接口”——不管什么工具,只要实现了MCP,就能被Agent调用。
记忆机制。Agent需要短期记忆(当前会话)和长期记忆(历史交互)。短期记忆直接放上下文,长期记忆要向量化存储。这里有个坑:不要把所有历史都塞进上下文,会爆token。我的做法是:最近三轮对话全量保留,更早的做摘要,关键事实做向量检索。
3. 能力层核心:MCP、RAG与工具调用
3.1 MCP到底是什么,为什么它重要
MCP这个词最近很热,但很多人说不清楚。我用一句话解释:MCP是一套让LLM应用以统一方式连接外部工具和数据源的协议。在没有MCP之前,你接一个数据库、一个API、一个文件系统,每个都要写一套适配代码。有了MCP,你只需要实现一个MCP Server,任何支持MCP的客户端都能调用。
MCP的架构分三部分:Host(宿主应用,比如你的AI应用)、Client(连接器,负责和Server通信)、Server(工具提供方,暴露资源和工具)。通信方式支持stdio和HTTP+SSE两种。
我实测下来,MCP最大的价值是解耦。以前工具调用逻辑和Agent逻辑混在一起,换一个工具就要改Agent代码。现在工具独立成MCP Server,Agent只认协议不认实现。比如你把“查询数据库”做成一个MCP Server,以后不管换MySQL还是PostgreSQL,Agent那边完全不用动。
注意:MCP Server要做好权限控制。我见过有人把数据库MCP Server暴露到公网,结果被扫到直接拖库。MCP Server至少要做认证、限流、审计日志。
3.2 RAG不是“向量检索”四个字能概括的
RAG(检索增强生成)是AI应用里最常用的能力,但很多人把它做成了“向量库+LLM”的简单拼接,效果很差。我总结RAG有五个关键环节:
第一,文档解析。PDF、Word、HTML、Excel,每种格式的解析策略不同。PDF要处理表格和图片,HTML要去导航栏和广告。我一般用Unstructured或LlamaParse做解析,复杂表格用多模态模型辅助。
第二,分块策略。固定长度分块是最懒的做法,效果也最差。我的做法是:按语义分块,段落优先,段落太长再按句子切,同时保留标题层级作为元数据。块大小控制在300-800 token,重叠50-100 token。
第三,嵌入模型选择。中文场景我常用BGE或M3E,英文用text-embedding-3-large。不要用LLM做嵌入,又贵又慢。嵌入维度不是越高越好,1024维通常够用,太高反而检索噪声大。
第四,检索策略。纯向量检索召回率有限,我一般用混合检索:向量检索+关键词检索(BM25),然后用Reranker重排。Reranker用BGE-Reranker或Cohere Rerank,效果提升明显。
第五,生成约束。检索到的内容怎么给LLM?不要直接拼接,要加引用标记,让模型知道每段话来自哪里。Prompt里明确要求“只根据提供的资料回答,不知道就说不知道”。
3.3 工具调用的三种实现方式对比
工具调用是Agent的手脚。我对比过三种实现方式:
| 方式 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| Function Calling | 模型原生支持,输出结构化调用 | 准确率高、延迟低 | 依赖模型能力、工具数量受限 | 主流LLM、工具少于20个 |
| ReAct | 模型输出思考+动作,解析后执行 | 灵活、可解释 | 解析易出错、多轮延迟高 | 复杂推理、工具多 |
| MCP | 标准化协议,独立Server | 解耦、可复用、生态好 | 需要额外部署、协议较新 | 企业级、多应用共享工具 |
我的建议是:优先用Function Calling,工具多了用MCP,ReAct只在需要复杂推理时用。Function Calling现在主流模型都支持,DeepSeek、Qwen、GPT系列都没问题。MCP适合把工具做成公共服务,多个AI应用共享。
4. 模型层与基础设施层:选型、部署与监控
4.1 LLM选型:不要只看榜单
选模型是AI应用架构里最容易拍脑袋的环节。很多人看几个榜单就定了,结果上线发现成本爆炸或者延迟太高。我选模型看五个维度:
能力匹配。你的任务需要什么能力?分类任务用小模型就够,别上大模型。代码生成要选代码强的,中文理解要选中文语料多的。我一般先拿100条真实数据做评测,不看公开榜单。
成本。按token算,GPT-4级别大概是DeepSeek的几十倍。如果你的应用QPS高,成本差异会非常夸张。我算过一笔账:每天10万次请求,每次平均500 token,用GPT-4一个月要几万块,用国产模型可能只要几百块。
延迟。流式输出首token延迟很关键。大模型首token可能要1-2秒,小模型可以做到300毫秒以内。如果做实时对话,延迟比能力更重要。
可控性。能不能私有化部署?能不能微调?数据会不会被用于训练?企业场景这些是硬要求。
生态。工具调用支持好不好?有没有官方SDK?社区活跃不活跃?
我的常用组合是:主力用DeepSeek或Qwen做通用推理,轻量任务用7B小模型,嵌入用BGE,重排用BGE-Reranker。这套组合性价比很高,而且都能私有化部署。
4.2 模型部署:本地、云、混合怎么选
部署方式直接影响架构设计。三种方式:
纯云API。最简单,按量付费,适合快速验证和低QPS场景。缺点是数据出域、延迟不可控、长期成本高。
纯本地部署。用vLLM或TGI部署开源模型,数据不出域,延迟可控,长期成本低。缺点是要买GPU、要运维、模型能力受限于开源。
混合部署。敏感数据走本地,通用任务走云API。这是我最推荐的方案。架构上要做模型路由:根据任务类型、数据敏感度、当前负载,动态选择模型。
模型路由的实现不复杂,一个简单的策略引擎就行。比如:包含“内部”“机密”关键词的请求走本地,其他走云;本地GPU利用率超过80%时溢出到云。
4.3 基础设施层:监控、日志、安全一个都不能少
基础设施层是AI应用的底座,但最容易被忽视。我重点说三个:
监控。AI应用的监控和传统应用不同,要额外关注:token消耗、模型延迟、工具调用成功率、幻觉率、检索命中率。我用Prometheus+Grafana做基础监控,用LangFuse做LLM专项监控。LangFuse能记录每次调用的输入输出、token数、延迟,排查问题非常方便。
日志。AI应用的日志要记录完整链路:用户输入→意图分类→检索结果→Prompt→模型输出→后处理→最终响应。这样出问题才能定位。注意日志要脱敏,不要把用户隐私写进去。
安全。AI应用的安全面比传统应用大:Prompt注入、工具滥用、数据泄露、模型越狱。我的做法是:输入层做Prompt注入检测,工具层做权限校验和参数校验,输出层做敏感信息过滤。另外,所有工具调用都要有审计日志,谁在什么时候调了什么工具、传了什么参数、返回了什么,全部记录。
5. 实操:从零搭一个可用的AI应用架构
5.1 环境准备与依赖安装
我以一个“企业知识助手”为例,走一遍完整流程。这个助手能回答内部文档问题、查询数据库、发送通知。
先准备环境。Python 3.10以上,推荐用uv做包管理,比pip快很多。
# 安装uv curl -LsSf https://astral.sh/uv/install.sh | sh # 创建项目 uv init ai-assistant cd ai-assistant # 安装核心依赖 uv add fastapi uvicorn langchain langchain-community uv add chromadb sentence-transformers uv add openai httpx pydantic uv add langfuse prometheus-client模型方面,我本地用vLLM部署Qwen2.5-7B做轻量任务,云上用DeepSeek做复杂推理。嵌入用BGE-M3,重排用BGE-Reranker-v2。
# 本地部署嵌入模型 uv add FlagEmbedding5.2 核心模块实现:检索、编排、工具
先实现检索模块。文档解析用Unstructured,分块按语义,嵌入用BGE-M3。
from FlagEmbedding import BGEM3FlagModel from unstructured.partition.auto import partition import chromadb class Retriever: def __init__(self): self.model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True) self.client = chromadb.PersistentClient(path="./chroma_db") self.collection = self.client.get_or_create_collection("docs") def parse_and_chunk(self, file_path): elements = partition(filename=file_path) chunks = [] current = "" for el in elements: if len(current) + len(str(el)) > 800: chunks.append(current) current = str(el) else: current += "\n" + str(el) if current: chunks.append(current) return chunks def index(self, chunks, metadata): embeddings = self.model.encode(chunks)['dense_vecs'] self.collection.add( embeddings=embeddings.tolist(), documents=chunks, metadatas=metadata, ids=[f"doc_{i}" for i in range(len(chunks))] ) def search(self, query, top_k=10): q_emb = self.model.encode([query])['dense_vecs'] results = self.collection.query( query_embeddings=q_emb.tolist(), n_results=top_k ) return results['documents'][0]然后是编排模块。我用一个简单的状态机做Workflow,入口做意图分类,根据意图路由到不同流程。
from enum import Enum from pydantic import BaseModel class Intent(Enum): QA = "qa" DB_QUERY = "db_query" NOTIFY = "notify" class State(BaseModel): intent: Intent = None query: str = "" context: list = [] answer: str = "" step: int = 0 class Orchestrator: def __init__(self, llm, retriever, tools): self.llm = llm self.retriever = retriever self.tools = tools def classify(self, query): prompt = f"判断以下问题的意图,只输出qa/db_query/notify之一:{query}" result = self.llm.invoke(prompt).strip() return Intent(result) def run(self, query): state = State(query=query) state.intent = self.classify(query) if state.intent == Intent.QA: docs = self.retriever.search(query) state.context = docs state.answer = self.llm.invoke( f"根据资料回答:{docs}\n问题:{query}" ) elif state.intent == Intent.DB_QUERY: state.answer = self.tools['db'].run(query) elif state.intent == Intent.NOTIFY: state.answer = self.tools['notify'].run(query) return state工具层用MCP标准化。我写一个简单的MCP Server暴露数据库查询和通知发送。
from mcp.server import Server from mcp.types import Tool, TextContent app = Server("enterprise-tools") @app.list_tools() async def list_tools(): return [ Tool(name="query_db", description="查询数据库", inputSchema={"type":"object","properties":{"sql":{"type":"string"}}}), Tool(name="send_notify", description="发送通知", inputSchema={"type":"object","properties":{"msg":{"type":"string"}}}) ] @app.call_tool() async def call_tool(name, arguments): if name == "query_db": result = execute_sql(arguments["sql"]) return [TextContent(type="text", text=str(result))] elif name == "send_notify": send(arguments["msg"]) return [TextContent(type="text", text="sent")]5.3 参数计算与性能调优
分块大小怎么定?我一般按嵌入模型的最大长度反推。BGE-M3最大8192 token,但实际检索时块太大噪声多。我的经验值是300-800 token,对应中文大概200-500字。重叠50-100 token保证边界信息不丢。
检索top_k怎么定?太小召回不够,太大噪声多。我一般先取20,Reranker后取5。实测下来,top_k=20+Rerank到5,比直接top_k=5效果好很多。
并发怎么控制?LLM调用是IO密集,用asyncio做并发。但要注意模型服务的并发上限,vLLM默认并发可能不够,要调--max-num-seqs参数。云API要注意限流,加退避重试。
成本怎么算?我一般按“每千次请求成本”来算。公式是:(输入token×输入单价 + 输出token×输出单价) × 1000。比如DeepSeek输入1元/百万token,输出2元/百万token,每次请求输入2000输出500,那每千次成本是(2000×1 + 500×2)/1000000×1000 = 3元。这个数要心里有数。
6. 常见问题与排查技巧实录
6.1 Agent陷入循环怎么办
这是Agent最典型的问题。模型反复调同一个工具,或者在不同工具之间来回跳。我的排查思路:
先看日志,确认是模型输出问题还是工具返回问题。如果工具返回了错误信息但模型没识别,就在工具返回里加明确的错误标记。如果模型就是绕不出来,加最大步数限制和重复调用检测。我一般设最大10步,同一个工具连续调3次就强制中断。
更根本的解法是优化Prompt。在系统提示里明确写“如果工具返回错误,不要重试超过2次”“如果已经获得足够信息,立即给出答案”。另外,工具描述要写清楚什么时候用、什么时候不用。
6.2 检索结果不相关怎么调
RAG效果差,八成是检索环节的问题。我按这个顺序排查:
第一,看分块。把检索到的块打印出来,看是不是语义完整。如果块被切得七零八落,调分块策略。
第二,看嵌入。拿几个典型query,看检索到的文档和query的向量相似度。如果相似度普遍低于0.5,可能是嵌入模型不适合你的领域,换模型或微调。
第三,看检索方式。纯向量检索对关键词不敏感,加BM25做混合检索。我常用rank_bm25做关键词检索,和向量结果做RRF融合。
第四,看Reranker。如果混合检索后还是不准,加Reranker。Reranker对top20重排,通常能把正确结果提到前3。
6.3 模型输出不稳定怎么治
同一个问题,模型每次回答不一样,这在生产环境是灾难。我的做法:
降低temperature。创意任务用0.7-1.0,事实性任务用0-0.3。我一般默认0.1,需要创意时再调。
固定seed。有些模型支持seed参数,设了之后同样输入输出基本一致。
结构化输出。让模型输出JSON,用Pydantic做校验。校验不过就重试,重试还不过就降级到规则引擎。
缓存。相同query直接返回缓存结果。我用Redis做语义缓存,query嵌入相似度超过0.95就命中缓存。
6.4 常见问题速查表
| 问题 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| Agent循环 | Prompt不清、工具返回模糊 | 看调用日志 | 加步数限制、优化Prompt |
| 检索不准 | 分块差、嵌入不匹配 | 打印检索结果 | 调分块、换嵌入、加Reranker |
| 输出不稳定 | temperature高、无约束 | 多次调用对比 | 降temperature、结构化输出 |
| 延迟高 | 模型大、串行调用 | 打点计时 | 换小模型、并发调用、流式输出 |
| 成本高 | 上下文太长、重复调用 | 统计token | 压缩上下文、加缓存 |
| 工具调用失败 | 参数错、权限不足 | 看工具日志 | 加参数校验、权限检查 |
| 幻觉严重 | 检索没召回、Prompt没约束 | 对比检索结果 | 加引用、明确“不知道就说不知道” |
| MCP连接失败 | 协议版本不匹配 | 看MCP日志 | 统一协议版本、检查网络 |
6.5 几个我踩过的坑
第一个坑:上下文塞太多。刚开始做RAG,我把检索到的10个块全塞进Prompt,结果模型被噪声带偏,还贵。后来改成Rerank后只取3个,效果反而更好。
第二个坑:工具描述太简略。MCP工具只写了个名字,模型不知道怎么用。后来把描述写详细,包括什么时候用、参数什么含义、返回什么格式,调用准确率大幅提升。
第三个坑:没做超时控制。工具调用没设超时,一个慢查询把整个请求卡死。后来所有工具调用都加超时,超时就走降级逻辑。
第四个坑:日志没脱敏。排查问题时发现日志里有用户手机号,赶紧加脱敏。AI应用的日志一定要过一遍脱敏规则。
第五个坑:模型路由没做降级。云API挂了整个应用就挂了。后来加了本地模型兜底,云API超时就切本地,虽然能力差一点但至少能用。
7. 架构演进:从单体到分层到Agent Mesh
最后聊聊架构演进。我经历过三个阶段:
第一阶段,单体架构。所有逻辑写在一个服务里,检索、编排、模型调用混在一起。优点是简单,缺点是改一处动全身,没法独立扩展。
第二阶段,分层架构。就是我前面讲的五层架构,每层独立部署、独立扩展。检索层可以单独扩容,模型层可以换模型不影响上层。这是目前大多数AI应用应该采用的架构。
第三阶段,Agent Mesh。多个Agent各自负责一个领域,通过MCP互相调用。比如一个客服Agent、一个数据分析Agent、一个运维Agent,它们通过标准协议协作。这是未来的方向,但目前落地案例还不多,主要是协调和一致性问题没完全解决。
我的建议是:大多数团队先做好分层架构,把检索、编排、工具、模型四层解耦清楚,再考虑Agent Mesh。不要一上来就搞多Agent,复杂度会压垮你。
架构设计没有银弹,只有权衡。你选Agent就要接受它的不可控,选Workflow就要接受它的不灵活。关键是清楚你的场景需要什么,然后做出对应的取舍。我个人的经验是:生产环境优先选可控的方案,探索性任务再上Agent。这个原则帮我避开了很多坑。