ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

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

图解AI应用架构设计:从模型接入到Agent编排的六层架构实战 1. 从一张架构图说起AI应用到底该怎么搭很多人第一次接触AI应用开发脑子里冒出来的第一个念头是调个API不就完了。我刚开始也这么想直到真正把一个能跑通业务闭环的AI应用从零搭起来才发现事情远没有想象中那么简单。模型调用只是冰山露出水面的那一角水面之下还有上下文管理、工具编排、记忆存储、容错重试、流式输出、权限控制等一大堆工程问题在等着你。图解AI应用架构设计这个主题说白了就是要把这些藏在冰山之下的东西用一张清晰的架构图串起来让你知道一个生产级别的AI应用到底由哪些层组成、每层负责什么、层与层之间怎么交互。它解决的核心问题是当你从写个demo跨向做一个能上线的产品时脑子里得有一张全局地图否则你会在无数个技术选型和架构决策面前反复纠结。这篇文章适合三类人看一是刚入门AI应用开发、想知道整体框架长什么样的新手二是已经能调通API、但应用一复杂就手忙脚乱的进阶开发者三是需要做技术方案评审、想快速判断一个AI应用架构是否合理的团队负责人。我会围绕AI应用、架构设计、Agent、LLM、MCP这几个核心关键词把架构拆成一层一层来讲每一层都配上我实际踩过的坑和可复现的做法。需要提前说明的是架构设计没有银弹我下面讲的是一套经过多个项目验证的通用分层思路你可以根据自己的业务规模做裁剪。小项目可能只需要其中三四层大项目可能每层还要再细分。关键是理解每层为什么存在而不是照搬。2. AI应用架构的整体分层与设计思路2.1 为什么AI应用需要分层架构传统Web应用的分层接入层、业务层、数据层大家都熟但AI应用多了一个非常特殊的东西不确定性。同一个输入LLM可能给你不同的输出同一个工具调用可能这次成功下次超时同一个上下文token数量可能因为模型版本变化而波动。这种不确定性决定了AI应用不能简单套用传统三层架构必须把和模型打交道这件事单独抽出来做成可观测、可替换、可降级的一层。我见过太多项目把模型调用直接写在业务逻辑里结果想换个模型要改几十个文件想加个重试机制发现到处都要改。分层的第一价值就是隔离变化模型会换、prompt会调、工具会增删把这些变化都关在一层里业务层就不用跟着遭殃。另一个原因是可测试性。AI应用最难测的就是模型输出如果你把模型调用和业务逻辑混在一起单元测试基本没法写。分层之后你可以给模型层做mock业务层就能用传统方式测试模型层则用专门的评估集来测。2.2 六层架构总览我把一个完整的AI应用拆成六层从下往上依次是层级名称核心职责典型组件L1模型接入层统一封装各家LLM接口模型网关、适配器L2上下文与记忆层管理对话历史与长期记忆会话存储、向量库L3工具与能力层提供模型可调用的外部能力MCP、函数调用L4编排与Agent层决定下一步做什么Agent循环、工作流引擎L5应用服务层承载具体业务逻辑API、业务规则L6交互与接入层面向用户的界面与协议Web、流式输出这个分层不是拍脑袋定的它对应的是一次AI请求从用户输入到返回结果的完整链路。用户在第6层输入请求往下走到第5层做业务处理第4层决定要不要调工具、调几次第3层执行具体工具第2层负责把历史上下文拼进去第1层真正去调模型然后结果原路返回。注意分层是逻辑分层不是物理分层。小项目里L1和L2可能就在同一个进程里大项目里每一层都可能是独立微服务。别为了分层而分层先想清楚每层的边界在哪。2.3 分层设计的三个关键取舍第一个取舍是同步还是异步。模型调用动辄几秒到几十秒如果全同步用户体验会很差。我的做法是交互层用流式输出SSE或WebSocket让用户能实时看到token一个个蹦出来编排层内部如果涉及多个工具调用能并行就并行用异步任务编排。第二个取舍是状态放哪。Agent是有状态的它需要记住我上一步做了什么。状态可以放内存简单但重启就丢、放Redis快但容量有限、放数据库持久但慢。我的经验是短期会话状态放Redis长期记忆放向量库关系库两者通过会话ID关联。第三个取舍是工具调用的边界。工具是Agent的手脚但手脚太多会失控。我一般会限制单个Agent可调用的工具数量在10个以内超过就拆成多个Agent用主Agent调度子Agent。这个数字不是绝对的但工具一多模型选错工具的概率会明显上升。3. 核心层拆解模型接入与上下文管理3.1 模型接入层别让业务代码直接碰SDK模型接入层的核心目标是让上层感觉不到模型的存在。上层只说我要一段文本补全至于底层是哪个厂商的模型、用什么协议、怎么鉴权上层不关心。实现方式就是适配器模式。定义一个统一的接口比如class LLMProvider: def chat(self, messages, toolsNone, streamFalse): raise NotImplementedError def embed(self, texts): raise NotImplementedError然后每个厂商写一个适配器实现这个接口。这样换模型只需要加一个适配器业务代码一行不用改。这里有个细节很多人忽略不同厂商的message格式其实不一样。有的用role: user/assistant/system有的把system单独拎出来有的对tool调用结果的格式要求不同。适配器的一个重要职责就是把这些差异抹平对外暴露统一的格式。参数归一化也是接入层要做的。temperature、top_p、max_tokens这些参数各家叫法可能不同取值范围也不同。接入层要统一成一套标准参数再映射到各家的实际参数上。我踩过的坑是某家模型的temperature范围是0到1另一家是0到2直接透传会导致行为完全不一样。3.2 上下文管理token就是钱上下文管理的本质是在有限的token预算里塞进最有用的信息。模型的上下文窗口再大也是有限的而且token是要花钱的塞太多无关信息既浪费钱又降低效果。我的上下文组装策略是分优先级P0 系统提示词定义角色和基本规则永远保留P1 当前任务相关用户当前问题、相关工具定义P2 近期对话最近N轮对话保证连贯性P3 长期记忆从向量库检索出的相关历史P4 背景知识RAG检索出的文档片段当总token超预算时从P4往P0依次裁剪。裁剪不是简单截断而是要有策略对话历史可以摘要压缩文档片段可以只保留最相关的几段。实操心得我一般会预留20%的token预算给模型输出剩下80%给输入。如果输入经常超说明要么检索策略有问题要么该做对话摘要了。3.3 记忆层短期靠窗口长期靠检索记忆分两种。短期记忆就是对话历史放在上下文窗口里简单直接。长期记忆则需要跨会话保留比如这个用户上次说他喜欢简洁的回答这种信息不能一直占着窗口得存起来按需检索。长期记忆的实现通常是把重要信息embedding后存进向量库下次对话时用当前问题去检索相关记忆拼进上下文。这里的关键是什么该记、什么不该记。我的做法是让模型自己判断在prompt里加一条指令如果用户透露了值得长期记住的偏好或事实调用save_memory工具。这样记忆的写入就变成了一个工具调用很自然。检索的时候要注意向量相似度不等于相关性。我遇到过检索出一堆语义相似但实际没用的记忆反而干扰了模型。解决办法是加一层重排序rerank或者干脆限制检索数量宁缺毋滥。4. 工具层与MCP让模型长出双手4.1 工具调用的本质与常见误区工具调用Function Calling的本质是模型不直接执行任何操作它只是输出一个结构化的我想调用某个工具参数是这些的意图真正执行的是你的代码。理解这一点很重要因为它意味着安全边界在你手里不在模型手里。常见误区有三个。第一个是以为模型会自动调用工具其实你得在请求里把工具定义传进去模型才会考虑用。第二个是工具描述写得太随意模型根本看不懂什么时候该用。工具描述要写得像给新员工看的操作手册说清楚这个工具做什么、什么时候用、参数什么意思。第三个是不处理工具调用失败模型拿到一个错误结果可能就懵了你得把错误信息结构化地返回给它让它知道怎么调整。4.2 MCP是什么为什么它重要MCPModel Context Protocol是一套让模型和外部工具、数据源标准化对接的协议。你可以把它理解成AI世界的USB接口——以前每个工具都要为每个模型单独适配现在大家都遵循MCP工具写一次就能被所有支持MCP的客户端使用。MCP的核心价值在于解耦。工具提供方不需要知道谁在用客户端不需要知道工具怎么实现双方只通过协议交互。这对生态的意义很大就像USB统一了外设接口一样MCP在统一AI工具的接口。一个典型的MCP架构包含三部分MCP Host宿主应用比如你的AI应用、MCP Client协议客户端负责和Server通信、MCP Server工具提供方。Host里可以跑多个Client每个Client连一个Server这样就能同时接入多个工具源。4.3 自己写一个MCP Server的要点写MCP Server其实不难核心是实现几个标准方法列出可用工具list tools、执行工具call tool。工具的定义包括名称、描述、参数schema这些和Function Calling的定义基本一致。我建议从最简单的stdio传输开始本地进程间通信用stdio最省事。等需要远程调用时再换成SSE或HTTP传输。工具的实现要注意幂等性因为模型可能会重复调用同一个工具你的工具得能安全地处理重复请求。注意MCP Server暴露的工具等于给了模型操作你系统的能力权限控制必须做。我的做法是每个工具都标注风险等级高风险工具比如删除、支付需要额外确认不能让模型直接调。4.4 工具编排并行还是串行当一次任务需要调多个工具时编排策略很关键。如果工具之间没有依赖就并行调用能省不少时间。如果有依赖比如先查用户ID再查订单就得串行。Agent循环里模型可能一次返回多个工具调用请求这时候要判断这些调用能不能并行。我的经验是读操作基本都能并行写操作要谨慎涉及同一资源的写操作必须串行否则会有竞态问题。5. Agent编排层从工作流到自主决策5.1 Agent和工作流的区别很多人把Agent和工作流混为一谈其实它们解决的是不同问题。工作流是你预先定义好步骤模型在每一步里做具体的事Agent是你只给目标模型自己决定走哪些步骤。举个例子报销审批。如果是工作流你会定义第一步提取发票信息第二步校验金额第三步走审批模型只负责提取和校验。如果是Agent你只说帮我把这张发票处理了Agent自己决定要不要查政策、要不要问用户补充信息、要不要发起审批。选择哪个取决于任务的确定性。流程固定、要求可预测的用工作流稳定可控。流程多变、需要灵活应变的用Agent但要做好它乱来的准备。5.2 Agent循环的核心结构一个Agent循环基本就是思考-行动-观察的重复把当前状态目标、历史、可用工具发给模型模型输出要么是最终答案要么是工具调用请求如果是工具调用执行工具把结果加进历史回到第1步直到模型给出最终答案或达到最大轮数这个循环看起来简单但工程上有几个必须处理的点。最大轮数一定要设否则模型可能陷入死循环烧钱又烧时间。循环检测也要做如果模型连续几轮调用同样的工具、拿到同样的结果就该中断了。中间状态持久化也很重要长任务可能跑几分钟中途进程挂了得能恢复。5.3 多Agent协作的两种模式当任务复杂到单个Agent搞不定时就需要多Agent。常见两种模式主从模式一个主Agent负责规划和调度多个子Agent负责执行具体子任务。主Agent把大任务拆成小任务分给子Agent收集结果后汇总。这种模式适合任务可以清晰拆分的场景。对等模式多个Agent平等协作各自有专长通过消息传递协调。比如一个写代码的Agent和一个审查代码的Agent互相配合。这种模式适合需要多视角碰撞的场景但协调成本高容易扯皮。我的建议是先用单Agent实在不够再上多Agent。多Agent的调试难度是单Agent的好几倍很多问题在单Agent里加个工具就能解决没必要上多Agent。5.4 Agent的容错设计Agent自主性越强出错的可能性越大。容错设计要覆盖几个层面工具层工具调用失败要有重试和降级超时要有兜底模型层模型输出格式错误要能解析容错实在解析不了就重新请求循环层死循环检测、最大轮数限制、超时中断业务层关键操作要有人工确认不能让Agent自己拍板我踩过最坑的一次是Agent在调用支付工具时因为网络抖动失败了它自己重试了三次结果扣了三次钱。后来所有涉及资金的操作都加了幂等键同一个请求ID只处理一次。6. 实操从零搭一个最小可用的AI应用骨架6.1 环境准备与依赖选型我以一个Python项目为例搭一个包含模型接入、工具调用、Agent循环的最小骨架。依赖选型上我倾向于用轻量的方案不引入过重的框架这样你能看清每一层在做什么。核心依赖就几个一个HTTP客户端httpx、一个向量库客户端如果用RAG、一个Web框架FastAPI。模型SDK我建议直接用HTTP调不依赖厂商的SDK这样换厂商时改动最小。pip install fastapi uvicorn httpx pydantic目录结构按分层来组织app/ providers/ # L1 模型接入 memory/ # L2 上下文与记忆 tools/ # L3 工具 agent/ # L4 编排 services/ # L5 业务 api/ # L6 接口6.2 模型接入层的实现先定义统一接口再写一个具体实现。这里以OpenAI兼容接口为例因为很多厂商都兼容这个格式class OpenAICompatProvider: def __init__(self, base_url, api_key, model): self.base_url base_url self.api_key api_key self.model model async def chat(self, messages, toolsNone, streamFalse): payload { model: self.model, messages: messages, stream: stream, } if tools: payload[tools] tools async with httpx.AsyncClient() as client: resp await client.post( f{self.base_url}/chat/completions, jsonpayload, headers{Authorization: fBearer {self.api_key}}, timeout60, ) return resp.json()这个实现很朴素但已经够用了。生产环境要加的东西重试、超时分级、token计数、错误分类、日志埋点。重试要注意只对可重试的错误重试比如429、5xx对400这种参数错误重试没意义。6.3 工具注册与执行工具用一个注册表管理每个工具是一个函数加一份schemaTOOLS {} def register_tool(name, description, parameters): def decorator(func): TOOLS[name] { schema: { type: function, function: { name: name, description: description, parameters: parameters, } }, func: func, } return func return decorator register_tool( nameget_weather, description查询指定城市的当前天气, parameters{ type: object, properties: { city: {type: string, description: 城市名} }, required: [city], } ) async def get_weather(city): # 实际实现省略 return {city: city, temp: 25, condition: 晴}执行工具时要做参数校验模型给的参数不一定符合schema得用pydantic之类的工具校验一遍再执行。执行结果统一转成字符串返回给模型复杂结构用JSON序列化。6.4 Agent循环的实现Agent循环是整个骨架的心脏async def run_agent(provider, messages, max_turns10): for turn in range(max_turns): tools_schema [t[schema] for t in TOOLS.values()] resp await provider.chat(messages, toolstools_schema) msg resp[choices][0][message] messages.append(msg) tool_calls msg.get(tool_calls) if not tool_calls: return msg[content] for call in tool_calls: name call[function][name] args json.loads(call[function][arguments]) result await TOOLS[name][func](**args) messages.append({ role: tool, tool_call_id: call[id], content: json.dumps(result, ensure_asciiFalse), }) return 达到最大轮数任务未完成这段代码虽然短但包含了Agent的核心逻辑。实际使用时要加的东西每轮记录日志、token累计、异常捕获、流式输出。流式输出和工具调用同时存在时比较麻烦因为工具调用需要完整的JSON不能边流边解析通常的做法是流式输出文本部分工具调用部分等完整了再处理。6.5 流式输出到前端流式输出用SSE最简单。FastAPI里可以这样from fastapi.responses import StreamingResponse app.post(/chat) async def chat(req: ChatRequest): async def event_stream(): async for chunk in agent_stream(req.message): yield fdata: {json.dumps(chunk)}\n\n yield data: [DONE]\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)前端用EventSource接收每收到一个chunk就追加到界面上。要注意的是工具调用期间可能有一段时间没有文本输出前端要显示正在思考之类的状态否则用户以为卡死了。7. 常见问题与排查技巧实录7.1 模型输出格式不稳定怎么办这是最高频的问题。模型有时候返回纯文本有时候返回带markdown的JSON有时候JSON还缺个括号。我的处理策略是分三层第一层prompt里明确要求格式并给示例。第二层解析时做容错比如用正则提取JSON部分或者用宽松的JSON解析器。第三层解析失败就把错误信息返回给模型让它重新输出。如果某个格式要求特别严格可以考虑用结构化输出structured output功能很多厂商都支持强制JSON schema。但要注意强制schema有时会降低模型的其他能力得权衡。7.2 工具调用不触发或调错工具工具不触发通常是描述问题。模型判断要不要调工具全靠工具描述。描述要写清楚什么时候用而不只是是什么。比如查询天气不如当用户询问某地天气情况时用此工具查询实时天气。调错工具通常是工具之间边界模糊。两个工具功能重叠模型就会纠结。解决办法是让工具职责单一一个工具只做一件事。如果确实需要相似功能在描述里明确区分场景。7.3 上下文超长与token爆炸Agent循环里每一轮都会往messages里追加内容几轮下来token就爆了。解决办法是历史压缩当messages超过一定长度时把早期的对话总结成一段摘要用摘要替换原始消息。另一个技巧是工具结果截断。工具返回的结果可能很长比如查了一堆数据但模型往往只需要关键信息。可以在工具层就做截断只返回最相关的部分。7.4 常见问题速查表现象可能原因排查方向模型不调工具工具描述不清检查description是否说明使用场景调错工具工具职责重叠拆分工具明确边界输出格式错prompt约束不足加格式示例用结构化输出响应慢串行工具调用检查能否并行加超时token超限历史未压缩加摘要机制截断工具结果死循环无轮数限制设max_turns加循环检测重复扣费无幂等关键操作加幂等键7.5 几个我踩过的坑第一个坑是把system prompt写得太长。我一度以为system prompt越长越详细越好结果模型被一堆规则绕晕了反而忽略了核心指令。后来精简到只保留最关键的几条效果反而更好。第二个坑是忽略模型的幻觉工具。模型有时会编造一个不存在的工具名来调用如果不校验直接执行就会报错。所以执行前一定要检查工具名是否在注册表里。第三个坑是流式输出和工具调用混用时的顺序问题。有一次工具调用的结果还没返回模型就开始输出最终答案了导致答案里引用了还没拿到的数据。后来在prompt里明确要求必须先拿到工具结果再回答问题才解决。第四个坑是没有做请求去重。用户手抖点了两次发送Agent就跑了两遍重复执行了写操作。后来在前端加了防抖后端加了请求ID去重。8. 架构演进从小demo到生产系统8.1 什么时候该加什么架构不是一开始就全上的得按需演进。我的经验是日请求量100以内单进程内存存状态够用日请求量1000以内加Redis存会话加日志和监控日请求量1万以内模型接入层独立加缓存和限流日请求量10万以上全面微服务化各层独立扩缩容别过早优化但也别等到系统崩了才想起来加东西。监控要早做至少把请求量、延迟、错误率、token消耗这几个指标盯住。8.2 可观测性怎么做AI应用的可观测性比传统应用难因为输出是不确定的。我的做法是全链路记录每次请求记录输入、输出、调用的工具、token消耗、耗时。这些数据既能用来排查问题也能用来做效果评估。评估是个大话题。简单做法是人工抽检定期看一批请求的输出质量。进阶做法是搞一个评估集用LLM as judge自动打分。再进阶就是A/B测试对比不同prompt、不同模型的效果。8.3 成本控制token就是钱成本控制要从架构层面考虑。几个有效的手段缓存高频请求的结果、用小模型处理简单任务、压缩上下文、限制max_tokens。我一般会设一个每日token预算超了就降级到更便宜的模型或者直接拒绝服务避免账单失控。还有个容易被忽略的点是工具调用的成本。有些工具本身就要花钱比如调第三方APIAgent如果反复调用成本会很高。所以工具层也要做限流和缓存。9. 关于架构设计的一点个人体会搭了这么多AI应用我最大的体会是架构的价值不在于它有多复杂而在于它让复杂的事情变得可管理。一张好的架构图应该让新人看一眼就知道哦原来请求是这么流转的让老手在改代码时知道这个改动会影响哪几层。我见过一些团队架构图画得花里胡哨各种高大上的组件堆了一堆结果实际代码里全揉在一起图是图代码是代码。这种架构图没有意义。真正有用的架构是能指导代码组织的架构是每一层都有明确边界和职责的架构。另外AI应用架构和传统架构最大的不同是要为不确定性留出空间。模型会变、输出会飘、工具会挂架构要能容纳这些不确定性而不是假设一切都会按预期运行。容错、降级、可观测这些在传统应用里可能是加分项在AI应用里是必选项。最后分享一个我常用的判断标准如果你把模型换掉、把工具换掉、把交互方式换掉业务逻辑层需要改多少改得越少说明架构分层越成功。这个标准帮我避免了很多次为了用新技术而过度设计的冲动。
返回列表