做后台系统做了快一年产品经理突然说要在现有应用里加一个AI聊天入口。说实话最初我下意识的想法是直接找模型服务商拿一套SDK在业务代码里写个调用方法就完事。结果真接进去才发现问题一堆换模型要改代码上下文管理要写在业务逻辑里请求重试、流式输出、费用控制全搅在一起。后来我把这个功能拆成一个独立的“AI聊天功能插件”也就是标题里说的进阶篇第一讲——这个插件架构才真正把问题理顺了。这篇内容适合谁看适合那些已经在自己的应用里做过基础AI功能、但想进一步把聊天能力做成可维护、可扩展模块的开发者。这篇不教你从零写大模型而是讲怎么把AI能力封装成插件并处理好接口抽象、上下文管理、流式交互这些真正麻烦的部分。全程是实操经验和踩坑记录。1. 为什么要用插件架构做AI聊天——硬编码调用逻辑的教训1.1 在业务代码里直接调API前期很爽后期很痛我见过很多项目的做法是控制层里拿到用户输入直接调用模型API把返回结果塞进页面。代码看起来只有十几行联调也快但上线几周后就开始难受。第一个难受的点是模型服务商升级模型版本或者调整了请求字段你得在业务代码里找出所有调用点逐个改。第二个点是多轮对话前端传来的历史消息要手动拼成一个大字符串拼错了或者超出上下文窗口对话质量直接崩塌。第三个点是排查问题困难你根本分不清是网络超时、接口限流还是模型返回了异常结构。我自己第一次做的版本就是这样核心业务代码里硬编码了一个请求函数参数写死超时写死连失败重试都是复制粘贴的。结果某天模型服务商把某个报错信息从英文换成了中文我这边解析逻辑就挂了。用“插件”的思路重构之后我才意识到AI聊天本质是一个独立的子系统它和业务模块之间应该隔着一层明确的接口。1.2 插件化拆分的核心思路四个部分各管各的我最终把整个AI聊天功能拆成了四块这也是这类插件最常见的骨架接口层定义“聊天请求”和“聊天响应”的标准结构业务代码只和这层打交道。服务商适配层每种模型服务商写一个适配器负责把标准请求翻译成服务商API的参数再把返回结果翻译回标准响应。会话管理层负责历史消息存储、上下文裁剪、多会话隔离。UI交互层输入框、消息列表、流式渲染、错误展示。这样拆完以后业务代码只需要知道“我要发一条用户消息然后拿回一段AI回复”完全不需要关心背后是哪个模型、上下文怎么拼、是不是重试过。换服务商时只需要新增一个适配器类改一行配置。维护成本下降得不是一点半点。1.3 配置驱动一行配置切换底层大脑插件架构下切换模型服务商应该是一个配置行为而不是代码行为。我习惯用一个config.yaml或application.properties来管理chat: provider: default # 可选 default / alternate model: chat-model-v2 temperature: 0.7 max_tokens: 2048 timeout_seconds: 60 max_context_tokens: 8000 api_key_env: LLM_API_KEY这样运营环境、测试环境可以用不同的provider代码完全不用动。有一次模型服务商临时维护我直接在配置里切到备用服务商两分钟完成切换不用重新发版。2. 接口层设计先定义“聊天”的边界再谈对接2.1 用“消息数组”作为统一输入格式我发现很多项目最大的误区是把请求体设计成“一个用户输入的字符串”。普通问答勉强能用一旦进入多轮对话、带系统提示词、带工具调用字符串结构根本承载不了。正确做法是设计一个标准消息结构也就是几乎所有大模型API通用的那种一组消息对象每条消息有role和content必要时带name或tool_call。# interface.py from dataclasses import dataclass, field from typing import Optional dataclass class ChatMessage: role: str # system / user / assistant content: str name: Optional[str] None dataclass class ChatRequest: messages: list[ChatMessage] temperature: float 0.7 max_tokens: int 2048 stream: bool False extra_params: dict field(default_factorydict) dataclass class ChatResponse: message: ChatMessage # roleassistant finish_reason: str stop usage: Optional[dict] None为什么选“消息数组”而不是“拼接字符串”因为模型服务商本质上就是在预测“一段对话中助手接下来应该回复什么”。消息数组天然表达了对话的历史进程也方便交给上下文化裁剪逻辑去过滤。我在代码审查时看到有人在前端把历史记录拼成用户说xxx助手说yyy再传回来且不说token浪费模型理解上下文的能力也会大打折扣。2.2 同步响应和流式响应必须分开接口层最容易忽略的是流式。普通问答用同步响应没问题但真正的聊天体验尤其是长回答必须用流式输出。一分钟的等待时间如果只给一个“加载中”的菊花转圈用户早就把页面关了。因此接口层至少需要两个方法class BaseChatProvider(ABC): async def chat(self, req: ChatRequest) - ChatResponse: 非流式一次性返回完整回复 pass async def stream_chat(self, req: ChatRequest) - AsyncIterator[str]: 流式按增量返回文本片段 passstream_chat返回的是一个异步迭代器每产出一段文本就yield一次。具体怎么对接模型服务商的SSE数据流是适配层的事情接口层不关心。我建议接口层的返回值永远是标准消息对象或文本增量流不要塞服务商自定义的字段。2.3 为什么接口层要独立成包我在实际项目里把接口层定义放在一个独立的Python包里比如叫chat_core它只有接口类和dataclass没有任何网络请求逻辑。这样做的原因是业务团队、前端团队、服务商适配层开发团队可以并行工作。前端能依据接口文档先做Mock页面业务代码能先调用一个FakeProvider把流程跑通后端再去填真实的适配器。接口定义稳定了整个插件的骨架就稳了。3. 服务商适配层换模型不换业务代码的关键3.1 适配器模式实践服务商适配层解决的核心痛点是各家API的请求路径、字段名、鉴权方式、错误返回结构都不一样。有的兼容OpenAI风格有的走自家SDK有的需要用动态签名。适配层的做法是为每个服务商写一个子类统一实现chat和stream_chat。# providers/default_provider.py import httpx, os class DefaultProvider(BaseChatProvider): def __init__(self, model: str, temperature: float, max_tokens: int): self.model model self.temperature temperature self.max_tokens max_tokens self.api_key os.environ[LLM_API_KEY] self.base_url https://your-llm-provider.example/v1 async def stream_chat(self, req: ChatRequest): payload { model: self.model, messages: [ {role: m.role, content: m.content} for m in req.messages ], temperature: req.temperature, max_tokens: req.max_tokens, stream: True, } headers {Authorization: fBearer {self.api_key}} async with httpx.AsyncClient(timeout60) as client: async with client.stream( POST, f{self.base_url}/chat/completions, jsonpayload, headersheaders ) as resp: async for line in resp.aiter_lines(): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break delta self._parse_delta(data) if delta: yield delta这段代码里有几个细节值得说。第一我用httpx.AsyncClient的stream上下文管理器而不是httpx.post这样连接不会一直占着不放。第二SSE返回的每一行以data:开头要正确切掉前缀同时注意[DONE]结束标记。第三_parse_delta里要处理choices[0].delta.content为空的情况服务商经常在首位返回一个空的role字段。3.2 参数映射不要把服务商参数直接漏给业务层接口层定义了temperature、max_tokens这些通用参数但不同模型对这些参数的解释有细微差异。有的模型用max_tokens有的用max_output_tokens有的不支持temperature还有的需要额外传top_p。适配层应该负责把这套“标准参数”翻译成服务商“方言”不支持的参数乖乖丢进extra_params或者丢弃。别让这些差异性污染业务层。还要注意重试逻辑。模型API经常因为网络抖动、限流、负载高返回5xx错误。适配层里我写了指数退避重试最多重试3次退避间隔1秒起翻倍。重试只应对幂等请求即请求本身未被真正处理或可安全重放。流式请求如果中途断流前端会看到消息流卡住这时我会在适配层捕获连接异常返回一个特殊的“中断标记”而不是硬撑。3.3 验证适配器的一小套测试方法换一个服务商时我建议先跑三组固定测试不要直接在应用里联调第一组发一条“请回复OK”的简单消息验证响应解析是否正确。第二组发带有多轮历史的对话验证消息顺序是否正确传入。第三组开启流式确认每个yield文本片段不丢失、顺序不乱。这套测试几分钟就能跑完却能把90%的适配问题挡在上线前。我踩过一次大坑某服务商在流式响应里每行data包含的是完整JSON而非增量适配层没改解析逻辑结果前端每秒钟刷新一整段重复文本。4. 上下文管理把“多轮聊天”做对核心是窗口裁剪4.1 上下文窗口和Token的现实约束大模型的上下文窗口是有限资源。哪怕宣称支持128k上下文你真的把所有历史消息都塞进去效果也会因为“信息过载”而变差而且成本肉眼可见地涨。上下文管理的本质是在有限的窗口里用尽量少的Token保留尽量多的有效信息。整个上下文由三部分组成系统提示词设定角色和行为、历史消息用户的提问和助手的回答、当前输入。你给模型设定的系统提示词可能就有几百Token历史消息越多占用越高。如果不做任何管理多轮对话后必定突破窗口上限直接报错。4.2 从旧消息开始丢系统提示词永远不丢我的裁剪策略很简单固定保留最近的N条消息超出预算的部分从头开始丢弃但永远保留系统提示词。实现上可以对每条消息估算Token数按从旧到新的顺序累加直到接近预算def trim_messages(messages, system_prompt, max_context_tokens8000, reserve_tokens1024): budget max_context_tokens - reserve_tokens # 给模型回复留出空间 kept [ChatMessage(rolesystem, contentsystem_prompt)] used estimate_tokens(system_prompt) for msg in reversed(messages): if msg.role system: continue # 避免重复追加系统提示 estimated estimate_tokens(msg.content) if used estimated budget: break # 放不下就停止 kept.insert(1, msg) # 插到系统提示之后 used estimated return kept注意这里我留了reserve_tokens大约1024Token给模型生成回答的空间。如果不预留上下文塞得太满模型可能答到一半就中断因为输出Token预算为0了。4.3 精确Token计数和估算的取舍真正精确的Token计数需要调用模型的tokenizer但对绝大多数应用来说按“字符数/3”或“字符数/4”估算足够了。我常用一个近似函数英文按4字符1Token中文按1字1.5Token。这个估算误差通常在10%以内对裁剪决策没有实质影响。如果用第三方SDK很多服务商会在请求返回里带usage.prompt_tokens这是精确计数。我建议把精确计数日志记录到会话存储里用来校准估算函数。初期校准一次后面就不用再关心这个问题了。4.4 会话持久化与多会话隔离插件的会话管理不仅要管“当前聊天”还要管“历史聊天记录”。我设计了Session结构每个会话有一个唯一ID包含会话元数据创建时间、标题、模型配置和消息数组。dataclass class ChatSession: session_id: str messages: list[ChatMessage] created_at: float last_active_at: float存储我用SQLite字段就是上面这几个。为什么不直接用文件因为多线程并发时文件读写要加锁SQLite天然处理并发读和串行写省心得多。多用户在线的系统要增加一个user_id索引保证每个用户只能访问自己的会话。这是最基本的数据隔离要求。上下文裁剪一定要读取完整的会话消息数组而不是只传最近几轮。我见过一个简化实现只传了最近3轮结果用户之前问过的“你帮我记住我的生日是5月20日”在后续对话中彻底失效。裁剪做的是“压缩粒度”不是“截断依赖”。5. UI交互层流式打字机效果和异常状态的人话化5.1 用SSE流还是WebSocket实现流式输出的传输层我倾向用HTTPSSE而不是WebSocket。虽然WebSocket也能推流但SSE是单向服务器推送语义更清晰配合HTTP状态码也更容易调试还能天然穿透大多数代理。后端接口如果有认证用Bearer Token放在Authorization头里就行。前端拿到SSE流之后要做的是逐段解析文本增量并追加到当前AI消息的DOM节点上实现“打字机”效果。核心逻辑是读流、解析、追加、交给渲染层async function streamChat(sessionId, userContent) { const resp await fetch(/api/chat/${sessionId}/stream, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify({content: userContent}) }); const reader resp.body.getReader(); const decoder new TextDecoder(); let aiMessageNode createMessageNode(assistant); while (true) { const {done, value} await reader.read(); if (done) break; const text decoder.decode(value, {stream: true}); appendTextToNode(aiMessageNode, text); autoScroll(aiMessageNode); } }这段代码虽然简单但涉及几个关键点。stream: true的TextDecoder是为了防止一个字符被拆到两个chunk里造成乱码这在中文环境下特别重要。autoScroll要在内容高度超过视口时自动滚动到最新消息但用户手动向上滚动时不打扰。实现这个交互要多花点心思但体验差距非常明显。5.2 渲染Markdown与代码高亮以及防注入模型回复通常包含Markdown格式尤其是代码块。如果直接把文本插入DOM代码块会变成一团没有格式的纯文本可读性差。前端用Markdown解析库先转成HTML再渲染同时对代码块做高亮。注意模型回复属于“不可信数据”渲染时要做HTML转义防止注入。我的做法是先用转义库把原文中的HTML标签转义成实体再用Markdown解析器处理。顺序不能反过来否则攻击脚本就趁虚而入了。5.3 输入框交互细节Enter发送、ShiftEnter换行、防重复提交聊天输入框中用户约定俗成的是Enter发送消息、ShiftEnter换行。这虽然是小细节但没做好的话用户按个Enter就把消息发出去了非常难受。同时要在发送状态加一个锁防止用户狂点发送键导致重复请求。更好的做法是用户点击发送后立即禁用输入框和按钮等流式生成结束或者用户主动点击停止后再恢复。5.4 错误状态的人话化AI聊天服务一定会遇到各种错误网络超时、限流、服务商维护、上下文超长。这些错误信息如果原样抛给用户非常劝退。我在UI层把错误信息映射成几个友好提示错误类型提示文案网络超时“网络不太顺畅请检查连接后重试”限流“当前问的人有点多请稍等片刻再试”上下文超长“这段内容有点长可以拆分后再继续聊”服务商维护“服务暂时不可用我会稍后恢复”同时UI上要有“重试”按钮用户点一下就能重新发起上次请求。实现重试时要注意幂等性如果上次请求已经部分生成流式文本直接重试会产生重复内容更稳妥的做法是请求前带上一个“重新生成”意图后端据此清空上一轮未完成内容。6. 排坑记流式中断、并发错乱、上下文爆炸6.1 中断生成前端的abort怎么与后端协调一个很常见但让人头疼的问题是用户点击“停止生成”后前端只是停止了渲染但后端的API请求还在继续。原因很简单用户可能等不到完整回答就点了停止或者回答质量不好要换一个。前端用AbortController中断fetch之后怎么通知后端呢我的方案是前端abort时发送一个cancel请求到后端后端用一个异步事件通知正在进行的生成任务主动退出流循环。具体做法是在会话对象里注册一个asyncio.Eventcancel请求触发该事件流式生成代码监听事件并break。测试中发现如果只依赖HTTP连接断开来感知取消服务端的请求可能会多跑好几秒浪费Token还会生成一条用户根本看不到的回复。6.2 用户快速连发消息导致上下文错乱这个坑非常隐蔽。前端虽然做了输入框禁用但高并发下仍然可能出现用户先发送了消息A还没生成完又通过快捷键发了消息B。后端如果不对会话做串行处理两个请求同时读取消息数组、同时调用模型API返回的结果可能交叉最后会话历史变得一团糟。解决方法是给每个会话加一个轻量锁。同一时间只允许一个正在生成的请求处理该会话其他请求要么排队要么直接返回“请等待上一条完成”。我采用的是异步队列把新请求挂到会话的pending队列当前生成结束后自动处理下一个。这比直接拒绝体验好用户连续追问时不必手动重发。6.3 裁剪策略误删关键指令有次用户反馈AI回答突然“失忆”了很多早期说过的要求都不再遵守。排查后发现问题出在裁剪逻辑上我的裁剪器把系统提示之后的消息按时间倒序保留但中间有一些用户设置的偏好信息比如“以后回答都用表格形式”被当成旧消息丢掉了。修正方案是把“用户偏好”“系统指令”这类消息打上sticky标记裁剪时这部分消息永远保留只在所有非sticky消息都保不住时才考虑丢弃。实现上很简单在ChatMessage里加一个sticky: bool False字段。这个教训告诉我上下文裁剪不是单纯的按时间截断还要考虑消息的类型。6.4 成本控制比想象中重要AI聊天插件上线后模型API费用成了不可忽视的成本项。我在插件里加了三层费用控制会话级每会话每天最多调用N次。模型级超出一定Token消耗后自动切换到更便宜的轻量模型。全局级每日总消耗达到阈值后直接停止调用返回提示。具体的限额在配置里写清楚运营人员可以随时调整。有一次我忘了设置限额测试中狂调用模型生成长文一天就消耗了数千Token额度虽然金额不大但这种失控很让人不安。代码里的限流我用的简单令牌桶import time class RateLimiter: def __init__(self, rate, capacity): self.rate rate self.capacity capacity self.tokens capacity self.last time.monotonic() def acquire(self): now time.monotonic() self.tokens min( self.capacity, self.tokens (now - self.last) * self.rate ) self.last now if self.tokens 1: return False self.tokens - 1 return True这个限流器不依赖外部Redis够单机用。多实例部署时需要使用分布式限流组件但思路是一样的每秒允许通过多少个请求、突发量多大、超限时是丢弃还是排队。7. 最后说说我重构后的体会做AI聊天功能插件最核心的不是模型本身而是边界。把“聊天”抽象成接口层之后换模型变成改配置把“历史”交给会话管理层之后上下文不再乱把“流式交互”交给UI层之后用户体验才真正像对话。如果你现在正打算在项目里加AI功能我建议不要急着粘贴一段API调用代码先花半天把接口层定义清楚。定义清楚了后面所有开发都会非常顺畅。最后再分享一个小技巧在插件目录里放一个mock_provider.py实现一个返回固定数据的假服务商。开发时直接用这个假provider跑通全流程不消耗任何真实Token还能在测试环境稳定复现各种边界场景。很多同事不理解为什么开发环境也要跑AI有了这个mock你就能在没有网络、没有API Key的时候调试所有交互。这个技巧比你想的更有用。