Langchain-Chatchat问答系统灰度期间服务降级预案

Langchain-Chatchat问答系统灰度期间服务降级预案

在企业知识管理日益智能化的今天，员工不再满足于翻阅冗长的PDF文档来查找一条报销政策。他们希望像问同事一样，直接提问就能得到准确、自然的回答。这种需求催生了基于大语言模型（LLM）与本地知识库融合的智能问答系统——Langchain-Chatchat 正是其中的典型代表。

然而，理想很丰满，现实却常有波折。当系统从测试环境走向真实用户场景，在灰度发布阶段，我们很快会遇到这样的问题：突然涌入的访问请求让GPU显存告急，LLM响应延迟飙升至十几秒；向量数据库因索引膨胀导致检索变慢；甚至某些敏感信息被非授权人员误触……这些问题若不加应对，轻则体验下降，重则服务雪崩。

如何在资源受限或异常情况下，依然保障核心服务能力？这就需要一套可执行、可切换、可回退的服务降级机制。它不是“出事了再说”的应急预案，而是系统设计之初就应内建的韧性能力。

Langchain-Chatchat 的核心技术架构由三大部分构成：LangChain 框架作为流程中枢，大型语言模型负责语义生成，向量数据库实现知识检索。这三者环环相扣，任何一个环节出现瓶颈，都可能引发连锁反应。因此，我们的降级策略必须覆盖全链路，而不是孤立地看待某一个组件。

先看LangChain。很多人把它当作简单的“胶水框架”，但实际上，它的模块化设计为服务弹性提供了极大空间。比如RetrievalQA链中的chain_type参数，常见的有stuff、map_reduce、refine等模式。在高负载时，我们可以主动降级到更轻量的链类型——虽然map_reduce会增加一点处理时间，但它能分批处理长文本，避免因上下文过长导致显存溢出。

更重要的是，LangChain 支持运行时动态替换组件。这意味着我们可以在检测到主模型不可用时，立即切换至备用链路。例如：

from langchain.chains import RetrievalQA from langchain.llms import HuggingFacePipeline from transformers import pipeline import torch # 主模型（高性能但耗资源） main_llm = HuggingFaceHub(repo_id="qwen/Qwen-7B", model_kwargs={"temperature": 0.7}) # 备用轻量模型（低显存占用） small_tokenizer = AutoTokenizer.from_pretrained("google/flan-t5-small") small_model = AutoModelForSeq2SeqLM.from_pretrained("google/flan-t5-small") pipe = pipeline( "text2text-generation", model=small_model, tokenizer=small_tokenizer, max_new_tokens=128, device=0 if torch.cuda.is_available() else -1 ) fallback_llm = HuggingFacePipeline(pipeline=pipe) # 动态切换逻辑示例 def get_qa_chain(use_fallback=False): llm = fallback_llm if use_fallback else main_llm return RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=vectorstore.as_retriever(), return_source_documents=True )

这段代码展示了如何通过一个开关控制使用主模型还是轻量模型。当监控系统发现 GPU 利用率持续高于90%，或平均推理延迟超过5秒，就可以自动触发降级，将流量导向flan-t5-small这类仅需2GB显存的小模型。虽然回答质量略有下降，但至少保证了“有答”，而非“无响应”。

再来看大型语言模型本身。LLM 是整个系统的“心脏”，但也最脆弱。本地部署的模型如 ChatGLM3-6B 或 Qwen-7B，通常需要12GB以上显存才能流畅运行。一旦并发请求增多，很容易出现 OOM（Out of Memory）错误。

除了切换模型外，还可以从生成策略上做优化。比如关闭采样（do_sample=False），改用贪婪解码，显著降低计算开销；限制最大输出长度（max_new_tokens=256），防止模型陷入无限生成；甚至可以预先缓存高频问题的答案，命中即返回，完全绕过推理过程。

实际工程中，我们曾在某金融客户部署时设置如下规则：
- 当前并发请求数 > 50：启用 Redis 缓存层，TTL设为1小时；
- 模型错误率连续5分钟 > 10%：触发告警并自动切换至轻量模型；
- 单次响应时间 > 8s：中断生成，返回“当前咨询量较大，请稍后再试”提示。

这套组合拳最终帮助其实现了灰度期间99.2%的可用性，平均响应时间稳定在1.8秒以内。

当然，也不能忽视向量数据库的表现。随着知识库扩大到数万条文档，检索效率往往会成为新的瓶颈。Chroma 虽然轻便易用，但在百万级向量下性能衰减明显。此时可以考虑分级检索策略：

1. 先根据元数据过滤（如部门、文档类别），缩小搜索范围；
2. 再在子集中进行向量相似度匹配；
3. 设置超时阈值（如1.5秒），超时则降级为关键词检索或直接返回空结果。

# 示例：带超时控制的检索封装 import signal class TimeoutError(Exception): pass def timeout_handler(signum, frame): raise TimeoutError("Vector search timed out") def safe_retrieve(query, timeout=2): signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: results = collection.query( query_embeddings=embed_model.encode([query]).tolist(), n_results=3 ) signal.alarm(0) # 取消定时器 return results except TimeoutError: return {"documents": [], "metadatas": []}

这种方法虽然牺牲了一定召回率，但避免了因一次慢查询拖垮整个服务的风险。

另一个容易被忽略的问题是权限与安全。即便系统部署在内网，也不意味着万事大吉。曾有案例显示，某员工通过构造特定查询，成功检索到了本不应看到的薪酬制度文件。为此，我们在架构中增加了多层防护：

• 前置API网关集成 JWT 鉴权，确保每个请求都有身份标识；
• 向量数据库按部门划分 Collection，实现物理隔离；
• 敏感文档打标签，并在检索前做权限校验；
• 所有查询记录写入审计日志，支持事后追溯。

这些措施看似增加了复杂度，但在涉及人事、财务等敏感领域时，却是必不可少的底线保障。

最后，真正的稳定性建设离不开可观测性。没有监控的数据就像盲人摸象。我们建议至少采集以下几类指标：

结合 Prometheus + Grafana 搭建可视化面板，配合 Alertmanager 设置动态阈值告警，才能做到问题早发现、早干预。

回到最初的问题：为什么需要服务降级预案？

因为技术从来不是孤岛。一个AI问答系统能否真正落地，不仅取决于模型多强大、效果多惊艳，更在于它能否在压力下“活着”。而服务降级，就是系统学会“求生”的第一步。

未来，随着 MoE（Mixture of Experts）架构和边缘推理的发展，这类系统有望进一步向端侧迁移——想象一下，每位员工的笔记本上都运行着专属的知识助手，既无需联网，又能实时响应。那一天或许不会太远。

而现在我们要做的，就是在通往那条路上，把每一块基石夯实。