尧图网页设计 
1. 从一张架构图说起AI应用到底该怎么分层很多人第一次接触AI应用架构设计脑子里冒出来的第一个念头是“不就是调个API吗”。我刚开始做AI应用的时候也这么想觉得前端发个请求后端拼个prompt调一下大模型接口把返回结果渲染出来就完事了。直到我接手一个需要同时对接三个模型供应商、支持多轮对话记忆、还要接入外部工具调用的项目才发现事情远没有这么简单。那次项目上线第一周就出了状况用户连续对话到第七轮时上下文长度直接超了模型的token上限整个会话崩掉另一个问题是工具调用的返回结果格式不稳定导致解析逻辑频繁报错。这两个问题逼着我重新思考AI应用到底该怎么架构。所谓AI应用架构设计本质上是在回答一个问题当你的系统核心能力来自大模型时传统的分层架构还够不够用我的结论是传统三层架构表现层、业务逻辑层、数据层依然有效但需要在每一层里嵌入AI特有的设计模式。具体来说AI应用架构可以拆成五个层次来理解从下往上依次是模型接入层、能力编排层、记忆与状态层、工具与协议层、交互与呈现层。这五层不是必须全部都有但当你开始做稍微复杂一点的AI应用时它们会自然而然地浮现出来。先说说模型接入层。这一层解决的是“怎么跟大模型说话”的问题。看起来简单实际上坑很多。不同供应商的API格式不一样有的用OpenAI兼容格式有的用自己的一套流式输出的实现方式不同错误码和限流策略也不同。我见过不少项目在这一层写死了某一家供应商的SDK结果后来想换模型或者做多模型路由时改造成本极高。合理的做法是在这一层做一层薄薄的抽象定义统一的请求和响应结构把供应商差异封装在适配器里。这样上层业务代码不需要关心底层用的是哪家模型。能力编排层是AI应用架构里最容易被低估的一层。很多人把prompt直接写在业务代码里散落在各个角落改一个prompt要翻半天代码。更麻烦的是当你需要做多步骤推理、条件分支、循环调用时散落的prompt根本没法维护。这一层应该承担起“AI工作流”的职责把prompt模板、调用顺序、条件判断、结果聚合都集中管理。你可以用代码实现也可以用专门的编排框架关键是要让AI能力的组合方式变得可读、可改、可测试。记忆与状态层是区分“玩具项目”和“生产级应用”的分水岭。没有记忆的AI应用每次对话都是重新开始用户体验很差。但记忆不是简单地把历史消息全部塞进上下文那样很快就会超出token限制。你需要设计记忆的存储、检索、压缩和淘汰策略。短期记忆可以用滑动窗口长期记忆需要向量化存储和相似度检索。这一层的设计质量直接决定了你的应用能不能支撑长对话和个性化体验。工具与协议层是最近一年变化最大的部分。大模型本身只能生成文本但用户需要的是它能查天气、发邮件、操作数据库。工具调用让模型能够与外部世界交互。而MCP协议的出现让工具的定义和发现有了标准化的方式。你可以把MCP理解成“AI应用的外设接口标准”就像USB让各种设备都能插到电脑上一样MCP让各种工具都能被AI应用统一调用。这一层的架构设计要考虑工具的注册、发现、权限控制和错误处理。交互与呈现层是最上面一层也是用户直接感知到的部分。流式输出、打字机效果、思考过程展示、引用来源标注这些细节决定了用户觉得你的应用“聪明”还是“笨”。这一层还要处理多模态输入输出的问题比如图片、语音、文件上传等。架构上要考虑前端如何高效地消费流式数据如何管理会话状态如何处理中断和重试。把这五层串起来看AI应用架构设计的核心思想是把“不确定性”控制在最小范围内。大模型的输出是不确定的但你的系统架构应该是确定的。通过分层你把模型的不确定性封装在特定层里让其他层保持稳定和可测试。这就是为什么我说AI应用架构设计不是简单地调API而是一套完整的工程方法论。2. 模型接入层的适配器模式与多模型路由策略模型接入层看起来是最简单的一层但如果你在这里偷懒后面会付出成倍的代价。我见过太多项目直接在业务代码里写openai.ChatCompletion.create(...)等到需要换模型或者加一个备用供应商时发现要改几十个文件。这一层的核心设计原则是面向接口编程把供应商差异关进适配器的笼子里。2.1 统一请求响应结构的定义先定义一套你自己的请求和响应结构。请求结构至少包含这些字段模型标识、消息列表、温度参数、最大输出token数、是否流式、工具定义列表。响应结构至少包含生成内容、结束原因、token使用量、工具调用请求。这套结构不依赖任何供应商的SDK是你自己的领域模型。from dataclasses import dataclass, field from typing import Optional dataclass class LLMRequest: model: str messages: list temperature: float 0.7 max_tokens: int 2048 stream: bool False tools: Optional[list] None dataclass class LLMResponse: content: str finish_reason: str prompt_tokens: int completion_tokens: int tool_calls: Optional[list] None定义好这套结构后每个供应商写一个适配器负责把你的LLMRequest翻译成供应商的API格式再把供应商的响应翻译回你的LLMResponse。这样上层代码只依赖你自己的结构换供应商只需要加一个适配器不改业务代码。2.2 多模型路由的决策逻辑多模型路由不是简单地随机选一个而是要根据任务类型、成本预算、延迟要求来动态决策。我的经验是把路由策略分成三类按任务复杂度路由、按成本路由、按可用性路由。按任务复杂度路由是最常用的。简单任务比如意图分类、关键词提取用小模型就够了速度快成本低复杂任务比如长文推理、代码生成用大模型效果更好。你可以在请求里加一个task_complexity字段路由层根据这个字段选择模型。按成本路由适合对成本敏感的场景。你可以给每个模型设置一个“每千token成本”的权重在满足质量要求的前提下优先选便宜的。但要注意便宜模型可能需要更多轮对话才能完成任务综合成本不一定低。按可用性路由是兜底策略。当主模型供应商出现限流或故障时自动切换到备用供应商。这里的关键是要做健康检查不能等到用户请求失败了才切换。我的做法是维护一个滑动窗口统计每个供应商最近的成功率和延迟低于阈值就自动降级。class ModelRouter: def __init__(self, adapters: dict, health_checker): self.adapters adapters self.health_checker health_checker def route(self, request: LLMRequest) - str: # 优先选择健康的供应商 healthy [name for name, adapter in self.adapters.items() if self.health_checker.is_healthy(name)] if not healthy: raise NoAvailableProviderError() # 根据任务复杂度选择 if request.task_complexity high: candidates [n for n in healthy if n in (gpt-4, claude-3-opus)] else: candidates [n for n in healthy if n in (gpt-3.5, claude-3-haiku)] # 按成本排序 return min(candidates, keylambda n: self.adapters[n].cost_per_1k)2.3 流式输出的统一处理流式输出是AI应用体验的关键但不同供应商的流式格式差异很大。有的用SSE有的用WebSocket有的返回的chunk结构完全不同。适配器层要把这些差异抹平向上层提供统一的流式接口。我的做法是定义一个StreamChunk结构包含delta_content、delta_tool_calls、finish_reason三个字段。适配器负责把供应商的原始chunk转换成这个结构。上层代码只需要遍历这个统一的流不需要关心底层是哪个供应商。这里有个容易踩的坑流式输出和工具调用同时使用时工具调用的参数是分多个chunk传过来的需要拼接。我见过有项目在流式模式下工具调用一直失败就是因为没有正确处理参数拼接。正确的做法是维护一个缓冲区把同一个工具调用的所有参数chunk拼起来等到finish_reason为tool_calls时再解析。提示流式模式下token使用量统计往往在最后一个chunk才返回如果你的计费逻辑依赖token数记得在流结束时再结算不要每个chunk都算一次。2.4 错误处理与重试的边界模型接入层的错误处理有个反直觉的地方不是所有错误都值得重试。限流错误429和超时错误值得重试但参数错误400和认证错误401重试多少次都没用。我的策略是把错误分成三类可重试错误、不可重试错误、需要降级的错误。可重试错误包括限流、超时、服务端5xx。这类错误用指数退避重试最多三次。不可重试错误包括参数格式错误、认证失败、模型不存在。这类错误直接抛给上层让业务逻辑决定怎么处理。需要降级的错误包括上下文超长、内容审核不通过。这类错误需要触发降级逻辑比如压缩上下文或者换一个模型重试。RETRYABLE_ERRORS {429, 500, 502, 503, 504} FATAL_ERRORS {400, 401, 403, 404} def call_with_retry(adapter, request, max_retries3): for attempt in range(max_retries): try: return adapter.call(request) except LLMError as e: if e.status_code in FATAL_ERRORS: raise if e.status_code in RETRYABLE_ERRORS and attempt max_retries - 1: time.sleep(2 ** attempt) continue raise这一层的设计质量直接决定了你的AI应用能不能在真实生产环境里稳定运行。我个人的经验是模型接入层的代码量不应该超过整个项目的20%如果超过了说明抽象没做好供应商差异泄漏到了上层。3. 能力编排层把散落的prompt变成可维护的工作流能力编排层是AI应用架构里最体现工程水平的一层。我见过两种极端一种是把所有prompt硬编码在业务逻辑里改一个标点符号都要重新部署另一种是过度设计搞了一套复杂的DSL结果团队里没人会用。我的观点是编排层的设计要遵循“够用就好”的原则先解决可维护性问题再考虑灵活性。3.1 Prompt模板的集中管理与版本控制Prompt是AI应用的“源代码”但很多人对待prompt的态度远不如对待代码认真。代码有版本控制、有code review、有测试prompt却经常是随手改改就上线。我的做法是把prompt当成代码来管理集中存放在一个目录里用模板引擎支持变量替换每次修改都走版本控制。# prompts/summarize.yaml name: summarize version: 3 template: | 你是一个专业的文本摘要助手。请对以下内容进行摘要 要求 1. 保留关键信息不超过{max_length}字 2. 使用{language}输出 3. 如果内容包含数字必须准确保留 内容 {content}用YAML或JSON管理prompt的好处是非技术人员也能参与修改而且可以方便地做A/B测试。你可以给每个prompt打上版本号在运行时根据配置选择版本。这样当新版本效果不好时可以快速回滚。这里有个实操心得prompt的变量替换一定要做转义处理。我踩过一次坑用户输入的内容里包含了模板语法字符导致渲染出来的prompt结构错乱。后来我在渲染前对所有变量值做了转义问题才解决。3.2 多步骤推理的编排模式单次模型调用能解决的问题很有限真实场景往往需要多步骤推理。比如一个客服场景先判断用户意图再根据意图检索知识库然后把检索结果和用户问题一起交给模型生成回答最后做一次质量检查。这四步如果写在一个prompt里效果往往不好因为模型很难同时做好这么多事。我的经验是把多步骤推理拆成独立的步骤每个步骤一个prompt步骤之间用代码串联。这样每个prompt的职责单一调试起来也容易。编排层要提供几种基本的编排模式顺序执行、条件分支、循环迭代、并行聚合。顺序执行最简单就是A步骤的输出作为B步骤的输入。条件分支是根据上一步的结果决定下一步走哪条路。循环迭代适合需要反复优化的场景比如让模型生成代码然后运行测试根据测试结果让模型修复直到测试通过或达到最大迭代次数。并行聚合是把一个任务拆成多个子任务同时执行最后把结果合并。class Workflow: def __init__(self): self.steps [] def add_step(self, name, prompt_name, conditionNone): self.steps.append({name: name, prompt: prompt_name, condition: condition}) return self def run(self, context): for step in self.steps: if step[condition] and not step[condition](context): continue prompt render_prompt(step[prompt], context) result llm_call(prompt) context[step[name]] result return context3.3 输出解析与结构化约束大模型的输出是自然语言但你的下游代码需要结构化数据。这个矛盾是AI应用开发中最常见的痛点。我的解决方案是三层保障prompt约束、解析容错、重试修复。Prompt约束是最基本的在prompt里明确要求输出JSON格式并给出schema示例。但光靠prompt约束不够可靠模型有时候会加一些解释性文字或者JSON格式有细微错误。所以需要解析容错用宽松的解析器能处理常见的格式问题比如多余的逗号、单引号、markdown代码块包裹等。import json import re def parse_json_output(text): # 去掉markdown代码块标记 text re.sub(r^(?:json)?\s*, , text.strip()) text re.sub(r\s*$, , text) # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个完整的JSON对象 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return None如果解析还是失败就需要重试修复。把解析错误信息和原始输出一起发给模型让它修复格式。这个重试通常一次就能成功因为模型看到具体错误后能理解问题所在。注意重试修复会增加延迟和成本所以要在prompt约束和解析容错上多下功夫把重试率控制在5%以下。3.4 编排层的可观测性设计编排层是AI应用里最需要可观测性的地方。一个请求经过多个步骤每个步骤的输入输出、耗时、token消耗都需要记录。没有这些数据你根本不知道问题出在哪一步。我的做法是在编排层埋点每个步骤执行时记录一条结构化日志包含步骤名称、输入prompt的hash、输出内容的hash、耗时、token数、是否成功。这些日志汇总后可以分析出很多有价值的信息哪个步骤最耗时、哪个prompt的token消耗最大、哪个步骤的失败率最高。import time import hashlib def execute_step(step_name, prompt, context): start time.time() try: result llm_call(prompt) log_step( stepstep_name, prompt_hashhashlib.md5(prompt.encode()).hexdigest(), output_hashhashlib.md5(result.encode()).hexdigest(), durationtime.time() - start, tokensresult.usage.total_tokens, statussuccess ) return result except Exception as e: log_step(stepstep_name, statuserror, errorstr(e)) raise这些数据积累一段时间后你就能做出有依据的优化决策。比如发现某个步骤平均耗时3秒占了整个请求的60%那优化这个步骤的收益最大。或者发现某个prompt的token消耗是其他prompt的5倍但输出质量没有明显提升那就可以考虑精简这个prompt。4. 记忆与状态层让AI应用记住该记住的没有记忆的AI应用就像金鱼每次对话都是全新的开始。但记忆不是简单地把历史消息全部塞进上下文那样做有两个问题一是token成本随对话轮数线性增长二是过长的上下文会稀释模型的注意力导致它忽略关键信息。记忆与状态层的设计目标是在有限的token预算内让模型获得最相关的历史信息。4.1 短期记忆的滑动窗口与摘要压缩短期记忆管理的是当前会话的上下文。最简单的策略是滑动窗口只保留最近N轮对话。N的取值取决于你的token预算和平均每轮对话的长度。如果每轮对话平均200token模型上下文窗口是8K那N可以取20左右留出空间给系统prompt和当前输入。但滑动窗口有个问题早期对话里的关键信息会被丢掉。比如用户在第三轮说了自己的名字到第二十轮时窗口已经滑过去了模型就忘了用户叫什么。解决这个问题有两种思路一种是关键信息提取在对话过程中实时提取实体和事实存到一个结构化的“用户档案”里每次请求都把这个档案带上另一种是滚动摘要当对话轮数超过窗口大小时把最早的一批对话压缩成摘要摘要和窗口内的原始对话一起送给模型。class ShortTermMemory: def __init__(self, max_turns20, max_tokens4000): self.turns [] self.summary self.max_turns max_turns self.max_tokens max_tokens def add(self, role, content): self.turns.append({role: role, content: content}) if len(self.turns) self.max_turns: self._compress() def _compress(self): # 把最早的一半对话压缩成摘要 to_compress self.turns[:len(self.turns) // 2] self.turns self.turns[len(self.turns) // 2:] summary_prompt f请将以下对话压缩成简洁的摘要保留关键信息\n{to_compress} new_summary llm_call(summary_prompt) self.summary f{self.summary}\n{new_summary} if self.summary else new_summary def get_context(self): context [] if self.summary: context.append({role: system, content: f之前的对话摘要{self.summary}}) context.extend(self.turns) return context4.2 长期记忆的向量化存储与检索长期记忆解决的是跨会话的信息保留问题。用户上周跟你聊过的话题这周再聊时你应该能想起来。实现方式是把对话内容向量化后存入向量数据库检索时用当前输入去查询最相关的历史片段。这里的关键是“什么该存”。不是所有对话都值得存入长期记忆。我的经验是只存三类内容用户明确表达的偏好和事实、重要的决策和结论、反复出现的话题。日常的寒暄和无关紧要的对话不需要存存了反而会干扰检索。向量化存储的另一个关键是分块策略。太长的文本块检索精度低太短的块又缺乏上下文。我的做法是按语义完整性分块每个块200到500字块之间保留一定的重叠。检索时返回top-k个最相关的块k一般取3到5。class LongTermMemory: def __init__(self, vector_store, embedder): self.vector_store vector_store self.embedder embedder def store(self, user_id, content, metadataNone): # 只存储有价值的内容 if not self._is_worth_storing(content): return embedding self.embedder.embed(content) self.vector_store.upsert( user_iduser_id, vectorembedding, contentcontent, metadatametadata or {} ) def retrieve(self, user_id, query, top_k5): query_vec self.embedder.embed(query) results self.vector_store.search(user_id, query_vec, top_k) return [r.content for r in results if r.score 0.7] def _is_worth_storing(self, content): # 过滤掉太短、纯寒暄、无信息量的内容 if len(content) 20: return False trivial [好的, 谢谢, 嗯, 收到] if content.strip() in trivial: return False return True4.3 状态管理的会话隔离与并发安全AI应用的状态管理有个容易被忽视的问题并发。当同一个用户同时发起多个请求时如果状态管理没有做好隔离会出现数据串扰。比如用户A的对话历史被写到了用户B的会话里这种bug在生产环境里是灾难性的。我的做法是每个会话有一个唯一的session_id所有状态操作都必须带上session_id。状态存储用带乐观锁的键值存储每次更新时检查版本号版本号不匹配就重试。这样即使有并发请求也不会互相覆盖。class SessionState: def __init__(self, store, session_id): self.store store self.session_id session_id def update(self, key, value): max_retries 3 for _ in range(max_retries): state, version self.store.get_with_version(self.session_id) state[key] value if self.store.compare_and_set(self.session_id, state, version): return raise ConcurrentModificationError()提示如果你的应用需要支持多设备同步状态存储要选择支持跨设备访问的方案并且要考虑冲突解决策略。简单的做法是“最后写入胜出”但更好的做法是让用户选择保留哪个版本。4.4 记忆的遗忘机制与隐私边界记忆不是越多越好。过时的、不相关的记忆会干扰模型的判断。所以需要设计遗忘机制。我的策略是给每条记忆设置一个“新鲜度”分数随着时间衰减。检索时优先返回新鲜度高的记忆低于阈值的记忆定期清理。隐私边界是另一个必须考虑的问题。用户的敏感信息比如密码、身份证号、银行卡号绝对不能存入长期记忆。我的做法是在存储前做一次敏感信息检测命中规则的内容直接丢弃并记录一条审计日志。同时给用户提供“清除我的记忆”的功能让用户对自己的数据有控制权。SENSITIVE_PATTERNS [ r\d{17}[\dXx], # 身份证 r\d{16,19}, # 银行卡 rpassword\s*[:]\s*\S, ] def contains_sensitive(text): for pattern in SENSITIVE_PATTERNS: if re.search(pattern, text): return True return False记忆与状态层的设计质量直接决定了用户觉得你的AI应用是“聪明”还是“健忘”。我个人的经验是这一层的投入产出比很高花时间把记忆管理做好用户体验的提升非常明显。5. 工具与协议层MCP如何统一AI与外部世界的接口工具调用是AI应用从“聊天机器人”进化到“智能助手”的关键一步。没有工具调用模型只能基于训练数据回答问题有了工具调用模型可以查实时数据、操作外部系统、执行具体任务。但工具调用的实现方式一直很混乱每个框架有自己的定义格式每个模型供应商有自己的调用协议。MCP协议的出现让这个领域有了标准化的希望。5.1 工具定义的标准结构与注册机制在MCP之前工具定义是各家各户自己定的。OpenAI有一套function calling的格式Anthropic有自己的tool use格式开源框架又有自己的定义。这导致一个工具要在多个模型上使用需要写多份定义。MCP的核心价值就是定义了一套标准的工具描述格式让工具的定义和调用解耦。一个标准的工具定义包含这几个部分工具名称、功能描述、参数schema、返回值说明。工具名称要简洁明确功能描述要写清楚“什么时候该用这个工具”因为模型是根据描述来决定是否调用的。参数schema用JSON Schema定义要写清楚每个参数的类型、是否必填、取值范围。{ name: query_weather, description: 查询指定城市的实时天气。当用户询问天气相关问题时使用此工具。, parameters: { type: object, properties: { city: { type: string, description: 城市名称如北京、上海 }, unit: { type: string, enum: [celsius, fahrenheit], default: celsius } }, required: [city] } }工具注册机制要解决的是“工具从哪来”的问题。我的做法是维护一个工具注册表每个工具是一个独立的模块启动时自动注册。注册表提供查询接口编排层根据当前任务需要把相关工具的定义传给模型。class ToolRegistry: def __init__(self): self.tools {} def register(self, name, description, parameters, handler): self.tools[name] { definition: { name: name, description: description, parameters: parameters }, handler: handler } def get_definitions(self, namesNone): if names: return [self.tools[n][definition] for n in names if n in self.tools] return [t[definition] for t in self.tools.values()] def execute(self, name, arguments): if name not in self.tools: raise ToolNotFoundError(name) return self.tools[name][handler](**arguments)5.2 MCP协议的核心概念与工作流程MCP的全称是Model Context Protocol它定义了一套客户端-服务端的交互协议让AI应用能够以统一的方式发现和调用外部工具。你可以把MCP服务端理解成一个“工具提供者”它声明自己有哪些工具、每个工具需要什么参数MCP客户端是“工具消费者”它连接服务端获取工具列表在需要时调用工具。MCP的工作流程分三步发现、调用、返回。发现阶段客户端连接服务端获取工具列表和每个工具的定义。调用阶段客户端把工具名称和参数发给服务端服务端执行工具并返回结果。返回阶段结果以标准格式返回给客户端客户端再把它交给模型。class MCPClient: def __init__(self, server_url): self.server_url server_url self.tools [] def discover(self): response requests.get(f{self.server_url}/tools) self.tools response.json()[tools] return self.tools def call(self, tool_name, arguments): response requests.post( f{self.server_url}/call, json{name: tool_name, arguments: arguments} ) return response.json()MCP相比传统的function calling有几个优势。第一是标准化工具的定义和调用格式统一不同模型和框架可以复用同一套工具。第二是解耦工具的提供者和消费者分离工具可以独立部署和升级。第三是可组合多个MCP服务端可以同时连接客户端可以聚合多个来源的工具。5.3 工具调用的安全边界与权限控制工具调用让模型能够操作外部系统这带来了安全风险。如果权限控制没做好模型可能被诱导执行危险操作。我见过一个案例客服机器人的工具里有“发送邮件”功能结果被用户诱导给全公司发了垃圾邮件。安全控制的第一道防线是工具白名单。不是所有工具都对所有用户开放。根据用户的角色和场景只暴露必要的工具。比如普通用户只能查天气、查订单管理员才能操作数据库。第二道防线是参数校验。模型生成的参数不可信必须做严格的校验。比如查询订单的工具订单号必须符合格式且必须属于当前用户。参数校验要在工具执行前做不通过直接拒绝。def execute_tool_safely(tool_name, arguments, user_context): # 检查工具是否在用户的允许列表中 if tool_name not in user_context.allowed_tools: raise PermissionDeniedError(f用户无权调用工具 {tool_name}) # 参数校验 tool registry.get(tool_name) validate_arguments(arguments, tool[definition][parameters]) # 注入用户上下文防止越权 if user_id in tool[definition][parameters][properties]: arguments[user_id] user_context.user_id return registry.execute(tool_name, arguments)第三道防线是操作审计。所有工具调用都要记录日志包括谁调的、调了什么、参数是什么、结果是什么。这样出了问题可以追溯也能发现异常调用模式。注意对于有副作用的工具比如写操作、发送操作建议加一道人工确认。模型生成调用请求后先展示给用户确认用户点了确认才真正执行。这样能避免模型误操作。5.4 工具编排与多工具协作真实场景往往需要多个工具协作完成一个任务。比如用户说“帮我订明天去上海的机票然后告诉张三”这需要调用航班查询、机票预订、联系人查询、消息发送四个工具。工具编排层要负责决定调用顺序、传递中间结果、处理失败情况。我的做法是把工具编排也做成工作流。每个步骤可以是一个工具调用也可以是模型推理。步骤之间通过上下文传递数据。编排层要处理几种情况串行调用A的结果是B的输入、并行调用A和B同时执行、条件调用根据A的结果决定是否调B。class ToolOrchestrator: def __init__(self, llm, registry): self.llm llm self.registry registry def run(self, user_input, max_steps10): context {user_input: user_input, history: []} for _ in range(max_steps): # 让模型决定下一步 decision self.llm.decide_next_action( context, self.registry.get_definitions() ) if decision.action final_answer: return decision.content elif decision.action tool_call: result self.registry.execute( decision.tool_name, decision.arguments ) context[history].append({ tool: decision.tool_name, result: result }) return 任务步骤过多已终止工具编排的难点在于错误处理。工具调用可能失败失败后是重试、换工具、还是告诉用户我的策略是可重试的错误自动重试一次不可重试的错误把错误信息返回给模型让模型决定下一步。模型看到错误信息后可能会换一个工具或者告诉用户当前无法完成。6. 交互与呈现层流式输出、多模态与用户体验细节交互与呈现层是用户唯一直接感知的层。同样的模型能力交互设计的好坏会让用户体验天差地别。我见过两个功能完全相同的AI应用一个用户留存率是另一个的三倍差别就在交互细节上。6.1 流式输出的前端消费与渲染优化流式输出是AI应用体验的基石。用户不需要等整个回答生成完而是看到文字一个字一个字地出现这种即时反馈大大降低了等待焦虑。但流式输出的前端实现有不少坑。第一个坑是SSE连接的管理。SSE是单向的服务端推、客户端收。但用户可能在中途取消生成或者网络中断。前端要能正确处理这些情况取消时关闭连接中断时自动重连并带上最后收到的位置。第二个坑是渲染性能。如果每个字符都触发一次DOM更新页面会卡顿。正确的做法是用缓冲区积累一批字符再统一更新。通常每50毫秒更新一次或者每积累10个字符更新一次。class StreamRenderer { constructor(element) { this.element element; this.buffer ; this.timer null; } append(text) { this.buffer text; if (!this.timer) { this.timer setTimeout(() this.flush(), 50); } } flush() { this.element.textContent this.buffer; this.buffer ; this.timer null; this.scrollToBottom(); } scrollToBottom() { this.element.scrollTop this.element.scrollHeight; } }第三个坑是markdown渲染。模型输出的是markdown格式但流式过程中markdown是不完整的比如代码块只收到了一半。如果直接渲染会出现格式错乱。我的做法是流式过程中用纯文本渲染等流结束后再做一次完整的markdown渲染。6.2 多模态输入输出的架构设计多模态是AI应用的发展趋势。用户不再满足于打字他们想上传图片、语音、文件。架构上要支持这些输入类型并把它们统一转换成模型能理解的格式。图片输入的处理流程是前端上传图片到对象存储拿到URL后端把URL转换成模型能接受的格式通常是base64或URL模型返回图片描述或基于图片的回答。这里要注意图片大小限制太大的图片要先压缩。语音输入需要先做语音识别把音频转成文本再走文本对话流程。语音识别可以用专门的ASR服务也可以用多模态模型直接处理音频。我的经验是对于实时性要求高的场景用专门的ASR服务延迟更低对于需要理解语气和情感的場景用多模态模型效果更好。文件输入要区分类型。PDF和Word需要先提取文本图片需要OCR表格需要解析成结构化数据。这些预处理步骤可以在上传时异步完成等用户真正提问时直接使用处理好的文本。class MultimodalInput: def process(self, input_data): if input_data.type image: return self._process_image(input_data) elif input_data.type audio: return self._process_audio(input_data) elif input_data.type file: return self._process_file(input_data) else: return input_data.content def _process_image(self, data): # 压缩图片 compressed compress_image(data.content, max_size1024) # 转成base64 return {type: image_url, image_url: {url: to_base64(compressed)}} def _process_file(self, data): if data.mime_type application/pdf: text extract_pdf_text(data.content) elif data.mime_type.startswith(image/): text ocr(data.content) else: text data.content.decode(utf-8, errorsignore) return {type: text, text: text}6.3 思考过程展示与引用来源标注用户对AI的信任建立在透明之上。如果模型直接给一个答案用户不知道这个答案是怎么来的信任度会打折扣。展示思考过程和引用来源能显著提升用户信任。思考过程展示有两种方式一种是展示模型的推理链让用户看到模型是怎么一步步得出结论的另一种是展示工具调用过程让用户知道模型查了哪些数据。我的做法是在界面上加一个可折叠的“思考过程”区域默认折叠用户想看的时候可以展开。引用来源标注是另一个提升信任的手段。当模型基于检索到的文档回答时在回答中标注引用编号用户点击编号可以看到原文。实现方式是在prompt里要求模型用[1]、[2]这样的格式标注引用前端解析这些标记并渲染成可点击的链接。def format_with_citations(answer, sources): # 把 [1] 这样的标记替换成可点击的链接 for i, source in enumerate(sources, 1): answer answer.replace( f[{i}], fa href{source.url} target_blank[{i}]/a ) return answer6.4 错误状态与降级体验的设计AI应用不可能永远不出错。模型超时、限流、返回格式错误这些都会发生。关键是出错时用户体验不能崩。我的设计原则是永远给用户一个可操作的下一步。如果模型超时不要只显示“请求失败”而是显示“正在重试”并自动重试。如果重试也失败给用户两个选项重新发送或者简化问题。如果模型返回的内容被安全过滤了不要显示“内容违规”而是显示“这个问题我暂时无法回答你可以换个方式问问”。function handleError(error) { if (error.type timeout) { showMessage(响应时间较长正在重试...); retry(); } else if (error.type rate_limit) { showMessage(当前请求较多请稍后再试); showRetryButton(); } else if (error.type content_filter) { showMessage(这个问题我暂时无法回答你可以换个方式问问); } else { showMessage(出了点小问题请重试); showRetryButton(); } }降级体验还包括功能降级。当高级模型不可用时自动切换到基础模型并在界面上提示“当前使用基础模式”。当工具调用失败时告诉用户“暂时无法查询实时数据以下是基于已有知识的回答”。交互与呈现层的设计没有标准答案核心是站在用户角度思考用户在这个时刻需要什么信息、需要什么操作。把这两个问题回答好体验就不会差。7. 从架构图到生产部署、监控与迭代的实战经验架构设计得再好最终要落到生产环境里跑起来。这一章我想分享一些从实际项目中总结的经验都是踩过坑之后才明白的道理。7.1 部署架构的选型与成本控制AI应用的部署架构和传统Web应用有个显著区别GPU资源。如果你用的是自部署模型GPU成本是大头如果你用的是API那成本主要在token消耗上。两种情况的优化策略完全不同。用API的情况下成本控制的核心是减少不必要的token消耗。我做过一个统计一个AI客服应用里40%的token消耗在系统prompt上。系统prompt往往很长包含角色设定、行为规范、工具定义等。优化方法是把系统prompt里不变的部分缓存起来很多供应商支持prompt caching能省不少钱。自部署模型的情况下GPU利用率是关键。我的经验是把推理服务做成无状态的前面加一个队列请求来了先入队推理服务从队列里取任务。这样能平滑流量波动避免GPU忽忙忽闲。队列的长度要监控太长说明GPU不够太短说明资源浪费。# docker-compose.yml 示例 services: api: image: myapp-api environment: - LLM_PROVIDERopenai - REDIS_URLredis://redis:6379 depends_on: - redis worker: image: myapp-worker environment: - QUEUE_URLredis://redis:6379 deploy: replicas: 3 redis: image: redis:7-alpine7.2 监控指标除了QPS和延迟还要看什么传统应用的监控指标是QPS、延迟、错误率。AI应用还需要关注一些特有指标。第一个是token消耗速率按小时统计突然飙升往往意味着有异常调用。第二个是上下文长度分布如果平均上下文长度持续增长说明记忆管理可能有问题。第三个是工具调用成功率这个指标下降说明外部依赖出了问题。第四个指标是“有效回答率”。什么是有效回答用户没有在回答后立刻重新提问没有点“踩”没有关闭会话。这个指标比单纯的“请求成功率”更能反映真实质量。我通常用“用户追问率”作为代理指标追问率高说明上一次回答没解决问题。class MetricsCollector: def record_request(self, request, response, duration): self.histogram(request_duration, duration) self.counter(token_usage, response.usage.total_tokens) self.counter(context_length, len(request.messages)) if response.tool_calls: self.counter(tool_calls, len(response.tool_calls)) def record_feedback(self, session_id, feedback_type): self.counter(ffeedback_{feedback_type}) # 计算追问率 if feedback_type follow_up: self.counter(follow_up_rate)7.3 灰度发布与Prompt迭代的安全流程Prompt的修改比代码修改更危险因为prompt的效果很难用单元测试验证。我的做法是prompt修改必须走灰度发布。新prompt先对5%的流量生效观察核心指标有效回答率、追问率、token消耗有没有恶化。如果指标正常逐步扩大到20%、50%、100%。如果指标恶化立即回滚。灰度发布的关键是流量分割要一致。同一个用户应该始终看到同一个版本的prompt否则体验会不一致。我的做法是用user_id做hashhash值落在灰度区间内的用户用新prompt其他的用旧prompt。def select_prompt_version(user_id, prompt_name): versions get_prompt_versions(prompt_name) if len(versions) 1: return versions[0] # 灰度10%流量用新版本 hash_val int(hashlib.md5(user_id.encode()).hexdigest(), 16) if hash_val % 100 10: return versions[-1] # 新版本 return versions[0] # 稳定版本7.4 常见线上问题的排查链路线上问题排查最怕没有线索。我的经验是提前埋好日志每个请求记录完整的调用链用户输入、系统prompt、模型输出、工具调用、最终响应。这样出问题时能快速定位。一个常见问题是“模型回答质量突然下降”。排查链路是先看是不是prompt被改了对比当前prompt和昨天的版本再看是不是模型供应商更新了模型很多供应商会静默更新模型版本然后看是不是检索到的上下文质量下降了检查向量检索的召回率和相关性分数最后看是不是用户输入分布变了统计最近一周的用户输入类型。另一个常见问题是“响应变慢”。排查链路是先看模型API的延迟如果模型本身慢了那是供应商的问题再看工具调用的延迟某个外部服务变慢会拖累整个请求然后看队列等待时间如果队列积压说明需要扩容最后看数据库查询记忆检索的查询可能随着数据量增长而变慢。def diagnose_slow_request(request_id): trace get_trace(request_id) timings { queue_wait: trace.queue_wait, model_call: trace.model_call_duration, tool_calls: sum(t.duration for t in trace.tool_calls), memory_retrieval: trace.memory_retrieval_duration, total: trace.total_duration } # 找出耗时最长的环节 bottleneck max(timings, keytimings.get) return f瓶颈在 {bottleneck}耗时 {timings[bottleneck]:.2f}s7.5 架构演进从单体到分层的实际路径最后聊聊架构演进。没有人一开始就设计出完美的五层架构。我的实际路径是第一版把所有逻辑写在一个文件里能跑就行第二版把prompt抽出来模型调用封装成函数第三版加入记忆管理把会话状态独立出来第四版加入工具调用把工具注册和执行分离第五版引入MCP把工具层标准化。每一步演进都是被需求驱动的不是为了架构而架构。我的建议是不要一开始就追求完美的分层先让功能跑起来然后在遇到痛点时重构。重构的时机是当你发现“改一个地方要动很多文件”或者“加一个新功能要复制粘贴很多代码”的时候。架构演进的另一个经验是保持每一层的可替换性。模型接入层可以换供应商能力编排层可以换框架记忆层可以换存储工具层可以换协议。只要层与层之间的接口稳定内部实现怎么换都行。这种可替换性让你在技术选型变化时不会推倒重来。我在实际项目里最深的体会是AI应用架构设计的核心不是技术而是对“不确定性”的管理。模型输出不确定、工具返回不确定、用户输入不确定。好的架构把这些不确定性隔离在特定层里让系统的其他部分保持确定和可控。这才是架构设计的价值所在。 AI应用架构设计：从模型接入到交互呈现的五层架构实战
1. 从一张架构图说起AI应用到底该怎么分层很多人第一次接触AI应用架构设计脑子里冒出来的第一个念头是“不就是调个API吗”。我刚开始做AI应用的时候也这么想觉得前端发个请求后端拼个prompt调一下大模型接口把返回结果渲染出来就完事了。直到我接手一个需要同时对接三个模型供应商、支持多轮对话记忆、还要接入外部工具调用的项目才发现事情远没有这么简单。那次项目上线第一周就出了状况用户连续对话到第七轮时上下文长度直接超了模型的token上限整个会话崩掉另一个问题是工具调用的返回结果格式不稳定导致解析逻辑频繁报错。这两个问题逼着我重新思考AI应用到底该怎么架构。所谓AI应用架构设计本质上是在回答一个问题当你的系统核心能力来自大模型时传统的分层架构还够不够用我的结论是传统三层架构表现层、业务逻辑层、数据层依然有效但需要在每一层里嵌入AI特有的设计模式。具体来说AI应用架构可以拆成五个层次来理解从下往上依次是模型接入层、能力编排层、记忆与状态层、工具与协议层、交互与呈现层。这五层不是必须全部都有但当你开始做稍微复杂一点的AI应用时它们会自然而然地浮现出来。先说说模型接入层。这一层解决的是“怎么跟大模型说话”的问题。看起来简单实际上坑很多。不同供应商的API格式不一样有的用OpenAI兼容格式有的用自己的一套流式输出的实现方式不同错误码和限流策略也不同。我见过不少项目在这一层写死了某一家供应商的SDK结果后来想换模型或者做多模型路由时改造成本极高。合理的做法是在这一层做一层薄薄的抽象定义统一的请求和响应结构把供应商差异封装在适配器里。这样上层业务代码不需要关心底层用的是哪家模型。能力编排层是AI应用架构里最容易被低估的一层。很多人把prompt直接写在业务代码里散落在各个角落改一个prompt要翻半天代码。更麻烦的是当你需要做多步骤推理、条件分支、循环调用时散落的prompt根本没法维护。这一层应该承担起“AI工作流”的职责把prompt模板、调用顺序、条件判断、结果聚合都集中管理。你可以用代码实现也可以用专门的编排框架关键是要让AI能力的组合方式变得可读、可改、可测试。记忆与状态层是区分“玩具项目”和“生产级应用”的分水岭。没有记忆的AI应用每次对话都是重新开始用户体验很差。但记忆不是简单地把历史消息全部塞进上下文那样很快就会超出token限制。你需要设计记忆的存储、检索、压缩和淘汰策略。短期记忆可以用滑动窗口长期记忆需要向量化存储和相似度检索。这一层的设计质量直接决定了你的应用能不能支撑长对话和个性化体验。工具与协议层是最近一年变化最大的部分。大模型本身只能生成文本但用户需要的是它能查天气、发邮件、操作数据库。工具调用让模型能够与外部世界交互。而MCP协议的出现让工具的定义和发现有了标准化的方式。你可以把MCP理解成“AI应用的外设接口标准”就像USB让各种设备都能插到电脑上一样MCP让各种工具都能被AI应用统一调用。这一层的架构设计要考虑工具的注册、发现、权限控制和错误处理。交互与呈现层是最上面一层也是用户直接感知到的部分。流式输出、打字机效果、思考过程展示、引用来源标注这些细节决定了用户觉得你的应用“聪明”还是“笨”。这一层还要处理多模态输入输出的问题比如图片、语音、文件上传等。架构上要考虑前端如何高效地消费流式数据如何管理会话状态如何处理中断和重试。把这五层串起来看AI应用架构设计的核心思想是把“不确定性”控制在最小范围内。大模型的输出是不确定的但你的系统架构应该是确定的。通过分层你把模型的不确定性封装在特定层里让其他层保持稳定和可测试。这就是为什么我说AI应用架构设计不是简单地调API而是一套完整的工程方法论。2. 模型接入层的适配器模式与多模型路由策略模型接入层看起来是最简单的一层但如果你在这里偷懒后面会付出成倍的代价。我见过太多项目直接在业务代码里写openai.ChatCompletion.create(...)等到需要换模型或者加一个备用供应商时发现要改几十个文件。这一层的核心设计原则是面向接口编程把供应商差异关进适配器的笼子里。2.1 统一请求响应结构的定义先定义一套你自己的请求和响应结构。请求结构至少包含这些字段模型标识、消息列表、温度参数、最大输出token数、是否流式、工具定义列表。响应结构至少包含生成内容、结束原因、token使用量、工具调用请求。这套结构不依赖任何供应商的SDK是你自己的领域模型。from dataclasses import dataclass, field from typing import Optional dataclass class LLMRequest: model: str messages: list temperature: float 0.7 max_tokens: int 2048 stream: bool False tools: Optional[list] None dataclass class LLMResponse: content: str finish_reason: str prompt_tokens: int completion_tokens: int tool_calls: Optional[list] None定义好这套结构后每个供应商写一个适配器负责把你的LLMRequest翻译成供应商的API格式再把供应商的响应翻译回你的LLMResponse。这样上层代码只依赖你自己的结构换供应商只需要加一个适配器不改业务代码。2.2 多模型路由的决策逻辑多模型路由不是简单地随机选一个而是要根据任务类型、成本预算、延迟要求来动态决策。我的经验是把路由策略分成三类按任务复杂度路由、按成本路由、按可用性路由。按任务复杂度路由是最常用的。简单任务比如意图分类、关键词提取用小模型就够了速度快成本低复杂任务比如长文推理、代码生成用大模型效果更好。你可以在请求里加一个task_complexity字段路由层根据这个字段选择模型。按成本路由适合对成本敏感的场景。你可以给每个模型设置一个“每千token成本”的权重在满足质量要求的前提下优先选便宜的。但要注意便宜模型可能需要更多轮对话才能完成任务综合成本不一定低。按可用性路由是兜底策略。当主模型供应商出现限流或故障时自动切换到备用供应商。这里的关键是要做健康检查不能等到用户请求失败了才切换。我的做法是维护一个滑动窗口统计每个供应商最近的成功率和延迟低于阈值就自动降级。class ModelRouter: def __init__(self, adapters: dict, health_checker): self.adapters adapters self.health_checker health_checker def route(self, request: LLMRequest) - str: # 优先选择健康的供应商 healthy [name for name, adapter in self.adapters.items() if self.health_checker.is_healthy(name)] if not healthy: raise NoAvailableProviderError() # 根据任务复杂度选择 if request.task_complexity high: candidates [n for n in healthy if n in (gpt-4, claude-3-opus)] else: candidates [n for n in healthy if n in (gpt-3.5, claude-3-haiku)] # 按成本排序 return min(candidates, keylambda n: self.adapters[n].cost_per_1k)2.3 流式输出的统一处理流式输出是AI应用体验的关键但不同供应商的流式格式差异很大。有的用SSE有的用WebSocket有的返回的chunk结构完全不同。适配器层要把这些差异抹平向上层提供统一的流式接口。我的做法是定义一个StreamChunk结构包含delta_content、delta_tool_calls、finish_reason三个字段。适配器负责把供应商的原始chunk转换成这个结构。上层代码只需要遍历这个统一的流不需要关心底层是哪个供应商。这里有个容易踩的坑流式输出和工具调用同时使用时工具调用的参数是分多个chunk传过来的需要拼接。我见过有项目在流式模式下工具调用一直失败就是因为没有正确处理参数拼接。正确的做法是维护一个缓冲区把同一个工具调用的所有参数chunk拼起来等到finish_reason为tool_calls时再解析。提示流式模式下token使用量统计往往在最后一个chunk才返回如果你的计费逻辑依赖token数记得在流结束时再结算不要每个chunk都算一次。2.4 错误处理与重试的边界模型接入层的错误处理有个反直觉的地方不是所有错误都值得重试。限流错误429和超时错误值得重试但参数错误400和认证错误401重试多少次都没用。我的策略是把错误分成三类可重试错误、不可重试错误、需要降级的错误。可重试错误包括限流、超时、服务端5xx。这类错误用指数退避重试最多三次。不可重试错误包括参数格式错误、认证失败、模型不存在。这类错误直接抛给上层让业务逻辑决定怎么处理。需要降级的错误包括上下文超长、内容审核不通过。这类错误需要触发降级逻辑比如压缩上下文或者换一个模型重试。RETRYABLE_ERRORS {429, 500, 502, 503, 504} FATAL_ERRORS {400, 401, 403, 404} def call_with_retry(adapter, request, max_retries3): for attempt in range(max_retries): try: return adapter.call(request) except LLMError as e: if e.status_code in FATAL_ERRORS: raise if e.status_code in RETRYABLE_ERRORS and attempt max_retries - 1: time.sleep(2 ** attempt) continue raise这一层的设计质量直接决定了你的AI应用能不能在真实生产环境里稳定运行。我个人的经验是模型接入层的代码量不应该超过整个项目的20%如果超过了说明抽象没做好供应商差异泄漏到了上层。3. 能力编排层把散落的prompt变成可维护的工作流能力编排层是AI应用架构里最体现工程水平的一层。我见过两种极端一种是把所有prompt硬编码在业务逻辑里改一个标点符号都要重新部署另一种是过度设计搞了一套复杂的DSL结果团队里没人会用。我的观点是编排层的设计要遵循“够用就好”的原则先解决可维护性问题再考虑灵活性。3.1 Prompt模板的集中管理与版本控制Prompt是AI应用的“源代码”但很多人对待prompt的态度远不如对待代码认真。代码有版本控制、有code review、有测试prompt却经常是随手改改就上线。我的做法是把prompt当成代码来管理集中存放在一个目录里用模板引擎支持变量替换每次修改都走版本控制。# prompts/summarize.yaml name: summarize version: 3 template: | 你是一个专业的文本摘要助手。请对以下内容进行摘要 要求 1. 保留关键信息不超过{max_length}字 2. 使用{language}输出 3. 如果内容包含数字必须准确保留 内容 {content}用YAML或JSON管理prompt的好处是非技术人员也能参与修改而且可以方便地做A/B测试。你可以给每个prompt打上版本号在运行时根据配置选择版本。这样当新版本效果不好时可以快速回滚。这里有个实操心得prompt的变量替换一定要做转义处理。我踩过一次坑用户输入的内容里包含了模板语法字符导致渲染出来的prompt结构错乱。后来我在渲染前对所有变量值做了转义问题才解决。3.2 多步骤推理的编排模式单次模型调用能解决的问题很有限真实场景往往需要多步骤推理。比如一个客服场景先判断用户意图再根据意图检索知识库然后把检索结果和用户问题一起交给模型生成回答最后做一次质量检查。这四步如果写在一个prompt里效果往往不好因为模型很难同时做好这么多事。我的经验是把多步骤推理拆成独立的步骤每个步骤一个prompt步骤之间用代码串联。这样每个prompt的职责单一调试起来也容易。编排层要提供几种基本的编排模式顺序执行、条件分支、循环迭代、并行聚合。顺序执行最简单就是A步骤的输出作为B步骤的输入。条件分支是根据上一步的结果决定下一步走哪条路。循环迭代适合需要反复优化的场景比如让模型生成代码然后运行测试根据测试结果让模型修复直到测试通过或达到最大迭代次数。并行聚合是把一个任务拆成多个子任务同时执行最后把结果合并。class Workflow: def __init__(self): self.steps [] def add_step(self, name, prompt_name, conditionNone): self.steps.append({name: name, prompt: prompt_name, condition: condition}) return self def run(self, context): for step in self.steps: if step[condition] and not step[condition](context): continue prompt render_prompt(step[prompt], context) result llm_call(prompt) context[step[name]] result return context3.3 输出解析与结构化约束大模型的输出是自然语言但你的下游代码需要结构化数据。这个矛盾是AI应用开发中最常见的痛点。我的解决方案是三层保障prompt约束、解析容错、重试修复。Prompt约束是最基本的在prompt里明确要求输出JSON格式并给出schema示例。但光靠prompt约束不够可靠模型有时候会加一些解释性文字或者JSON格式有细微错误。所以需要解析容错用宽松的解析器能处理常见的格式问题比如多余的逗号、单引号、markdown代码块包裹等。import json import re def parse_json_output(text): # 去掉markdown代码块标记 text re.sub(r^(?:json)?\s*, , text.strip()) text re.sub(r\s*$, , text) # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个完整的JSON对象 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return None如果解析还是失败就需要重试修复。把解析错误信息和原始输出一起发给模型让它修复格式。这个重试通常一次就能成功因为模型看到具体错误后能理解问题所在。注意重试修复会增加延迟和成本所以要在prompt约束和解析容错上多下功夫把重试率控制在5%以下。3.4 编排层的可观测性设计编排层是AI应用里最需要可观测性的地方。一个请求经过多个步骤每个步骤的输入输出、耗时、token消耗都需要记录。没有这些数据你根本不知道问题出在哪一步。我的做法是在编排层埋点每个步骤执行时记录一条结构化日志包含步骤名称、输入prompt的hash、输出内容的hash、耗时、token数、是否成功。这些日志汇总后可以分析出很多有价值的信息哪个步骤最耗时、哪个prompt的token消耗最大、哪个步骤的失败率最高。import time import hashlib def execute_step(step_name, prompt, context): start time.time() try: result llm_call(prompt) log_step( stepstep_name, prompt_hashhashlib.md5(prompt.encode()).hexdigest(), output_hashhashlib.md5(result.encode()).hexdigest(), durationtime.time() - start, tokensresult.usage.total_tokens, statussuccess ) return result except Exception as e: log_step(stepstep_name, statuserror, errorstr(e)) raise这些数据积累一段时间后你就能做出有依据的优化决策。比如发现某个步骤平均耗时3秒占了整个请求的60%那优化这个步骤的收益最大。或者发现某个prompt的token消耗是其他prompt的5倍但输出质量没有明显提升那就可以考虑精简这个prompt。4. 记忆与状态层让AI应用记住该记住的没有记忆的AI应用就像金鱼每次对话都是全新的开始。但记忆不是简单地把历史消息全部塞进上下文那样做有两个问题一是token成本随对话轮数线性增长二是过长的上下文会稀释模型的注意力导致它忽略关键信息。记忆与状态层的设计目标是在有限的token预算内让模型获得最相关的历史信息。4.1 短期记忆的滑动窗口与摘要压缩短期记忆管理的是当前会话的上下文。最简单的策略是滑动窗口只保留最近N轮对话。N的取值取决于你的token预算和平均每轮对话的长度。如果每轮对话平均200token模型上下文窗口是8K那N可以取20左右留出空间给系统prompt和当前输入。但滑动窗口有个问题早期对话里的关键信息会被丢掉。比如用户在第三轮说了自己的名字到第二十轮时窗口已经滑过去了模型就忘了用户叫什么。解决这个问题有两种思路一种是关键信息提取在对话过程中实时提取实体和事实存到一个结构化的“用户档案”里每次请求都把这个档案带上另一种是滚动摘要当对话轮数超过窗口大小时把最早的一批对话压缩成摘要摘要和窗口内的原始对话一起送给模型。class ShortTermMemory: def __init__(self, max_turns20, max_tokens4000): self.turns [] self.summary self.max_turns max_turns self.max_tokens max_tokens def add(self, role, content): self.turns.append({role: role, content: content}) if len(self.turns) self.max_turns: self._compress() def _compress(self): # 把最早的一半对话压缩成摘要 to_compress self.turns[:len(self.turns) // 2] self.turns self.turns[len(self.turns) // 2:] summary_prompt f请将以下对话压缩成简洁的摘要保留关键信息\n{to_compress} new_summary llm_call(summary_prompt) self.summary f{self.summary}\n{new_summary} if self.summary else new_summary def get_context(self): context [] if self.summary: context.append({role: system, content: f之前的对话摘要{self.summary}}) context.extend(self.turns) return context4.2 长期记忆的向量化存储与检索长期记忆解决的是跨会话的信息保留问题。用户上周跟你聊过的话题这周再聊时你应该能想起来。实现方式是把对话内容向量化后存入向量数据库检索时用当前输入去查询最相关的历史片段。这里的关键是“什么该存”。不是所有对话都值得存入长期记忆。我的经验是只存三类内容用户明确表达的偏好和事实、重要的决策和结论、反复出现的话题。日常的寒暄和无关紧要的对话不需要存存了反而会干扰检索。向量化存储的另一个关键是分块策略。太长的文本块检索精度低太短的块又缺乏上下文。我的做法是按语义完整性分块每个块200到500字块之间保留一定的重叠。检索时返回top-k个最相关的块k一般取3到5。class LongTermMemory: def __init__(self, vector_store, embedder): self.vector_store vector_store self.embedder embedder def store(self, user_id, content, metadataNone): # 只存储有价值的内容 if not self._is_worth_storing(content): return embedding self.embedder.embed(content) self.vector_store.upsert( user_iduser_id, vectorembedding, contentcontent, metadatametadata or {} ) def retrieve(self, user_id, query, top_k5): query_vec self.embedder.embed(query) results self.vector_store.search(user_id, query_vec, top_k) return [r.content for r in results if r.score 0.7] def _is_worth_storing(self, content): # 过滤掉太短、纯寒暄、无信息量的内容 if len(content) 20: return False trivial [好的, 谢谢, 嗯, 收到] if content.strip() in trivial: return False return True4.3 状态管理的会话隔离与并发安全AI应用的状态管理有个容易被忽视的问题并发。当同一个用户同时发起多个请求时如果状态管理没有做好隔离会出现数据串扰。比如用户A的对话历史被写到了用户B的会话里这种bug在生产环境里是灾难性的。我的做法是每个会话有一个唯一的session_id所有状态操作都必须带上session_id。状态存储用带乐观锁的键值存储每次更新时检查版本号版本号不匹配就重试。这样即使有并发请求也不会互相覆盖。class SessionState: def __init__(self, store, session_id): self.store store self.session_id session_id def update(self, key, value): max_retries 3 for _ in range(max_retries): state, version self.store.get_with_version(self.session_id) state[key] value if self.store.compare_and_set(self.session_id, state, version): return raise ConcurrentModificationError()提示如果你的应用需要支持多设备同步状态存储要选择支持跨设备访问的方案并且要考虑冲突解决策略。简单的做法是“最后写入胜出”但更好的做法是让用户选择保留哪个版本。4.4 记忆的遗忘机制与隐私边界记忆不是越多越好。过时的、不相关的记忆会干扰模型的判断。所以需要设计遗忘机制。我的策略是给每条记忆设置一个“新鲜度”分数随着时间衰减。检索时优先返回新鲜度高的记忆低于阈值的记忆定期清理。隐私边界是另一个必须考虑的问题。用户的敏感信息比如密码、身份证号、银行卡号绝对不能存入长期记忆。我的做法是在存储前做一次敏感信息检测命中规则的内容直接丢弃并记录一条审计日志。同时给用户提供“清除我的记忆”的功能让用户对自己的数据有控制权。SENSITIVE_PATTERNS [ r\d{17}[\dXx], # 身份证 r\d{16,19}, # 银行卡 rpassword\s*[:]\s*\S, ] def contains_sensitive(text): for pattern in SENSITIVE_PATTERNS: if re.search(pattern, text): return True return False记忆与状态层的设计质量直接决定了用户觉得你的AI应用是“聪明”还是“健忘”。我个人的经验是这一层的投入产出比很高花时间把记忆管理做好用户体验的提升非常明显。5. 工具与协议层MCP如何统一AI与外部世界的接口工具调用是AI应用从“聊天机器人”进化到“智能助手”的关键一步。没有工具调用模型只能基于训练数据回答问题有了工具调用模型可以查实时数据、操作外部系统、执行具体任务。但工具调用的实现方式一直很混乱每个框架有自己的定义格式每个模型供应商有自己的调用协议。MCP协议的出现让这个领域有了标准化的希望。5.1 工具定义的标准结构与注册机制在MCP之前工具定义是各家各户自己定的。OpenAI有一套function calling的格式Anthropic有自己的tool use格式开源框架又有自己的定义。这导致一个工具要在多个模型上使用需要写多份定义。MCP的核心价值就是定义了一套标准的工具描述格式让工具的定义和调用解耦。一个标准的工具定义包含这几个部分工具名称、功能描述、参数schema、返回值说明。工具名称要简洁明确功能描述要写清楚“什么时候该用这个工具”因为模型是根据描述来决定是否调用的。参数schema用JSON Schema定义要写清楚每个参数的类型、是否必填、取值范围。{ name: query_weather, description: 查询指定城市的实时天气。当用户询问天气相关问题时使用此工具。, parameters: { type: object, properties: { city: { type: string, description: 城市名称如北京、上海 }, unit: { type: string, enum: [celsius, fahrenheit], default: celsius } }, required: [city] } }工具注册机制要解决的是“工具从哪来”的问题。我的做法是维护一个工具注册表每个工具是一个独立的模块启动时自动注册。注册表提供查询接口编排层根据当前任务需要把相关工具的定义传给模型。class ToolRegistry: def __init__(self): self.tools {} def register(self, name, description, parameters, handler): self.tools[name] { definition: { name: name, description: description, parameters: parameters }, handler: handler } def get_definitions(self, namesNone): if names: return [self.tools[n][definition] for n in names if n in self.tools] return [t[definition] for t in self.tools.values()] def execute(self, name, arguments): if name not in self.tools: raise ToolNotFoundError(name) return self.tools[name][handler](**arguments)5.2 MCP协议的核心概念与工作流程MCP的全称是Model Context Protocol它定义了一套客户端-服务端的交互协议让AI应用能够以统一的方式发现和调用外部工具。你可以把MCP服务端理解成一个“工具提供者”它声明自己有哪些工具、每个工具需要什么参数MCP客户端是“工具消费者”它连接服务端获取工具列表在需要时调用工具。MCP的工作流程分三步发现、调用、返回。发现阶段客户端连接服务端获取工具列表和每个工具的定义。调用阶段客户端把工具名称和参数发给服务端服务端执行工具并返回结果。返回阶段结果以标准格式返回给客户端客户端再把它交给模型。class MCPClient: def __init__(self, server_url): self.server_url server_url self.tools [] def discover(self): response requests.get(f{self.server_url}/tools) self.tools response.json()[tools] return self.tools def call(self, tool_name, arguments): response requests.post( f{self.server_url}/call, json{name: tool_name, arguments: arguments} ) return response.json()MCP相比传统的function calling有几个优势。第一是标准化工具的定义和调用格式统一不同模型和框架可以复用同一套工具。第二是解耦工具的提供者和消费者分离工具可以独立部署和升级。第三是可组合多个MCP服务端可以同时连接客户端可以聚合多个来源的工具。5.3 工具调用的安全边界与权限控制工具调用让模型能够操作外部系统这带来了安全风险。如果权限控制没做好模型可能被诱导执行危险操作。我见过一个案例客服机器人的工具里有“发送邮件”功能结果被用户诱导给全公司发了垃圾邮件。安全控制的第一道防线是工具白名单。不是所有工具都对所有用户开放。根据用户的角色和场景只暴露必要的工具。比如普通用户只能查天气、查订单管理员才能操作数据库。第二道防线是参数校验。模型生成的参数不可信必须做严格的校验。比如查询订单的工具订单号必须符合格式且必须属于当前用户。参数校验要在工具执行前做不通过直接拒绝。def execute_tool_safely(tool_name, arguments, user_context): # 检查工具是否在用户的允许列表中 if tool_name not in user_context.allowed_tools: raise PermissionDeniedError(f用户无权调用工具 {tool_name}) # 参数校验 tool registry.get(tool_name) validate_arguments(arguments, tool[definition][parameters]) # 注入用户上下文防止越权 if user_id in tool[definition][parameters][properties]: arguments[user_id] user_context.user_id return registry.execute(tool_name, arguments)第三道防线是操作审计。所有工具调用都要记录日志包括谁调的、调了什么、参数是什么、结果是什么。这样出了问题可以追溯也能发现异常调用模式。注意对于有副作用的工具比如写操作、发送操作建议加一道人工确认。模型生成调用请求后先展示给用户确认用户点了确认才真正执行。这样能避免模型误操作。5.4 工具编排与多工具协作真实场景往往需要多个工具协作完成一个任务。比如用户说“帮我订明天去上海的机票然后告诉张三”这需要调用航班查询、机票预订、联系人查询、消息发送四个工具。工具编排层要负责决定调用顺序、传递中间结果、处理失败情况。我的做法是把工具编排也做成工作流。每个步骤可以是一个工具调用也可以是模型推理。步骤之间通过上下文传递数据。编排层要处理几种情况串行调用A的结果是B的输入、并行调用A和B同时执行、条件调用根据A的结果决定是否调B。class ToolOrchestrator: def __init__(self, llm, registry): self.llm llm self.registry registry def run(self, user_input, max_steps10): context {user_input: user_input, history: []} for _ in range(max_steps): # 让模型决定下一步 decision self.llm.decide_next_action( context, self.registry.get_definitions() ) if decision.action final_answer: return decision.content elif decision.action tool_call: result self.registry.execute( decision.tool_name, decision.arguments ) context[history].append({ tool: decision.tool_name, result: result }) return 任务步骤过多已终止工具编排的难点在于错误处理。工具调用可能失败失败后是重试、换工具、还是告诉用户我的策略是可重试的错误自动重试一次不可重试的错误把错误信息返回给模型让模型决定下一步。模型看到错误信息后可能会换一个工具或者告诉用户当前无法完成。6. 交互与呈现层流式输出、多模态与用户体验细节交互与呈现层是用户唯一直接感知的层。同样的模型能力交互设计的好坏会让用户体验天差地别。我见过两个功能完全相同的AI应用一个用户留存率是另一个的三倍差别就在交互细节上。6.1 流式输出的前端消费与渲染优化流式输出是AI应用体验的基石。用户不需要等整个回答生成完而是看到文字一个字一个字地出现这种即时反馈大大降低了等待焦虑。但流式输出的前端实现有不少坑。第一个坑是SSE连接的管理。SSE是单向的服务端推、客户端收。但用户可能在中途取消生成或者网络中断。前端要能正确处理这些情况取消时关闭连接中断时自动重连并带上最后收到的位置。第二个坑是渲染性能。如果每个字符都触发一次DOM更新页面会卡顿。正确的做法是用缓冲区积累一批字符再统一更新。通常每50毫秒更新一次或者每积累10个字符更新一次。class StreamRenderer { constructor(element) { this.element element; this.buffer ; this.timer null; } append(text) { this.buffer text; if (!this.timer) { this.timer setTimeout(() this.flush(), 50); } } flush() { this.element.textContent this.buffer; this.buffer ; this.timer null; this.scrollToBottom(); } scrollToBottom() { this.element.scrollTop this.element.scrollHeight; } }第三个坑是markdown渲染。模型输出的是markdown格式但流式过程中markdown是不完整的比如代码块只收到了一半。如果直接渲染会出现格式错乱。我的做法是流式过程中用纯文本渲染等流结束后再做一次完整的markdown渲染。6.2 多模态输入输出的架构设计多模态是AI应用的发展趋势。用户不再满足于打字他们想上传图片、语音、文件。架构上要支持这些输入类型并把它们统一转换成模型能理解的格式。图片输入的处理流程是前端上传图片到对象存储拿到URL后端把URL转换成模型能接受的格式通常是base64或URL模型返回图片描述或基于图片的回答。这里要注意图片大小限制太大的图片要先压缩。语音输入需要先做语音识别把音频转成文本再走文本对话流程。语音识别可以用专门的ASR服务也可以用多模态模型直接处理音频。我的经验是对于实时性要求高的场景用专门的ASR服务延迟更低对于需要理解语气和情感的場景用多模态模型效果更好。文件输入要区分类型。PDF和Word需要先提取文本图片需要OCR表格需要解析成结构化数据。这些预处理步骤可以在上传时异步完成等用户真正提问时直接使用处理好的文本。class MultimodalInput: def process(self, input_data): if input_data.type image: return self._process_image(input_data) elif input_data.type audio: return self._process_audio(input_data) elif input_data.type file: return self._process_file(input_data) else: return input_data.content def _process_image(self, data): # 压缩图片 compressed compress_image(data.content, max_size1024) # 转成base64 return {type: image_url, image_url: {url: to_base64(compressed)}} def _process_file(self, data): if data.mime_type application/pdf: text extract_pdf_text(data.content) elif data.mime_type.startswith(image/): text ocr(data.content) else: text data.content.decode(utf-8, errorsignore) return {type: text, text: text}6.3 思考过程展示与引用来源标注用户对AI的信任建立在透明之上。如果模型直接给一个答案用户不知道这个答案是怎么来的信任度会打折扣。展示思考过程和引用来源能显著提升用户信任。思考过程展示有两种方式一种是展示模型的推理链让用户看到模型是怎么一步步得出结论的另一种是展示工具调用过程让用户知道模型查了哪些数据。我的做法是在界面上加一个可折叠的“思考过程”区域默认折叠用户想看的时候可以展开。引用来源标注是另一个提升信任的手段。当模型基于检索到的文档回答时在回答中标注引用编号用户点击编号可以看到原文。实现方式是在prompt里要求模型用[1]、[2]这样的格式标注引用前端解析这些标记并渲染成可点击的链接。def format_with_citations(answer, sources): # 把 [1] 这样的标记替换成可点击的链接 for i, source in enumerate(sources, 1): answer answer.replace( f[{i}], fa href{source.url} target_blank[{i}]/a ) return answer6.4 错误状态与降级体验的设计AI应用不可能永远不出错。模型超时、限流、返回格式错误这些都会发生。关键是出错时用户体验不能崩。我的设计原则是永远给用户一个可操作的下一步。如果模型超时不要只显示“请求失败”而是显示“正在重试”并自动重试。如果重试也失败给用户两个选项重新发送或者简化问题。如果模型返回的内容被安全过滤了不要显示“内容违规”而是显示“这个问题我暂时无法回答你可以换个方式问问”。function handleError(error) { if (error.type timeout) { showMessage(响应时间较长正在重试...); retry(); } else if (error.type rate_limit) { showMessage(当前请求较多请稍后再试); showRetryButton(); } else if (error.type content_filter) { showMessage(这个问题我暂时无法回答你可以换个方式问问); } else { showMessage(出了点小问题请重试); showRetryButton(); } }降级体验还包括功能降级。当高级模型不可用时自动切换到基础模型并在界面上提示“当前使用基础模式”。当工具调用失败时告诉用户“暂时无法查询实时数据以下是基于已有知识的回答”。交互与呈现层的设计没有标准答案核心是站在用户角度思考用户在这个时刻需要什么信息、需要什么操作。把这两个问题回答好体验就不会差。7. 从架构图到生产部署、监控与迭代的实战经验架构设计得再好最终要落到生产环境里跑起来。这一章我想分享一些从实际项目中总结的经验都是踩过坑之后才明白的道理。7.1 部署架构的选型与成本控制AI应用的部署架构和传统Web应用有个显著区别GPU资源。如果你用的是自部署模型GPU成本是大头如果你用的是API那成本主要在token消耗上。两种情况的优化策略完全不同。用API的情况下成本控制的核心是减少不必要的token消耗。我做过一个统计一个AI客服应用里40%的token消耗在系统prompt上。系统prompt往往很长包含角色设定、行为规范、工具定义等。优化方法是把系统prompt里不变的部分缓存起来很多供应商支持prompt caching能省不少钱。自部署模型的情况下GPU利用率是关键。我的经验是把推理服务做成无状态的前面加一个队列请求来了先入队推理服务从队列里取任务。这样能平滑流量波动避免GPU忽忙忽闲。队列的长度要监控太长说明GPU不够太短说明资源浪费。# docker-compose.yml 示例 services: api: image: myapp-api environment: - LLM_PROVIDERopenai - REDIS_URLredis://redis:6379 depends_on: - redis worker: image: myapp-worker environment: - QUEUE_URLredis://redis:6379 deploy: replicas: 3 redis: image: redis:7-alpine7.2 监控指标除了QPS和延迟还要看什么传统应用的监控指标是QPS、延迟、错误率。AI应用还需要关注一些特有指标。第一个是token消耗速率按小时统计突然飙升往往意味着有异常调用。第二个是上下文长度分布如果平均上下文长度持续增长说明记忆管理可能有问题。第三个是工具调用成功率这个指标下降说明外部依赖出了问题。第四个指标是“有效回答率”。什么是有效回答用户没有在回答后立刻重新提问没有点“踩”没有关闭会话。这个指标比单纯的“请求成功率”更能反映真实质量。我通常用“用户追问率”作为代理指标追问率高说明上一次回答没解决问题。class MetricsCollector: def record_request(self, request, response, duration): self.histogram(request_duration, duration) self.counter(token_usage, response.usage.total_tokens) self.counter(context_length, len(request.messages)) if response.tool_calls: self.counter(tool_calls, len(response.tool_calls)) def record_feedback(self, session_id, feedback_type): self.counter(ffeedback_{feedback_type}) # 计算追问率 if feedback_type follow_up: self.counter(follow_up_rate)7.3 灰度发布与Prompt迭代的安全流程Prompt的修改比代码修改更危险因为prompt的效果很难用单元测试验证。我的做法是prompt修改必须走灰度发布。新prompt先对5%的流量生效观察核心指标有效回答率、追问率、token消耗有没有恶化。如果指标正常逐步扩大到20%、50%、100%。如果指标恶化立即回滚。灰度发布的关键是流量分割要一致。同一个用户应该始终看到同一个版本的prompt否则体验会不一致。我的做法是用user_id做hashhash值落在灰度区间内的用户用新prompt其他的用旧prompt。def select_prompt_version(user_id, prompt_name): versions get_prompt_versions(prompt_name) if len(versions) 1: return versions[0] # 灰度10%流量用新版本 hash_val int(hashlib.md5(user_id.encode()).hexdigest(), 16) if hash_val % 100 10: return versions[-1] # 新版本 return versions[0] # 稳定版本7.4 常见线上问题的排查链路线上问题排查最怕没有线索。我的经验是提前埋好日志每个请求记录完整的调用链用户输入、系统prompt、模型输出、工具调用、最终响应。这样出问题时能快速定位。一个常见问题是“模型回答质量突然下降”。排查链路是先看是不是prompt被改了对比当前prompt和昨天的版本再看是不是模型供应商更新了模型很多供应商会静默更新模型版本然后看是不是检索到的上下文质量下降了检查向量检索的召回率和相关性分数最后看是不是用户输入分布变了统计最近一周的用户输入类型。另一个常见问题是“响应变慢”。排查链路是先看模型API的延迟如果模型本身慢了那是供应商的问题再看工具调用的延迟某个外部服务变慢会拖累整个请求然后看队列等待时间如果队列积压说明需要扩容最后看数据库查询记忆检索的查询可能随着数据量增长而变慢。def diagnose_slow_request(request_id): trace get_trace(request_id) timings { queue_wait: trace.queue_wait, model_call: trace.model_call_duration, tool_calls: sum(t.duration for t in trace.tool_calls), memory_retrieval: trace.memory_retrieval_duration, total: trace.total_duration } # 找出耗时最长的环节 bottleneck max(timings, keytimings.get) return f瓶颈在 {bottleneck}耗时 {timings[bottleneck]:.2f}s7.5 架构演进从单体到分层的实际路径最后聊聊架构演进。没有人一开始就设计出完美的五层架构。我的实际路径是第一版把所有逻辑写在一个文件里能跑就行第二版把prompt抽出来模型调用封装成函数第三版加入记忆管理把会话状态独立出来第四版加入工具调用把工具注册和执行分离第五版引入MCP把工具层标准化。每一步演进都是被需求驱动的不是为了架构而架构。我的建议是不要一开始就追求完美的分层先让功能跑起来然后在遇到痛点时重构。重构的时机是当你发现“改一个地方要动很多文件”或者“加一个新功能要复制粘贴很多代码”的时候。架构演进的另一个经验是保持每一层的可替换性。模型接入层可以换供应商能力编排层可以换框架记忆层可以换存储工具层可以换协议。只要层与层之间的接口稳定内部实现怎么换都行。这种可替换性让你在技术选型变化时不会推倒重来。我在实际项目里最深的体会是AI应用架构设计的核心不是技术而是对“不确定性”的管理。模型输出不确定、工具返回不确定、用户输入不确定。好的架构把这些不确定性隔离在特定层里让系统的其他部分保持确定和可控。这才是架构设计的价值所在。