LangChain WebUI 部署智能客服：从零搭建到生产环境优化

Ne0inhk

21 Mar 2026 — 12 min read

背景痛点：传统智能客服的局限与 LangChain 的破局

在构建智能客服系统的实践中，传统方案常常面临几个核心痛点。首先是状态管理困难，多轮对话的上下文（Context）需要在服务器端持久化，简单的内存存储无法应对服务重启或分布式部署，而引入数据库又会带来性能损耗和复杂性。其次是扩展性差，当需要接入新的知识库、工具（如查询订单、天气）或更换大语言模型（LLM）时，往往需要大动干戈地修改核心代码。最后是对话流编排复杂，一个完整的客服流程可能涉及意图识别、知识检索、条件分支、工具调用等多个环节，手动编写和维护这些逻辑链条极易出错。

LangChain 框架的出现，为这些问题提供了优雅的解决方案。它本质上是一个用于构建由 LLM 驱动的应用程序的框架，其核心优势在于模块化和链式编排。它将对话系统拆解为可复用的组件，如提示词模板（Prompt Templates）、记忆（Memory）、检索器（Retrievers）和工具（Tools），并通过“链”（Chains）的概念将它们灵活地组装起来。这使得开发者可以像搭积木一样构建复杂的对话逻辑，同时轻松管理多轮对话的上下文状态。

技术选型：FastAPI 还是 Django？

为 LangChain 智能客服构建 WebUI 后端，FastAPI 和 Django 是两个主流选择。选择哪一个，取决于项目对性能、开发速度和复杂度的要求。

Django 是一个“大而全”的全栈 Web 框架，自带 ORM、Admin 后台、用户认证等众多开箱即用的功能。如果你的智能客服系统需要复杂的数据模型、强大的后台管理界面，并且团队对 Django 更熟悉，它是一个稳妥的选择。然而，其同步的架构在处理 LangChain 可能涉及的长时间运行、高并发的 AI 请求时，可能需要搭配 Celery 等异步任务队列，架构会变得稍重。

FastAPI 是一个现代的、高性能的异步 Web 框架。它天生支持异步请求处理，与 LangChain 的异步调用（如异步的 LLM 接口、数据库查询）能完美契合，可以更高效地处理并发连接，特别适合需要流式响应（Streaming Response） 的对话场景。此外，它基于 Pydantic 提供了自动的请求/响应数据验证和 OpenAPI 文档生成，开发体验非常流畅。

吞吐量测试数据参考：在相同硬件条件下，处理简单的 API 请求，FastAPI 的吞吐量（Requests per second）通常比同步的 Django 高一个数量级。当涉及 I/O 密集型操作（如调用外部 LLM API、查询向量数据库）时，FastAPI 的异步优势更为明显。

选择建议：

追求极致性能、需要处理大量并发流式请求、偏好声明式 API 开发，选择 FastAPI。
需要快速构建包含复杂业务逻辑和后台管理的完整应用，且团队熟悉其生态，选择 Django。

本文后续将以 FastAPI 为核心进行演示。

核心实现：构建 LangChain 对话引擎

1. 使用 ConversationChain 构建对话逻辑

ConversationChain 是 LangChain 中用于处理对话的最简单链。它自动将对话历史（上下文）注入到每次与 LLM 的交互中。

from langchain.chains import ConversationChain from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferMemory # 1. 初始化大语言模型，例如使用 OpenAI GPT # 注意：此处需替换为你的实际 API Key，生产环境应从环境变量读取 llm = ChatOpenAI( model_name="gpt-3.5-turbo", temperature=0.7, # 控制创造性，客服场景建议调低（如0.2）以保持稳定 openai_api_key="your-api-key-here" ) # 2. 初始化对话记忆（Memory），用于存储上下文 memory = ConversationBufferMemory(return_messages=True) # 3. 创建对话链 conversation_chain = ConversationChain( llm=llm, memory=memory, verbose=True # 设置为 True 可在控制台查看详细的链执行过程，调试用 ) # 4. 进行对话 response = conversation_chain.run("你好，我想咨询一下退货政策。") print(f"AI: {response}") # 后续对话会自动包含之前的上下文 response2 = conversation_chain.run("如果我商品已经拆封了，还能退吗？") print(f"AI: {response2}")

2. 集成 RAG（检索增强生成）模块

单纯的 LLM 可能无法回答特定领域或实时性的问题。RAG 通过从外部知识库检索相关信息，并将其作为上下文提供给 LLM，从而生成更准确、可靠的回答。

