LangChain实战：用Ollama本地大模型搭建智能客服系统（完整代码+配置流程）
·
LangChain实战：用Ollama本地大模型搭建智能客服系统（完整代码+配置流程）
最近和几个做企业服务的朋友聊天，他们都在头疼同一个问题：客服成本越来越高，但服务质量却很难标准化。招人难、培训周期长、夜间服务覆盖不足，这些痛点几乎成了行业通病。有人尝试过市面上的SaaS客服机器人，但要么效果不佳，要么担心数据安全问题，尤其是涉及内部知识库和客户隐私的时候。
这让我想起了去年自己折腾的一个项目——用本地大模型搭建智能客服。当时主要考虑两点：一是数据完全留在本地，安全可控；二是可以深度定制，贴合业务场景。经过几个月的迭代，这套基于LangChain和Ollama的方案已经能处理大部分常见咨询，甚至在特定领域的回答比人工客服更精准。
今天我就把这套系统的完整搭建过程拆解出来，从环境准备到代码实现，再到性能调优，手把手带你走一遍。无论你是想给现有业务加个智能助手，还是单纯想探索本地大模型的应用，这篇文章都能给你实实在在的参考。
1. 系统架构设计与核心组件选型
搭建一个能用的客服系统和搭建一个好用的客服系统，中间隔着好几个层级。我们先从顶层设计开始，理清楚各个模块的职责和交互方式。
1.1 整体架构分层
我设计的这套系统采用典型的分层架构，从上到下依次是交互层、应用层、服务层和数据层。这种设计的好处是模块清晰，后期扩展和维护都比较方便。
用户界面 (Web/API)
│
▼
应用层 (LangChain Chain/Agent)
│
▼
服务层 (向量检索/记忆管理)
│
▼
数据层 (文档/向量库/配置)
交互层负责与用户对接，可以是Web界面、API接口，甚至集成到微信、钉钉等平台。考虑到大多数企业的实际情况，我建议先从API开始，这样前端可以灵活选择。
应用层是整个系统的智能核心，基于LangChain构建。这里需要设计合适的Chain（处理链）或Agent（智能体）来组织对话流程。我的经验是，对于客服场景，一个精心设计的Chain往往比复杂的Agent更稳定可靠。
服务层提供关键能力支持，主要包括：
- 向量检索服务：用于从知识库中快速找到相关文档
- 对话记忆管理：记住用户的历史对话，提供连贯的体验
- 意图识别模块：判断用户问题的类型，路由到不同的处理流程
数据层存储所有需要的信息，包括：
- 原始文档（产品手册、FAQ、政策文件等）
- 向量化后的知识库
- 对话历史记录
- 系统配置信息
1.2 核心组件技术选型
选型是个技术活，既要考虑功能需求，也要权衡开发成本和维护难度。下面是我经过实际测试后的推荐方案：
| 组件 | 推荐方案 | 备选方案 | 选择理由 |
|---|---|---|---|
| 大模型 | Ollama + Llama 3.1 8B | DeepSeek-R1, Qwen2.5 | Llama 3.1在8B参数级别表现均衡，中英文支持好，推理速度适中 |
| 向量数据库 | ChromaDB | FAISS, Pinecone | Chroma轻量易用，支持持久化，Python集成友好 |
| 文本分割器 | RecursiveCharacterTextSplitter | TokenTextSplitter | 按字符递归分割能更好保持语义完整性 |
| 嵌入模型 | BAAI/bge-small-zh-v1.5 | text-embedding-ada-002 | 专门优化中文，本地部署，无需API调用 |
| Web框架 | FastAPI | Flask, Django | 异步支持好，性能出色，自动生成API文档 |
这里重点说说模型选择。很多人一上来就想用最大的模型，觉得参数越多效果越好。但实际部署时你会发现，模型大小直接影响到响应速度和硬件成本。对于客服场景，大部分问题都是相对标准的，不需要太强的推理能力。
我测试过几个主流模型在客服问答上的表现：
# 模型响应速度测试对比（本地RTX 4060显卡）
models = {
"llama3.1:8b": "响应时间: 2-3秒，质量: 优秀",
"qwen2.5:7b": "响应时间: 1.5-2.5秒，质量: 良好",
"deepseek-r1:7b": "响应时间: 3-4秒，质量: 优秀（推理强）",
"mistral:7b": "响应时间: 1-2秒，质量: 良好"
}
从实际使用来看，Llama 3.1 8B在中文客服场景下表现最均衡。它的回答比较规范，不太会“胡言乱语”，这对于企业应用很重要。如果对推理能力要求高（比如需要多步计算的问题），可以考虑DeepSeek-R1，但要注意它的响应时间会稍长一些。
提示：选择模型时一定要在自己的数据集上做AB测试。不同模型在不同类型问题上的表现差异很大，别人的推荐只能作为参考。
1.3 知识库设计原则
知识库的质量直接决定客服系统的上限。很多人以为把文档扔进去就行，其实这里面有很多讲究。
首先，文档预处理是关键一步。我通常会把原始文档（PDF、Word、网页等）转换成纯文本，然后进行清洗：
- 移除无关的页眉页脚、水印
- 处理表格数据，保持结构信息
- 统一格式和编码
- 分段处理，确保每个段落语义完整
其次，分块策略直接影响检索效果。太小的块会丢失上下文，太大的块会引入噪声。我的经验是：
- 对于FAQ类文档，按问题-答案对分块
- 对于手册类文档，按章节或主题分块，每块500-800字
- 重叠设置150-200字，确保边界问题也能被检索到
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 推荐的分块配置
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=800, # 每个块的最大字符数
chunk_overlap=150, # 块之间的重叠字符数
separators=["\n\n", "\n", "。", "；", "，", " ", ""] # 分割符优先级
)
最后，元数据管理经常被忽视，但很有用。给每个文档块添加元数据，比如来源文档、创建时间、主题标签等，可以在检索时进行过滤，提高精准度。
2. 环境搭建与依赖安装
环境配置是第一步，也是最容易踩坑的地方。我建议按照下面的顺序一步步来，确保每个环节都验证通过。
2.1 基础环境准备
硬件配置方面，如果你的客服系统预计并发量不大（比如同时在线用户<100），那么一台配置不错的台式机就够用了。我的开发机配置供你参考：
- CPU: Intel i7-13700K（16核24线程）
- 内存: 32GB DDR5
- 显卡: NVIDIA RTX 4060 12GB（如果只用CPU推理，显卡可选）
- 存储: 1TB NVMe SSD
- 系统: Ubuntu 22.04 LTS / Windows 11 WSL2
软件环境需要提前准备好：
- Python 3.10+：3.10是个比较稳定的版本，新特性支持好，社区包兼容性也不错
- CUDA 12.1（如果使用GPU加速）：NVIDIA显卡必备
- Docker（可选）：方便部署和隔离环境
注意：Windows用户强烈建议使用WSL2（Windows Subsystem for Linux），很多AI相关的库在Linux环境下更稳定。WSL2的安装和配置微软官方有详细文档，这里不展开。
2.2 Ollama安装与模型部署
Ollama是目前最方便的本地大模型管理工具，支持一键安装、模型下载和API服务。
安装Ollama：
# Linux/macOS一键安装
curl -fsSL https://ollama.ai/install.sh | sh
# Windows用户直接下载安装包
# 访问 https://ollama.ai/download 下载exe文件
安装完成后，启动Ollama服务：
# 启动服务（Linux/macOS）
ollama serve
# Windows会自动创建服务，可以在任务管理器查看
下载模型：
# 拉取Llama 3.1 8B模型（推荐）
ollama pull llama3.1:8b
# 如果想试试其他模型
ollama pull qwen2.5:7b
ollama pull deepseek-r1:7b
模型下载完成后，验证是否正常运行：
# 测试模型响应
ollama run llama3.1:8b "你好，请介绍一下你自己"
# 或者通过API测试
curl http://localhost:11434/api/generate -d '{
"model": "llama3.1:8b",
"prompt": "你好",
"stream": false
}'
如果看到正常的回复，说明模型部署成功。第一次运行可能会慢一些，因为要加载模型到内存。
2.3 Python环境与依赖包
我习惯为每个项目创建独立的虚拟环境，避免包冲突。这里用conda和venv都可以，看个人习惯。
# 创建虚拟环境（conda方式）
conda create -n langchain-cs python=3.10
conda activate langchain-cs
# 或者使用venv
python -m venv langchain-cs
source langchain-cs/bin/activate # Linux/macOS
langchain-cs\Scripts\activate # Windows
安装核心依赖包：
# 基础框架
pip install langchain langchain-core langchain-community
# Ollama集成
pip install langchain-ollama
# 向量数据库和嵌入模型
pip install chromadb sentence-transformers
# Web框架和工具
pip install fastapi uvicorn python-multipart
# 文档处理（按需安装）
pip install pypdf python-docx beautifulsoup4 markdown
验证安装是否成功：
import langchain
print(f"LangChain版本: {langchain.__version__}")
from langchain_ollama import OllamaLLM
llm = OllamaLLM(model="llama3.1:8b")
response = llm.invoke("测试")
print(f"模型响应: {response[:50]}...")
如果一切正常，你会看到LangChain的版本号和模型的简短回复。到这里，基础环境就准备好了。
3. 核心功能模块实现
有了基础环境，我们开始实现系统的核心功能。我会按照数据准备、检索增强、对话管理的顺序来讲解。
3.1 知识库构建与向量化
知识库是客服系统的“大脑”，构建过程需要仔细处理。我通常把这个过程分为四个步骤：文档加载、文本分割、向量化、存储索引。
第一步：文档加载 LangChain提供了丰富的文档加载器，支持各种格式：
from langchain_community.document_loaders import (
PyPDFLoader,
DocxLoader,
TextLoader,
WebBaseLoader,
UnstructuredMarkdownLoader
)
# 根据文件类型选择加载器
def load_documents(file_path):
if file_path.endswith('.pdf'):
loader = PyPDFLoader(file_path)
elif file_path.endswith('.docx'):
loader = DocxLoader(file_path)
elif file_path.endswith('.txt'):
loader = TextLoader(file_path, encoding='utf-8')
elif file_path.endswith('.md'):
loader = UnstructuredMarkdownLoader(file_path)
else:
raise ValueError(f"不支持的格式: {file_path}")
documents = loader.load()
return documents
# 批量加载文档
documents = []
for file_path in document_files:
docs = load_documents(file_path)
documents.extend(docs)
print(f"共加载 {len(documents)} 个文档")
第二步：文本分割与处理 直接使用原始文档效果不好，需要合理分割：
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document
def process_documents(documents, chunk_size=800, chunk_overlap=150):
"""处理文档，包括分割和添加元数据"""
# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", "。", "；", "，", " ", ""]
)
# 分割文档
chunks = text_splitter.split_documents(documents)
# 为每个块添加元数据
processed_chunks = []
for i, chunk in enumerate(chunks):
# 提取或生成摘要
if len(chunk.page_content) > 100:
summary = chunk.page_content[:100] + "..."
else:
summary = chunk.page_content
# 添加元数据
metadata = chunk.metadata.copy()
metadata.update({
"chunk_id": i,
"summary": summary,
"word_count": len(chunk.page_content),
"process_time": datetime.now().isoformat()
})
processed_chunks.append(Document(
page_content=chunk.page_content,
metadata=metadata
))
return processed_chunks
第三步：向量化与存储 这里我选择ChromaDB作为向量数据库，BGE模型做嵌入：
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
import os
class KnowledgeBase:
def __init__(self, persist_directory="./chroma_db"):
self.persist_directory = persist_directory
self.embeddings = None
self.vectorstore = None
def init_embeddings(self):
"""初始化嵌入模型"""
# 使用BAAI的中文嵌入模型
model_name = "BAAI/bge-small-zh-v1.5"
self.embeddings = HuggingFaceEmbeddings(
model_name=model_name,
model_kwargs={'device': 'cpu'}, # 可用'cuda'加速
encode_kwargs={'normalize_embeddings': True}
)
return self.embeddings
def create_vectorstore(self, documents):
"""创建向量存储"""
if not self.embeddings:
self.init_embeddings()
# 创建向量存储
self.vectorstore = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
# 持久化
self.vectorstore.persist()
print(f"向量库创建完成，共 {len(documents)} 个文档块")
return self.vectorstore
def load_vectorstore(self):
"""加载已有的向量存储"""
if not self.embeddings:
self.init_embeddings()
if os.path.exists(self.persist_directory):
self.vectorstore = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings
)
print("向量库加载成功")
return self.vectorstore
else:
print("向量库不存在，请先创建")
return None
def similarity_search(self, query, k=5, filter_dict=None):
"""相似度搜索"""
if not self.vectorstore:
self.load_vectorstore()
results = self.vectorstore.similarity_search(
query=query,
k=k,
filter=filter_dict
)
return results
实际使用时，你可以这样构建知识库：
# 初始化知识库
kb = KnowledgeBase()
# 加载和处理文档
raw_docs = load_documents("产品手册.pdf")
processed_docs = process_documents(raw_docs)
# 创建向量存储
vectorstore = kb.create_vectorstore(processed_docs)
# 测试检索
results = kb.similarity_search("如何退货？", k=3)
for i, doc in enumerate(results):
print(f"结果 {i+1}: {doc.metadata.get('summary', '无摘要')}")
3.2 检索增强生成（RAG）实现
RAG是当前最实用的知识增强方案，它让模型能够基于特定知识回答问题，而不是仅凭训练数据“编造”。
基础RAG链实现：
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain_ollama import OllamaLLM
class RAGChatbot:
def __init__(self, model_name="llama3.1:8b"):
self.llm = OllamaLLM(
model=model_name,
temperature=0.1, # 低温度使回答更确定
num_predict=512 # 最大生成长度
)
self.vectorstore = None
self.qa_chain = None
def init_chain(self, vectorstore):
"""初始化RAG链"""
self.vectorstore = vectorstore
# 自定义提示模板
prompt_template = """你是一个专业的客服助手，请根据提供的上下文信息回答问题。
如果上下文中有相关信息，请基于上下文回答。
如果上下文中没有相关信息，请根据你的知识诚实回答"我不知道"。
上下文：
{context}
问题：{question}
请用中文回答，保持专业和友好："""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
# 创建检索器
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 4} # 检索4个最相关的文档块
)
# 创建RAG链
self.qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff", # 简单合并上下文
retriever=retriever,
chain_type_kwargs={"prompt": PROMPT},
return_source_documents=True
)
return self.qa_chain
def ask(self, question):
"""提问并获取回答"""
if not self.qa_chain:
raise ValueError("请先初始化RAG链")
result = self.qa_chain({"query": question})
return {
"answer": result["result"],
"sources": [
{
"content": doc.page_content[:200] + "...",
"metadata": doc.metadata
}
for doc in result["source_documents"]
]
}
高级RAG优化技巧： 基础RAG有时候效果不够理想，特别是当问题复杂或者知识库庞大时。我总结了几个优化点：
- 重排序（Re-ranking）：相似度搜索返回的结果不一定最相关，可以用更精细的模型重新排序。
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name="BAAI/bge-reranker-base"):
self.model = CrossEncoder(model_name)
def rerank(self, query, documents, top_k=3):
"""对检索结果重新排序"""
pairs = [[query, doc.page_content] for doc in documents]
scores = self.model.predict(pairs)
# 按分数排序
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, score in ranked[:top_k]]
# 在RAG链中使用重排序
def enhanced_retrieval(query, vectorstore, reranker, k=10, top_k=3):
# 先检索更多结果
docs = vectorstore.similarity_search(query, k=k)
# 重排序
ranked_docs = reranker.rerank(query, docs, top_k=top_k)
return ranked_docs
- 查询扩展：有时候用户的问题表述不完整，可以自动扩展查询词。
def expand_query(original_query, llm):
"""使用LLM扩展查询词"""
expansion_prompt = f"""
原始问题：{original_query}
请生成3个相关的查询，用于在知识库中搜索相关信息。
每个查询应该从不同角度表达相同或相关的意图。
输出格式：
1. [查询1]
2. [查询2]
3. [查询3]
"""
response = llm.invoke(expansion_prompt)
# 解析响应，提取查询词
queries = [original_query] # 包含原始查询
lines = response.split('\n')
for line in lines:
if line.strip() and any(char.isdigit() for char in line[:3]):
query = line.split('.', 1)[-1].strip()
if query:
queries.append(query)
return queries[:4] # 最多4个查询
- 混合检索：结合关键词搜索和向量搜索，取长补短。
from langchain.retrievers import BM25Retriever
from langchain.retrievers.ensemble import EnsembleRetriever
def create_hybrid_retriever(vectorstore, text_docs):
"""创建混合检索器"""
# 向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# BM25关键词检索器
bm25_retriever = BM25Retriever.from_documents(text_docs)
bm25_retriever.k = 5
# 集成检索器
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.7, 0.3] # 向量检索权重更高
)
return ensemble_retriever
3.3 对话管理与记忆机制
好的客服系统需要记住对话历史，否则每次都要用户重复信息，体验很差。LangChain提供了多种记忆机制，我根据客服场景做了定制。
对话记忆实现：
from langchain.memory import ConversationBufferWindowMemory
from langchain.schema import BaseMessage, HumanMessage, AIMessage
import json
from datetime import datetime
class CustomerServiceMemory:
def __init__(self, window_size=10, persist_file="./memory.json"):
self.memory = ConversationBufferWindowMemory(
memory_key="chat_history",
return_messages=True,
k=window_size # 记住最近10轮对话
)
self.persist_file = persist_file
self.user_context = {} # 用户上下文信息
def add_context(self, user_id, key, value):
"""添加上下文信息"""
if user_id not in self.user_context:
self.user_context[user_id] = {}
self.user_context[user_id][key] = value
self._save_context()
def get_context(self, user_id, key=None):
"""获取上下文信息"""
if user_id not in self.user_context:
return None if key else {}
if key:
return self.user_context[user_id].get(key)
return self.user_context[user_id].copy()
def add_conversation(self, user_id, human_input, ai_response):
"""添加对话记录"""
# 转换为LangChain的消息格式
human_msg = HumanMessage(content=human_input)
ai_msg = AIMessage(content=ai_response)
# 添加到记忆
self.memory.chat_memory.add_message(human_msg)
self.memory.chat_memory.add_message(ai_msg)
# 提取可能的用户信息
self._extract_user_info(user_id, human_input, ai_response)
self._save_memory()
def _extract_user_info(self, user_id, human_input, ai_response):
"""从对话中提取用户信息（简化版）"""
# 这里可以集成NER模型来提取实体信息
info_keywords = {
"姓名": ["我叫", "我是", "名字是", "称呼我"],
"订单号": ["订单号", "订单编号", "order number"],
"手机号": ["手机", "电话", "联系方式"],
"邮箱": ["邮箱", "email", "电子邮件"]
}
for info_type, keywords in info_keywords.items():
for keyword in keywords:
if keyword in human_input:
# 简单提取，实际应该用更复杂的方法
start_idx = human_input.find(keyword)
if start_idx != -1:
# 提取关键词后的内容
extracted = human_input[start_idx + len(keyword):].strip()
if extracted and len(extracted) < 50: # 防止提取过多
self.add_context(user_id, info_type, extracted)
break
def get_conversation_history(self, user_id=None):
"""获取对话历史"""
# 这里简化处理，实际应该按用户ID隔离
history = self.memory.load_memory_variables({})
return history.get("chat_history", [])
def _save_memory(self):
"""保存记忆到文件"""
try:
data = {
"conversations": [
{
"role": msg.type,
"content": msg.content,
"timestamp": datetime.now().isoformat()
}
for msg in self.memory.chat_memory.messages[-20:] # 保存最近20条
],
"user_context": self.user_context,
"last_updated": datetime.now().isoformat()
}
with open(self.persist_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"保存记忆失败: {e}")
def _save_context(self):
"""保存上下文到文件"""
self._save_memory() # 简化处理，一起保存
def load_memory(self):
"""从文件加载记忆"""
try:
if os.path.exists(self.persist_file):
with open(self.persist_file, 'r', encoding='utf-8') as f:
data = json.load(f)
# 恢复对话历史
self.memory.chat_memory.clear()
for conv in data.get("conversations", []):
if conv["role"] == "human":
msg = HumanMessage(content=conv["content"])
else:
msg = AIMessage(content=conv["content"])
self.memory.chat_memory.add_message(msg)
# 恢复用户上下文
self.user_context = data.get("user_context", {})
print(f"记忆加载成功，共 {len(data.get('conversations', []))} 条记录")
return True
except Exception as e:
print(f"加载记忆失败: {e}")
return False
带记忆的RAG链： 有了记忆模块，我们需要修改RAG链来利用这些信息：
from langchain.chains import ConversationalRetrievalChain
class MemoryEnhancedChatbot(RAGChatbot):
def __init__(self, model_name="llama3.1:8b"):
super().__init__(model_name)
self.memory_manager = CustomerServiceMemory()
def init_conversational_chain(self, vectorstore):
"""初始化带记忆的对话链"""
# 加载记忆
self.memory_manager.load_memory()
# 创建对话检索链
self.qa_chain = ConversationalRetrievalChain.from_llm(
llm=self.llm,
retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
memory=self.memory_manager.memory,
chain_type="stuff",
verbose=False, # 调试时可设为True
max_tokens_limit=4000, # 上下文token限制
get_chat_history=lambda h: h # 获取历史函数
)
return self.qa_chain
def ask_with_context(self, user_id, question):
"""带上下文的提问"""
# 获取用户上下文
user_context = self.memory_manager.get_context(user_id)
# 如果有用户信息，可以增强问题
enhanced_question = question
if user_context:
# 简单示例：在问题前加上用户信息
context_str = " ".join([f"{k}:{v}" for k, v in user_context.items()])
enhanced_question = f"[用户信息: {context_str}] {question}"
# 执行查询
result = self.qa_chain({"question": enhanced_question})
# 保存对话
self.memory_manager.add_conversation(user_id, question, result["answer"])
return {
"answer": result["answer"],
"user_id": user_id,
"context_used": bool(user_context)
}
4. 系统集成与API服务
单个功能模块测试通过后，我们需要把它们集成起来，并通过API提供服务。这里我用FastAPI搭建Web服务，因为它性能好，异步支持完善。
4.1 FastAPI服务搭建
主应用结构：
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import uvicorn
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# 定义数据模型
class ChatRequest(BaseModel):
question: str
user_id: str = "anonymous"
session_id: Optional[str] = None
context: Optional[Dict[str, Any]] = None
class ChatResponse(BaseModel):
answer: str
session_id: str
timestamp: str
sources: Optional[List[Dict]] = None
processing_time: float
class KnowledgeUpdateRequest(BaseModel):
documents: List[str] # 文档路径列表
clear_existing: bool = False
class SystemStatus(BaseModel):
model: str
vectorstore_size: int
memory_entries: int
uptime: float
last_request: Optional[str] = None
# 创建FastAPI应用
app = FastAPI(
title="智能客服系统API",
description="基于LangChain和Ollama的本地大模型客服系统",
version="1.0.0"
)
# 添加CORS中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境应该限制域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 全局组件
chatbot = None
knowledge_base = None
@app.on_event("startup")
async def startup_event():
"""应用启动时初始化组件"""
global chatbot, knowledge_base
logger.info("正在初始化智能客服系统...")
try:
# 初始化知识库
from knowledge_base import KnowledgeBase
knowledge_base = KnowledgeBase(persist_directory="./data/chroma_db")
kb_status = knowledge_base.load_vectorstore()
if not kb_status:
logger.warning("知识库不存在，请先创建知识库")
# 初始化聊天机器人
from chatbot import MemoryEnhancedChatbot
chatbot = MemoryEnhancedChatbot(model_name="llama3.1:8b")
if kb_status:
chatbot.init_conversational_chain(knowledge_base.vectorstore)
logger.info("对话链初始化成功")
else:
logger.info("聊天机器人初始化完成（无知识库）")
logger.info("系统启动完成")
except Exception as e:
logger.error(f"系统初始化失败: {e}")
raise
@app.get("/")
async def root():
"""根路径，返回服务信息"""
return {
"service": "智能客服系统",
"version": "1.0.0",
"status": "running",
"endpoints": {
"chat": "/api/chat",
"knowledge": "/api/knowledge/update",
"status": "/api/status"
}
}
@app.post("/api/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
"""处理用户聊天请求"""
start_time = datetime.now()
if not chatbot:
raise HTTPException(status_code=503, detail="系统未初始化完成")
try:
# 处理用户问题
if request.context:
# 如果有额外上下文，添加到用户信息中
for key, value in request.context.items():
chatbot.memory_manager.add_context(
request.user_id, key, value
)
# 获取回答
result = chatbot.ask_with_context(
user_id=request.user_id,
question=request.question
)
# 计算处理时间
processing_time = (datetime.now() - start_time).total_seconds()
# 生成会话ID（如果未提供）
session_id = request.session_id or f"session_{datetime.now().timestamp()}"
response = ChatResponse(
answer=result["answer"],
session_id=session_id,
timestamp=datetime.now().isoformat(),
processing_time=processing_time
)
logger.info(f"用户 {request.user_id} 提问: {request.question[:50]}...")
logger.info(f"处理时间: {processing_time:.2f}秒")
return response
except Exception as e:
logger.error(f"处理聊天请求失败: {e}")
raise HTTPException(status_code=500, detail=f"处理失败: {str(e)}")
@app.post("/api/knowledge/update")
async def update_knowledge(request: KnowledgeUpdateRequest):
"""更新知识库"""
if not knowledge_base:
raise HTTPException(status_code=503, detail="知识库未初始化")
try:
from document_processor import load_and_process_documents
# 加载和处理文档
documents = load_and_process_documents(request.documents)
if request.clear_existing:
# 清空现有知识库（实际应该更安全地处理）
import shutil
if os.path.exists(knowledge_base.persist_directory):
shutil.rmtree(knowledge_base.persist_directory)
knowledge_base.vectorstore = None
# 创建或更新向量存储
vectorstore = knowledge_base.create_vectorstore(documents)
# 更新聊天机器人的检索器
if chatbot:
chatbot.init_conversational_chain(vectorstore)
return {
"status": "success",
"message": f"知识库更新成功，新增 {len(documents)} 个文档块",
"total_chunks": len(documents)
}
except Exception as e:
logger.error(f"更新知识库失败: {e}")
raise HTTPException(status_code=500, detail=f"更新失败: {str(e)}")
@app.get("/api/status", response_model=SystemStatus)
async def get_status():
"""获取系统状态"""
if not chatbot or not knowledge_base:
raise HTTPException(status_code=503, detail="系统未初始化")
# 获取向量库大小
vs_size = 0
if knowledge_base.vectorstore:
try:
# 这里需要根据实际向量库实现获取大小
vs_size = knowledge_base.vectorstore._collection.count()
except:
vs_size = -1 # 无法获取
# 获取记忆条目数
memory_count = len(chatbot.memory_manager.get_conversation_history())
# 计算运行时间（简化）
uptime = 3600 # 示例值，实际应该从启动时间计算
return SystemStatus(
model="llama3.1:8b",
vectorstore_size=vs_size,
memory_entries=memory_count,
uptime=uptime,
last_request=datetime.now().isoformat()
)
if __name__ == "__main__":
# 启动服务
uvicorn.run(
app,
host="0.0.0.0",
port=8000,
log_level="info"
)
API测试客户端： 为了方便测试，我写了一个简单的客户端：
import requests
import json
from typing import Dict, Any
class ChatClient:
def __init__(self, base_url="http://localhost:8000"):
self.base_url = base_url
self.session_id = None
def chat(self, question: str, user_id: str = "test_user") -> Dict[str, Any]:
"""发送聊天请求"""
url = f"{self.base_url}/api/chat"
payload = {
"question": question,
"user_id": user_id,
"session_id": self.session_id
}
try:
response = requests.post(url, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
# 保存会话ID
if not self.session_id:
self.session_id = result.get("session_id")
return result
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return {"error": str(e)}
def update_knowledge(self, document_paths: list, clear_existing: bool = False):
"""更新知识库"""
url = f"{self.base_url}/api/knowledge/update"
payload = {
"documents": document_paths,
"clear_existing": clear_existing
}
try:
response = requests.post(url, json=payload, timeout=60)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"更新知识库失败: {e}")
return {"error": str(e)}
def get_status(self):
"""获取系统状态"""
url = f"{self.base_url}/api/status"
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"获取状态失败: {e}")
return {"error": str(e)}
# 使用示例
if __name__ == "__main__":
client = ChatClient()
# 测试系统状态
status = client.get_status()
print(f"系统状态: {status}")
# 测试聊天
response = client.chat("你们公司的退货政策是什么？")
print(f"回答: {response['answer']}")
print(f"处理时间: {response['processing_time']}秒")
# 测试上下文记忆
response2 = client.chat("那我需要准备什么材料？")
print(f"后续回答: {response2['answer']}")
4.2 前端界面示例
虽然API服务已经足够，但有个简单的前端界面会更方便测试和演示。这里用HTML+JavaScript实现一个极简的聊天界面：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>智能客服系统</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
min-height: 100vh;
display: flex;
justify-content: center;
align-items: center;
padding: 20px;
}
.chat-container {
width: 100%;
max-width: 800px;
background: white;
border-radius: 20px;
box-shadow: 0 20px 60px rgba(0,0,0,0.3);
overflow: hidden;
display: flex;
flex-direction: column;
height: 80vh;
}
.chat-header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 20px;
text-align: center;
}
.chat-header h1 {
font-size: 24px;
margin-bottom: 5px;
}
.chat-header .status {
font-size: 14px;
opacity: 0.9;
}
.chat-messages {
flex: 1;
padding: 20px;
overflow-y: auto;
background: #f8f9fa;
}
.message {
margin-bottom: 15px;
display: flex;
}
.message.user {
justify-content: flex-end;
}
.message.bot {
justify-content: flex-start;
}
.message-content {
max-width: 70%;
padding: 12px 16px;
border-radius: 18px;
position: relative;
}
.user .message-content {
background: #667eea;
color: white;
border-bottom-right-radius: 4px;
}
.bot .message-content {
background: white;
color: #333;
border: 1px solid #e0e0e0;
border-bottom-left-radius: 4px;
}
.message-time {
font-size: 12px;
color: #999;
margin-top: 4px;
text-align: right;
}
.chat-input {
padding: 20px;
border-top: 1px solid #e0e0e0;
background: white;
}
.input-group {
display: flex;
gap: 10px;
}
#messageInput {
flex: 1;
padding: 12px 16px;
border: 2px solid #e0e0e0;
border-radius: 24px;
font-size: 16px;
outline: none;
transition: border-color 0.3s;
}
#messageInput:focus {
border-color: #667eea;
}
#sendButton {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
border: none;
padding: 12px 24px;
border-radius: 24px;
font-size: 16px;
cursor: pointer;
transition: transform 0.2s;
}
#sendButton:hover {
transform: translateY(-2px);
}
#sendButton:disabled {
opacity: 0.5;
cursor: not-allowed;
}
.typing-indicator {
display: none;
padding: 12px 16px;
background: white;
border-radius: 18px;
border: 1px solid #e0e0e0;
margin-bottom: 15px;
width: fit-content;
}
.typing-dots {
display: flex;
gap: 4px;
}
.typing-dots span {
width: 8px;
height: 8px;
background: #999;
border-radius: 50%;
animation: typing 1.4s infinite;
}
.typing-dots span:nth-child(2) {
animation-delay: 0.2s;
}
.typing-dots span:nth-child(3) {
animation-delay: 0.4s;
}
@keyframes typing {
0%, 60%, 100% {
transform: translateY(0);
}
30% {
transform: translateY(-8px);
}
}
.system-info {
font-size: 12px;
color: #666;
text-align: center;
padding: 10px;
background: #f8f9fa;
border-top: 1px solid #e0e0e0;
}
</style>
</head>
<body>
<div class="chat-container">
<div class="chat-header">
<h1>🤖 智能客服助手</h1>
<div class="status" id="status">系统已连接</div>
</div>
<div class="chat-messages" id="chatMessages">
<div class="message bot">
<div class="message-content">
您好！我是智能客服助手，有什么可以帮您的吗？
</div>
</div>
</div>
<div class="typing-indicator" id="typingIndicator">
<div class="typing-dots">
<span></span>
<span></span>
<span></span>
</div>
</div>
<div class="chat-input">
<div class="input-group">
<input type="text" id="messageInput" placeholder="请输入您的问题..." autocomplete="off">
<button id="sendButton" onclick="sendMessage()">发送</button>
</div>
</div>
<div class="system-info">
基于 LangChain + Ollama 本地大模型 | 数据完全本地处理，安全可靠
</div>
</div>
<script>
const API_BASE_URL = 'http://localhost:8000';
let sessionId = null;
let userId = 'user_' + Math.random().toString(36).substr(2, 9);
// 检查系统状态
async function checkStatus() {
try {
const response = await fetch(`${API_BASE_URL}/api/status`);
if (response.ok) {
const data = await response.json();
document.getElementById('status').textContent =
`模型: ${data.model} | 知识库: ${data.vectorstore_size}条`;
}
} catch (error) {
console.error('状态检查失败:', error);
document.getElementById('status').textContent = '系统连接异常';
}
}
// 发送消息
async function sendMessage() {
const input = document.getElementById('messageInput');
const message = input.value.trim();
const button = document.getElementById('sendButton');
if (!message) return;
// 禁用输入和按钮
input.disabled = true;
button.disabled = true;
// 添加用户消息到界面
addMessage(message, 'user');
input.value = '';
// 显示输入中状态
showTypingIndicator();
try {
// 发送请求到API
const response = await fetch(`${API_BASE_URL}/api/chat`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
question: message,
user_id: userId,
session_id: sessionId
})
});
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
const data = await response.json();
// 保存会话ID
if (!sessionId) {
sessionId = data.session_id;
}
// 隐藏输入中状态
hideTypingIndicator();
// 添加机器人回复到界面
addMessage(data.answer, 'bot');
// 滚动到底部
scrollToBottom();
} catch (error) {
console.error('发送消息失败:', error);
hideTypingIndicator();
addMessage('抱歉，服务暂时不可用，请稍后重试。', 'bot');
} finally {
// 恢复输入和按钮
input.disabled = false;
button.disabled = false;
input.focus();
}
}
// 添加消息到界面
function addMessage(content, sender) {
const messagesDiv = document.getElementById('chatMessages');
const messageDiv = document.createElement('div');
messageDiv.className = `message ${sender}`;
const time = new Date().toLocaleTimeString([], {
hour: '2-digit',
minute: '2-digit'
});
messageDiv.innerHTML = `
<div class="message-content">
${content}
<div class="message-time">${time}</div>
</div>
`;
messagesDiv.appendChild(messageDiv);
scrollToBottom();
}
// 显示输入中状态
function showTypingIndicator() {
document.getElementById('typingIndicator').style.display = 'block';
scrollToBottom();
}
// 隐藏输入中状态
function hideTypingIndicator() {
document.getElementById('typingIndicator').style.display = 'none';
}
// 滚动到底部
function scrollToBottom() {
const messagesDiv = document.getElementById('chatMessages');
messagesDiv.scrollTop = messagesDiv.scrollHeight;
}
// 回车发送消息
document.getElementById('messageInput').addEventListener('keypress', function(e) {
if (e.key === 'Enter') {
sendMessage();
}
});
// 初始化
document.addEventListener('DOMContentLoaded', function() {
checkStatus();
// 每30秒检查一次状态
setInterval(checkStatus, 30000);
});
</script>
</body>
</html>
这个前端界面虽然简单，但包含了核心功能：实时聊天、状态显示、响应时间展示等。你可以把它保存为index.html，用浏览器打开就能直接使用。
4.3 部署与配置管理
实际部署时，我们需要考虑更多生产环境的问题。这里提供一个Docker部署方案和配置管理。
Docker部署文件：
# Dockerfile
FROM python:3.10-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
gcc \
g++ \
curl \
&& rm -rf /var/lib/apt/lists/*
# 安装Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh
# 复制Python依赖文件
COPY requirements.txt .
# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 下载模型（可以在运行时下载，这里预下载加快启动）
RUN ollama pull llama3.1:8b
# 创建数据目录
RUN mkdir -p /app/data
# 暴露端口
EXPOSE 8000
# 启动脚本
COPY start.sh .
RUN chmod +x start.sh
CMD ["./start.sh"]
启动脚本：
#!/bin/bash
# start.sh
# 启动Ollama服务（后台运行）
ollama serve &
# 等待Ollama启动
sleep 10
# 启动FastAPI服务
uvicorn main:app \
--host 0.0.0.0 \
--port 8000 \
--workers 4 \
--log-level info
配置文件：
# config.yaml
system:
name: "智能客服系统"
version: "1.0.0"
debug: false
model:
name: "llama3.1:8b"
temperature: 0.1
max_tokens: 1024
timeout: 30
vectorstore:
type: "chroma"
persist_directory: "./data/chroma_db"
embedding_model: "BAAI/bge-small-zh-v1.5"
chunk_size: 800
chunk_overlap: 150
api:
host: "0.0.0.0"
port: 8000
workers: 4
cors_origins: ["*"]
rate_limit: 100 # 每分钟请求数限制
memory:
window_size: 10
persist_file: "./data/memory.json"
context_ttl: 86400 # 上下文保存时间（秒）
logging:
level: "INFO"
file: "./logs/app.log"
max_size: 10485760 # 10MB
backup_count: 5
monitoring:
enabled: true
prometheus_port: 9090
health_check_interval: 30
配置管理类：
import yaml
import os
from typing import Any, Dict
class Config:
def __init__(self, config_path="./config.yaml"):
self.config_path = config_path
self.config = self._load_config()
def _load_config(self) -> Dict[str, Any]:
"""加载配置文件"""
if not os.path.exists(self.config_path):
# 使用默认配置
return self._get_default_config()
with open(self.config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def _get_default_config(self) -> Dict[str, Any]:
"""获取默认配置"""
return {
"system": {
"name": "智能客服系统",
"version": "1.0.0",
"debug": False
},
"model": {
"name": "llama3.1:8b",
"temperature": 0.1,
"max_tokens": 1024,
"timeout": 30
},
"vectorstore": {
"type": "chroma",
"persist_directory": "./data/chroma_db",
"embedding_model": "BAAI/bge-small-zh-v1.5",
"chunk_size": 800,
"chunk_overlap": 150
},
"api": {
"host": "0.0.0.0",
"port": 8000,
"workers": 4,
"cors_origins": ["*"],
"rate_limit": 100
},
"memory": {
"window_size": 10,
"persist_file": "./data/memory.json",
"context_ttl": 86400
},
"logging": {
"level": "INFO",
"file": "./logs/app.log",
"max_size": 10485760,
"backup_count": 5
},
"monitoring": {
"enabled": True,
"prometheus_port": 9090,
"health_check_interval": 30
}
}
def get(self, key: str, default: Any = None) -> Any:
"""获取配置值"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def update(self, key: str, value: Any):
"""更新配置值"""
keys = key.split('.')
config = self.config
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
# 保存到文件
self._save_config()
def _save_config(self):
"""保存配置到文件"""
os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
with open(self.config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, allow_unicode=True, default_flow_style=False)
def reload(self):
"""重新加载配置"""
self.config = self._load_config()
5. 性能优化与监控
系统能跑起来只是第一步，要真正好用还需要优化和监控。这里分享几个我在实际项目中积累的经验。
5.1 性能优化策略
响应时间优化： 本地大模型最大的挑战就是响应时间。经过测试，我发现以下几个优化点效果最明显：
- 模型量化：使用GGUF格式的量化模型，可以在几乎不损失精度的情况下大幅减少内存占用和提升推理速度。
# 使用量化模型（以Qwen2.5为例）
ollama pull qwen2.5:7b-q4_K_M # 4位量化，平衡精度和速度
# 或者
ollama pull llama3.1:8b-q4_K_S # 更小的量化，速度更快
- 批处理请求：如果有多个相似问题，可以批量处理。
from typing import List
import asyncio
class BatchProcessor:
def __init__(self, chatbot, batch_size=4):
self.chatbot = chatbot
self.batch_size = batch_size
async def process_batch(self, questions: List[str]) -> List[str]:
"""批量处理问题"""
results = []
# 分批处理
for i in range(0, len(questions), self.batch_size):
batch = questions[i:i + self.batch_size]
batch_results = await self._process_single_batch(batch)
results.extend(batch_results)
# 避免过热
if i + self.batch_size < len(questions):
await asyncio.sleep(0.1)
return results
async def _process_single_batch(self, questions: List[str]) -> List[str]:
"""处理单个批次"""
# 这里可以并行处理，但要注意Ollama的并发限制
tasks = []
for question in questions:
task = asyncio.create_task(
self._ask_question(question)
)
tasks.append(task)
return await asyncio.gather(*tasks)
async def _ask_question(self, question: str) -> str:
"""异步提问"""
# 注意：OllamaLLM可能不支持原生异步
# 实际使用时可能需要线程池
import threading
result = []
def sync_ask():
response = self.chatbot.ask_with_context(
user_id="batch_user",
question=question
)
result.append(response["answer"])
thread = threading.Thread(target=sync_ask)
thread.start()
thread.join()
return result[0] if result else "处理失败"
- 缓存机制：常见问题可以缓存答案，避免重复计算。
import hashlib
import pickle
from datetime import datetime, timedelta
class ResponseCache:
def __init__(self, cache_dir="./cache", ttl=3600):
self.cache_dir = cache_dir
self.ttl = ttl # 缓存有效期（秒）
os.makedirs(cache_dir, exist_ok=True)
def _get_cache_key(self, question: str, context_hash: str = "") -> str:
"""生成缓存键"""
content = question + context_hash
return hashlib.md5(content.encode()).hexdigest()
def _get_cache_path(self, key: str) -> str:
"""获取缓存文件路径"""
return os.path.join(self.cache_dir, f"{key}.pkl")
def get(self, question: str, context_hash: str = ""):
"""获取缓存响应"""
key = self._get_cache_key(question, context_hash)
cache_path = self._get_cache_path(key)
if os.path.exists(cache_path):
try:
with open(cache_path, 'rb') as f:
data = pickle.load(f)
# 检查是否过期
if datetime.now() - data["timestamp"] < timedelta(seconds=self.ttl):
return data["response"]
else:
# 过期删除
os.remove(cache_path)
except:
pass
return None
def set(self, question: str, response: dict, context_hash: str = ""):
"""设置缓存"""
key = self._get_cache_key(question, context_hash)
cache_path = self._get_cache_path(key)
data = {
"response": response,
"timestamp": datetime.now(),
"question": question,
"context_hash": context_hash
}
try:
with open(cache_path, 'wb') as f:
pickle.dump(data, f)
except:
pass
def clear_expired(self):
"""清理过期缓存"""
now = datetime.now()
for filename in os.listdir(self.cache_dir):
if filename.endswith('.pkl'):
path = os.path.join(self.cache_dir, filename)
try:
with open(path, 'rb') as f:
data = pickle.load(f)
if now - data["timestamp"] > timedelta(seconds=self.ttl):
os.remove(path)
except:
os.remove(path) # 损坏的文件也删除
内存优化： 大模型很吃内存，特别是同时服务多个用户时。我的优化策略：
- 动态加载模型：根据请求量动态调整模型加载。
class ModelManager:
def __init__(self, model_names):
self.models = {}
self.active_model = None
self.load_threshold = 5 # 每分钟请求数阈值
self.request_count = 0
self.last_reset = datetime.now()
def get_model(self, request_rate):
"""根据请求率返回合适的模型"""
# 重置计数器
if (datetime.now() - self.last_reset).seconds >= 60:
self.request_count = 0
self.last_reset = datetime.now()
self.request_count += 1
# 根据请求率选择模型
if request_rate < 2: # 低负载
model_name = "llama3.1:8b"
elif request_rate < 10: # 中等负载
model_name = "qwen2.5:7b" # 更快的模型
else: # 高负载
model_name = "mistral:7b" # 最快但能力稍弱
# 加载模型（如果未加载）
if model_name not in self.models:
self._load_model(model_name)
self.active_model = model_name
return self.models[model_name]
def _load_model(self, model_name):
"""加载模型"""
print(f"正在加载模型: {model_name}")
# 实际应该调用Ollama加载模型
# 这里简化处理
self.models[model_name] = {"name": model_name, "loaded": True}
def unload_unused_models(self):
"""卸载不用的模型释放内存"""
if len(self.models) > 1 and self.active_model:
for name in list(self.models.keys()):
if name != self.active_model:
print(f"卸载模型: {name}")
# 实际应该调用Ollama卸载模型
del self.models[name]
- 响应流式输出：对于长回答，可以边生成边返回。
from fastapi.responses import StreamingResponse
import json
@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
"""流式聊天接口"""
async def generate():
# 模拟流式输出
question = request.question
words = question.split()
for i in range(len(words) + 10): # 模拟生成过程
if i < len(words):
chunk = words[i] + " "
else:
chunk = f"[思考{i-len(words)}] "
yield f"data: {json.dumps({'chunk': chunk})}\n\n"
await asyncio.sleep(0.1) # 模拟生成延迟
yield f"data: {json.dumps({'done': True})}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # 禁用Nginx缓冲
}
)
5.2 监控与日志
生产环境必须要有完善的监控。我通常从几个维度来监控系统：
性能监控：
import time
from collections import deque
from dataclasses import dataclass
from typing import Dict, List
import psutil
import GPUtil
@dataclass
class SystemMetrics:
timestamp: float
cpu_percent: float
memory_percent: float
gpu_memory_percent: float
request_count: int
avg_response_time: float
error_rate: float
class Monitor:
def __init__(self, window_size=100):
self.metrics_history = deque(maxlen=window_size)
self.request_times = deque(maxlen=1000)
self.error_count = 0
self.total_requests = 0
def record_request(self, processing_time: float, success: bool = True):
"""记录请求指标"""
self.request_times.append(processing_time)
self.total_requests += 1
if not success:
self.error_count += 1
def collect_metrics(self) -> SystemMetrics:
"""收集系统指标"""
# CPU使用率
cpu_percent = psutil.cpu_percent(interval=0.1)
# 内存使用率
memory = psutil.virtual_memory()
memory_percent = memory.percent
# GPU使用率（如果有）
gpu_memory_percent = 0
try:
gpus = GPUtil.getGPUs()
if gpus:
gpu_memory_percent = gpus[0].memoryUtil * 100
except:
pass
# 计算平均响应时间
avg_response_time = 0
if self.request_times:
avg_response_time = sum(self.request_times) / len(self.request_times)
# 计算错误率
error_rate = 0
if self.total_requests > 0:
error_rate = self.error_count / self.total_requests * 100
metrics = SystemMetrics(
timestamp=time.time(),
cpu_percent=cpu_percent,
memory_percent=memory_percent,
gpu_memory_percent=gpu_memory_percent,
request_count=self.total_requests,
avg_response_time=avg_response_time,
error_rate=error_rate
)
self.metrics_history.append(metrics)
return metrics
def get_alerts(self) -> List[Dict]:
"""检查是否需要告警"""
alerts = []
if not self.metrics_history:
return alerts
latest = self.metrics_history[-1]
# CPU告警
if latest.cpu_percent > 80:
alerts.append({
"level": "warning",
"metric": "cpu",
"value": latest.cpu_percent,
"message": f"CPU使用率过高: {latest.cpu_percent:.1f}%"
})
# 内存告警
if latest.memory_percent > 85:
alerts.append({
"level": "critical",
"metric": "memory",
"value": latest.memory_percent,
"message": f"内存使用率过高: {latest.memory_percent:.1f}%"
})
# 响应时间告警
if latest.avg_response_time > 5.0: # 超过5秒
alerts.append({
"level": "warning",
"metric": "response_time",
"value": latest.avg_response_time,
"message": f"平均响应时间过长: {latest.avg_response_time:.2f}秒"
})
# 错误率告警
if latest.error_rate > 5.0: # 超过5%
alerts.append({
"level": "critical",
"metric": "error_rate",
"value": latest.error_rate,
"message": f"错误率过高: {latest.error_rate:.1f}%"
})
return alerts
def get_performance_report(self) -> Dict:
"""生成性能报告"""
if not self.metrics_history:
return {}
metrics_list = list(self.metrics_history)
return {
"current": {
"cpu_percent": metrics_list[-1].cpu_percent,
"memory_percent": metrics_list[-1].memory_percent,
"gpu_memory_percent": metrics_list[-1].gpu_memory_percent,
"avg_response_time": metrics_list[-1].avg_response_time,
"error_rate": metrics_list[-1].error_rate
},
"stats": {
"total_requests": self.total_requests,
"error_count": self.error_count,
"p95_response_time": self._calculate_percentile(95),
"p99_response_time": self._calculate_percentile(99),
"max_response_time": max(self.request_times) if self.request_times else 0
},
"alerts": self.get_alerts()
}
def _calculate_percentile(self, percentile: float) -> float:
"""计算百分位数"""
if not self.request_times:
return 0
sorted_times = sorted(self.request_times)
index = (percentile / 100) * (len(sorted_times) - 1)
if index.is_integer():
return sorted_times[int(index)]
else:
lower = sorted_times[int(index)]
upper = sorted_times[int(index) + 1]
return lower + (upper - lower) * (index - int(index))
日志配置：
import logging
import logging.handlers
from pathlib import Path
def setup_logging(config):
"""配置日志系统"""
log_config = config.get("logging", {})
log_level = getattr(logging, log_config.get("level", "INFO"))
log_file = log_config.get("file", "./logs/app.log")
# 创建日志目录
log_path = Path(log_file).parent
log_path.mkdir(parents=True, exist_ok=True)
# 配置根日志
logger = logging.getLogger()
logger.setLevel(log_level)
# 清除现有处理器
logger.handlers.clear()
# 控制台处理器
console_handler = logging.StreamHandler()
console_format = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
console_handler.setFormatter(console_format)
logger.addHandler(console_handler)
# 文件处理器（按大小轮转）
file_handler = logging.handlers.RotatingFileHandler(
filename=log_file,
maxBytes=log_config.get("max_size", 10485760), # 10MB
backupCount=log_config.get("backup_count", 5),
encoding='utf-8'
)
file_format = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
)
file_handler.setFormatter(file_format)
logger.addHandler(file_handler)
# 特定模块的日志级别
logging.getLogger("uvicorn").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)
return logger
# 使用示例
logger = setup_logging(config)
# 记录不同级别的日志
logger.debug("调试信息")
logger.info("系统启动完成")
logger.warning("内存使用率超过80%")
logger.error("数据库连接失败")
5.3 测试与验证
系统上线前一定要充分测试。我通常从几个维度进行测试：
功能测试：
import unittest
from unittest.mock import Mock, patch
import sys
import os
sys.path.append(os.path.dirname(os更多推荐
5
0- 0
已为社区贡献11条内容
所有评论(0)