企业官网建设流程全解析

做后台系统做了快一年，产品经理突然说要在现有应用里加一个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_factory=dict) @dataclass class ChatResponse: message: ChatMessage # role="assistant" 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]: """流式：按增量返回文本片段""" pass

stream_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": f"Bearer {self.api_key}"} async with httpx.AsyncClient(timeout=60) as client: async with client.stream( "POST", f"{self.base_url}/chat/completions", json=payload, headers=headers ) 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_tokens=8000, reserve_tokens=1024): budget = max_context_tokens - reserve_tokens # 给模型回复留出空间 kept = [ChatMessage(role="system", content=system_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

实现流式输出的传输层，我倾向用HTTP+SSE而不是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发送、Shift+Enter换行、防重复提交

聊天输入框中，用户约定俗成的是Enter发送消息、Shift+Enter换行。这虽然是小细节，但没做好的话，用户按个Enter就把消息发出去了，非常难受。同时要在发送状态加一个锁，防止用户狂点发送键导致重复请求。更好的做法是：用户点击发送后立即禁用输入框和按钮，等流式生成结束或者用户主动点击停止后再恢复。

5.4 错误状态的人话化

AI聊天服务一定会遇到各种错误：网络超时、限流、服务商维护、上下文超长。这些错误信息如果原样抛给用户，非常劝退。我在UI层把错误信息映射成几个友好提示：

同时UI上要有“重试”按钮，用户点一下就能重新发起上次请求。实现重试时要注意幂等性，如果上次请求已经部分生成流式文本，直接重试会产生重复内容，更稳妥的做法是请求前带上一个“重新生成”意图，后端据此清空上一轮未完成内容。

6. 排坑记：流式中断、并发错乱、上下文爆炸

6.1 中断生成：前端的abort怎么与后端协调

一个很常见但让人头疼的问题是：用户点击“停止生成”后，前端只是停止了渲染，但后端的API请求还在继续。原因很简单，用户可能等不到完整回答就点了停止，或者回答质量不好要换一个。前端用AbortController中断fetch之后，怎么通知后端呢？

我的方案是：前端abort时发送一个cancel请求到后端，后端用一个异步事件通知正在进行的生成任务，主动退出流循环。具体做法是在会话对象里注册一个asyncio.Event，cancel请求触发该事件，流式生成代码监听事件并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的时候调试所有交互。这个技巧比你想的更有用。