from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import TextLoader from langchain.chains import RetrievalQA # 1. 加载并分割知识库文档（例如公司客服手册） loader = TextLoader("./knowledge_base/customer_service_manual.txt") documents = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) texts = text_splitter.split_documents(documents) # 2. 创建向量存储（Vector Store） embeddings = OpenAIEmbeddings(openai_api_key="your-api-key-here") vectorstore = Chroma.from_documents(texts, embeddings, persist_directory="./chroma_db") # 持久化到磁盘，下次可直接加载：vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) # 3. 将检索器与 LLM 结合，创建 RetrievalQA 链 retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 检索最相关的3个片段 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 将检索到的所有文档“塞”进提示词 retriever=retriever, return_source_documents=True # 返回参考来源，便于前端展示或审计 ) # 4. 提问 result = qa_chain.run("你们的七天无理由退货条件是什么？") print(f"Answer: {result['result']}") print(f"Sources: {[doc.metadata for doc in result['source_documents']]}")

3. WebSocket 实现双工通信

对于智能客服，WebSocket 是实现实时、双向对话的理想选择，尤其适合流式输出 LLM 的回复。

# websocket_handler.py from fastapi import WebSocket, WebSocketDisconnect from langchain.callbacks import AsyncIteratorCallbackHandler import asyncio import json class ConnectionManager: def __init__(self): self.active_connections: List[WebSocket] = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def send_text(self, websocket: WebSocket, text: str): await websocket.send_text(text) manager = ConnectionManager() async def handle_conversation(websocket: WebSocket, user_id: str): """ 处理 WebSocket 对话的主函数 """ await manager.connect(websocket) try: # 为每个会话创建独立的记忆，键名使用 user_id if user_id not in st.session_state: st.session_state[user_id] = ConversationBufferMemory(return_messages=True) memory = st.session_state[user_id] # 创建带有流式回调的链 callback_handler = AsyncIteratorCallbackHandler() streaming_llm = ChatOpenAI( streaming=True, callbacks=[callback_handler], **llm_kwargs ) conversation = ConversationChain(llm=streaming_llm, memory=memory) while True: # 1. 接收用户消息 user_message = await websocket.receive_text() data = json.loads(user_message) message = data.get("message", "") if not message: await websocket.send_text(json.dumps({"error": "Empty message"})) continue # 2. 启动一个异步任务来流式生成 AI 回复 async def generate_response(): try: # 这里使用 arun 进行异步运行 task = asyncio.create_task(conversation.arun(input=message)) async for token in callback_handler.aiter(): # 将每个 token 流式发送给前端 await websocket.send_text(json.dumps({"token": token})) await websocket.send_text(json.dumps({"status": "[DONE]"})) except Exception as e: # 捕获生成过程中的异常并通知客户端 error_msg = json.dumps({"error": f"Generation failed: {str(e)}"}) await websocket.send_text(error_msg) # 可以选择记录日志 print(f"Error for user {user_id}: {e}") asyncio.create_task(generate_response()) except WebSocketDisconnect: # 客户端断开连接 manager.disconnect(websocket) print(f"User {user_id} disconnected.") except json.JSONDecodeError: await websocket.send_text(json.dumps({"error": "Invalid JSON format"})) manager.disconnect(websocket) except Exception as e: # 捕获其他未预料到的异常 print(f"Unexpected error for user {user_id}: {e}") try: await websocket.send_text(json.dumps({"error": "Internal server error"})) except: pass finally: manager.disconnect(websocket) # 在 FastAPI 路由中挂载 @app.websocket("/ws/{user_id}") async def websocket_endpoint(websocket: WebSocket, user_id: str): await handle_conversation(websocket, user_id)

生产级优化：保障稳定与安全

1. 使用 Redis 进行会话状态缓存

在生产环境中，内存记忆（ConversationBufferMemory）不可靠，需要使用外部存储。Redis 是高性能键值数据库，非常适合存储会话状态。

from langchain.memory import RedisChatMessageHistory from langchain.memory import ConversationBufferMemory def get_memory_for_session(session_id: str, redis_url: str = "redis://localhost:6379/0"): """ 为指定会话创建基于 Redis 的记忆 """ message_history = RedisChatMessageHistory( session_id=session_id, # 使用唯一的会话ID，如用户ID或随机生成的UUID url=redis_url, key_prefix="langchain:memory:" # 添加前缀便于管理 ) memory = ConversationBufferMemory( memory_key="chat_history", chat_memory=message_history, return_messages=True ) return memory # 在对话链中使用 session_id = "user_12345" memory = get_memory_for_session(session_id) conversation = ConversationChain(llm=llm, memory=memory)

2. 负载测试方案（Locust 脚本示例）

上线前，必须对服务进行压力测试。Locust 是一个用 Python 编写的易用的负载测试工具。

# locustfile.py from locust import HttpUser, task, between, events import websocket import json import threading import time class WebSocketClient: def __init__(self, user_id): self.user_id = user_id self.ws = None self.received_messages = [] def connect(self): ws_url = f"ws://localhost:8000/ws/{self.user_id}" self.ws = websocket.WebSocketApp( ws_url, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close ) wst = threading.Thread(target=self.ws.run_forever) wst.daemon = True wst.start() time.sleep(1) # 等待连接建立 def send_message(self, message): if self.ws and self.ws.sock and self.ws.sock.connected: self.ws.send(json.dumps({"message": message})) def on_message(self, ws, message): self.received_messages.append(json.loads(message)) def on_error(self, ws, error): print(f"WebSocket Error for {self.user_id}: {error}") def on_close(self, ws, close_status_code, close_msg): print(f"WebSocket closed for {self.user_id}") class ChatUser(HttpUser): wait_time = between(1, 3) # 用户任务执行间隔 1-3 秒 abstract = True def on_start(self): """每个虚拟用户启动时，建立 WebSocket 连接""" self.client = WebSocketClient(self.environment.runner.user_id) self.client.connect() @task(1) def send_chat_message(self): """模拟发送一条消息并等待流式响应结束""" self.client.send_message("你好，介绍一下你们的产品。") # 简单等待一段时间模拟用户阅读，实际可更复杂地解析响应 time.sleep(5) class WebsiteUser(ChatUser): host = "http://localhost:8000"

运行测试：locust -f locustfile.py --host=http://localhost:8000，然后访问 Web UI 配置并发用户数和孵化速率。

3. JWT 鉴权与速率限制实现

保护 API 接口，防止滥用。

# auth.py from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from jose import JWTError, jwt from datetime import datetime, timedelta from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded import secrets # JWT 配置 SECRET_KEY = secrets.token_urlsafe(32) # 生产环境应从安全配置读取 ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 security = HTTPBearer() def create_access_token(data: dict): to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM]) user_id: str = payload.get("sub") if user_id is None: raise credentials_exception except JWTError: raise credentials_exception return user_id # 速率限制配置 limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # 受保护的路由示例 @app.post("/api/chat") @limiter.limit("10/minute") # 限制每分钟10次请求 async def protected_chat_endpoint(request: Request, message: str, current_user: str = Depends(get_current_user)): # 这里处理聊天逻辑，current_user 是 JWT 中的用户ID return {"user": current_user, "response": "模拟回复"}

避坑指南：实战经验总结

1. 解决 LangChain 内存泄漏的 3 种方法

长时间运行 LangChain 服务，尤其是处理大量会话时，可能会遇到内存缓慢增长的问题。

方法一：及时清理记忆对象。对于使用 ConversationBufferMemory 的场景，如果会话生命周期结束（如用户退出），应主动将其从全局存储中移除，并确保没有其他对象引用它，以便 Python 垃圾回收器能正常工作。
方法三：监控并重启 Worker。在生产环境使用 Gunicorn 或 Uvicorn 管理 FastAPI 进程时，可以设置 --max-requests 和 --max-requests-jitter 参数，让 Worker 在处理一定数量的请求后优雅重启，释放可能积累的内存碎片。

方法二：使用带窗口或摘要的记忆。ConversationBufferWindowMemory 只保留最近 K 轮对话，ConversationSummaryMemory 则对历史对话进行摘要，都能有效控制单个会话的内存占用。

from langchain.memory import ConversationSummaryMemory memory = ConversationSummaryMemory(llm=llm, return_messages=True)

2. 异步任务队列的选型建议（Celery vs RQ）

对于耗时较长的任务（如文档索引生成、复杂分析），应将其放入后台任务队列，避免阻塞 Web 请求。

Celery：功能强大、生态成熟，支持多种消息代理（RabbitMQ, Redis）、任务调度、工作流（Canvas）。配置相对复杂，但适合大型、需要复杂任务编排的项目。
RQ (Redis Queue)：轻量级、简单易用，只基于 Redis。如果你的项目已经使用 Redis 作为缓存和会话存储，并且后台任务需求不复杂（简单的异步执行、重试），RQ 是更快速上手的方案。它学习曲线平缓，与 Flask/FastAPI 集成简单。

建议：如果项目规模中等，且团队希望快速实现异步任务，优先选择 RQ。如果需要分布式任务、复杂的定时调度或任务链，则选择 Celery。

总结

通过 LangChain 构建智能客服 WebUI，我们能够以模块化的方式高效处理对话逻辑、上下文管理和知识检索。选择 FastAPI 作为异步后端为高并发和流式响应提供了坚实基础。在生产环境中，引入 Redis 管理状态、实施 JWT 鉴权和速率限制、进行充分的负载测试，是保障服务稳定和安全的关键步骤。同时，注意内存泄漏的防范和合理选择异步任务队列，能帮助系统长期平稳运行。这套从零到生产的实践路径，希望能为开发者部署自己的 AI 对话应用提供清晰的指引和可靠的代码基础。