粤ICP备17114055号
微信扫码
添加专属顾问
支持私有化部署
我要投稿
运维的智能革命，让知识触手可及。 核心内容： 1. 四大核心模块构建的智能知识库系统 2. 技术亮点：文档处理与语义搜索的智慧 3. 真实场景下运维效率的飞跃提升
凌晨3点，某公司服务器突然宕机！值班运维小哥小王急得满头大汗，面对上千份文档，他却在10秒内找到了解决方案——这不是科幻电影，而是一个真实运维知识库系统的日常。
今天，我们就来揭秘这个让运维人告别“熬夜翻文档”的神器！用技术解决知识的“最后一公里”问题。
（运维老张：“以前找文档像大海捞针，现在像用饿了么点外卖——精准送达！”）
实习生小李面对“数据库主从同步”一脸懵，系统直接甩出操作手册+原理图解，附赠一句：“亲，同步前记得备份哦~”
某电商大促期间服务器崩了，系统秒回：“《高并发应急预案》第4.2节+自动扩容脚本”，运维团队30分钟恢复业务！
老师傅的“祖传笔记”上传系统后，新员工也能随时调用——从此告别“人走经验凉”的尴尬。
#python版本
python --version
Python 3.12.9
#pip依赖包
pip install flask flask-httpauth werkzeug elasticsearch langchain-elasticsearch langchain-community python-docx pdfplumber pandas tika langchain-nomic requests urllib3
#es本地docker启动
docker run -d --name elasticsearch -p 9200:9200 -p 9300:9300 \
-e "discovery.type=single-node" \
-e "xpack.security.enabled=false" \
elasticsearch:7.17.10
#创建索引knowledge_index
curl -X PUT "http://localhost:9200/knowledge_index" -H 'Content-Type: application/json' -d'
{
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0
},
"mappings": {
"properties": {
"title": {
"type": "text"
},
"content": {
"type": "text"
},
"timestamp": {
"type": "date"
}
}
}
}
'
# 配置文件
# Elasticsearch 的连接 URL
ES_URL = "http://192.168.31.136:9200"
# Elasticsearch 中存储知识的索引名称
INDEX_NAME = "knowledge_index"
此文件负责数据库的初始化、数据添加和查询操作，使用 SQLite 作为数据库。
import sqlite3
# 初始化数据库，创建历史记录表格
def init_db():
# 连接到 SQLite 数据库
conn = sqlite3.connect('history.db')
# 创建游标对象，用于执行 SQL 语句
c = conn.cursor()
# 创建历史记录表格，如果表格不存在
c.execute('''CREATE TABLE IF NOT EXISTS history
(id INTEGER PRIMARY KEY AUTOINCREMENT,
user TEXT,
question TEXT,
answer TEXT)''')
# 提交数据库事务
conn.commit()
# 关闭数据库连接
conn.close()
# 向历史记录表格中添加用户的问答记录
def add_history(user, question, answer):
# 连接到 SQLite 数据库
conn = sqlite3.connect('history.db')
# 创建游标对象
c = conn.cursor()
# 插入用户的问答记录到历史记录表格
c.execute("INSERT INTO history (user, question, answer) VALUES (?,?,?)", (user, question, answer))
# 提交数据库事务
conn.commit()
# 关闭数据库连接
conn.close()
# 获取指定用户的历史问答记录
def get_history(user):
# 连接到 SQLite 数据库
conn = sqlite3.connect('history.db')
# 创建游标对象
c = conn.cursor()
# 查询指定用户的历史问答记录
c.execute("SELECT question, answer FROM history WHERE user =?", (user,))
# 获取查询结果
history = c.fetchall()
# 关闭数据库连接
conn.close()
return history
负责文档的加载、处理和向 Elasticsearch 中添加向量数据。
# 导入所需的库
from langchain_community.document_loaders import TextLoader, Docx2txtLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_nomic.embeddings import NomicEmbeddings
from langchain_elasticsearch import ElasticsearchStore
from langchain.document_loaders import PyPDFLoader
import re
from config import ES_URL, INDEX_NAME
# 清理文本，去除多余的空格
def clean_text(text):
text = re.sub(r'\s+', ' ', text).strip()
return text
# 加载并处理文档，将文档分割成小块
def load_and_process_documents(file_path):
if file_path.endswith('.txt'):
# 使用 TextLoader 加载文本文件
loader = TextLoader(file_path)
elif file_path.endswith('.pdf'):
# 使用 PyPDFLoader 加载 PDF 文件
loader = PyPDFLoader(file_path)
elif file_path.endswith('.docx'):
# 使用 Docx2txtLoader 加载 DOCX 文件
loader = Docx2txtLoader(file_path)
else:
# 如果文件格式不支持，抛出异常
raise ValueError("不支持的文件格式")
# 加载文档
documents = loader.load()
# 创建文本分割器，设置每个块的大小和重叠部分
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
# 分割文档
docs = text_splitter.split_documents(documents)
# 清理每个文档块的文本
cleaned_docs = [clean_text(doc.page_content) for doc in docs]
return cleaned_docs
# 将处理后的文档添加到 Elasticsearch 向量存储中
def add_to_vectorstore(file_path):
# 加载并处理文档
docs = load_and_process_documents(file_path)
# 创建 NomicEmbeddings 对象，用于生成文档的嵌入向量
embeddings = NomicEmbeddings()
# 将文档添加到 Elasticsearch 向量存储中
ElasticsearchStore.from_texts(
texts=docs,
embedding=embeddings,
es_url=ES_URL,
index_name=INDEX_NAME
)
实现问题回答的核心逻辑。
import requests
import random
from langchain_nomic.embeddings import NomicEmbeddings
from langchain_elasticsearch.vectorstores import ElasticsearchStore
from config import ES_URL, INDEX_NAME
import os
import json
# 随机选择一个 DeepSeek 节点
def select_node():
return random.choice(DEEPSEEK_NODES)
# 处理用户的问题，返回答案
def ask_question(question):
try:
# 从环境变量中获取 Nomic API 密钥
nomic_api_key = os.getenv('NOMIC_API_KEY')
# 初始化 NomicEmbeddings
embeddings = NomicEmbeddings(model="nomic-embed-text-v1.5", nomic_api_key=nomic_api_key)
# 创建 Elasticsearch 向量存储对象
vectorstore = ElasticsearchStore(embedding=embeddings, es_url=ES_URL, index_name=INDEX_NAME)
# 从向量存储中搜索与问题相关的文档
query = vectorstore.similarity_search(question, k=3)
print(json.dumps(query, indent=2))
# 再次搜索与问题相关的文档
relevant_docs = vectorstore.similarity_search(question, k=3)
# 将相关文档的内容拼接成上下文
context = "\n".join([doc.page_content for doc in relevant_docs])
# 选择一个 DeepSeek 节点
node_url = select_node()
# 构建请求体
payload = {
"model": "deepseek",
"prompt": f"根据以下文档内容回答问题：{context}\n问题：{question}"
}
# 发送 POST 请求到 DeepSeek 节点
response = requests.post(node_url, json=payload, timeout=10)
# 检查响应状态码，如果不是 200，抛出异常
response.raise_for_status()
# 解析响应的 JSON 数据
result = response.json()
# 返回答案，如果没有找到答案，返回默认信息
return result.get("answer", "未找到合适答案")
except requests.RequestException as e:
# 处理请求异常
print(f"与 DeepSeek 模型通信时出错: {e}")
return"请求出错，请稍后重试"
except Exception as e:
# 处理其他异常
print(f"处理问题时出现未知错误: {e}")
return"出现未知错误，请联系管理员"
Flask 应用的主文件，处理用户的请求。
import os
import logging
import requests
from elasticsearch import Elasticsearch
from flask import Flask, request, jsonify, render_template, redirect, url_for
from werkzeug.security import generate_password_hash, check_password_hash
from flask_httpauth import HTTPBasicAuth
import re
import urllib3
import pandas as pd
import docx
import pdfplumber
import tempfile
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import json
from werkzeug.utils import secure_filename
from database import init_db, add_history, get_history
from question_answering import ask_question
from document_processor import add_to_vectorstore
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 忽略 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 用户认证管理类
class UserAuth:
def __init__(self):
# 存储用户的用户名和加密后的密码
self.users = {
"admin": generate_password_hash("adminpassword"),
"user": generate_password_hash("userpassword")
}
# 创建 HTTP 基本认证对象
self.auth = HTTPBasicAuth()
@self.auth.verify_password
def verify_password(username, password):
# 验证用户的用户名和密码
if username in self.users and check_password_hash(self.users.get(username), password):
return username
# Elasticsearch 管理类
class ElasticsearchManager:
def __init__(self):
# 连接到 Elasticsearch
self.es = Elasticsearch([{'scheme': 'http', 'host': 'localhost', 'port': 9200}])
# 定义 Elasticsearch 索引名称
self.index_name = 'ops_knowledge'
# 初始化索引
self.init_index()
def init_index(self):
try:
# 检查索引是否存在，如果不存在则创建
ifnot self.es.indices.exists(index=self.index_name):
self.es.indices.create(index=self.index_name)
logger.info(f"索引 '{self.index_name}' 创建成功")
except Exception as e:
# 记录创建索引时的错误信息
logger.error(f"创建索引时出错: {e}")
raise
def add_document(self, title, content, file_path):
# 构建文档对象
doc = {
'title': title,
'content': content,
'file_path': file_path
}
try:
# 将文档添加到 Elasticsearch 索引中
res = self.es.index(index=self.index_name, body=doc)
logger.info(f"文档 {title} 已添加到知识库，ID: {res['_id']}")
return res['result']
except Exception as e:
# 记录添加文档时的错误信息
logger.error(f"添加文档 {title} 到知识库时出错: {e}")
returnNone
def search_knowledge(self, query):
try:
# 构建搜索请求体
body = {
"query": {
"multi_match": {
"query": query,
"fields": ["title", "content"]
}
}
}
# 执行搜索请求
res = self.es.search(index=self.index_name, body=body)
# 获取搜索结果
hits = res['hits']['hits']
results = []
for hit in hits:
# 提取搜索结果的相关信息
results.append({
'id': hit['_id'],
'title': hit['_source']['title'],
'content': hit['_source']['content'],
'file_path': hit['_source']['file_path']
})
return results
except Exception as e:
# 记录搜索知识时的错误信息
logger.error(f"搜索知识时出错: {e}")
return []
# 文档解析类
class DocumentParser:
@staticmethod
def parse_document(file_path):
try:
if file_path.endswith('.docx'):
# 解析 DOCX 文件
doc = docx.Document(file_path)
content = ""
for para in doc.paragraphs:
content += para.text + "\n"
return content.strip()
elif file_path.endswith('.xlsx'):
# 解析 XLSX 文件
df = pd.read_excel(file_path)
content = ""
for _, row in df.iterrows():
content += str(row.values) + "\n"
return content.strip()
elif file_path.endswith('.pdf'):
# 解析 PDF 文件
with pdfplumber.open(file_path) as pdf:
content = ""
for page in pdf.pages:
content += page.extract_text() + "\n"
return content.strip()
else:
# 使用 Tika 解析其他文件格式
from tika import parser
parsed = parser.from_file(file_path, requestOptions={"verify": False})
content = parsed['content']
if content:
content = content.strip()
return content
except Exception as e:
# 记录解析文件时的错误信息
logger.error(f"解析文件 {file_path} 时出错: {e}")
returnNone
# dify 管理类
class DifyManager:
def __init__(self, api_key):
# 存储 Dify API 密钥
self.api_key = api_key
# 设置请求头
self.headers = {"Authorization": f"Bearer {self.api_key}"}
# Dify 文档创建 API URL
self.api_url = "https://api.dify.ai/v1/datasets/810ff8fb-fd85-4b5b-a4cxxx_xcxxc_b473c/document/create-by-file"
# 创建请求会话
self.session = requests.Session()
# 添加重试机制
retries = Retry(total=3, backoff_factor=1)
self.session.mount('https://', HTTPAdapter(max_retries=retries))
def add_to_dify(self, file_path):
try:
# 打开文件
with open(file_path, 'rb') as file_obj:
# 构建文件上传请求体
files = {
'file': (secure_filename(os.path.basename(file_path)), file_obj, self._get_mime_type(file_path))
}
# 构建数据请求体
data = {
'indexing_technique': 'high_quality',
'process_rule': json.dumps({"mode": "automatic"}) # 确保 JSON 序列化
}
# 发送 POST 请求到 Dify API
response = self.session.post(
self.api_url,
headers=self.headers,
files=files,
data=data,
timeout=30
)
# 检查响应状态码，如果不是 200，抛出异常
response.raise_for_status()
return response.json()
except requests.RequestException as e:
# 处理请求异常
response = getattr(e, 'response', None)
error_msg = {
'error': str(e),
'status_code': response.status_code if response elseNone,
'response_text': response.text[:200] if response elseNone
}
logger.error(f"Dify上传失败详情: {json.dumps(error_msg, indent=2)}")
returnNone
def _get_mime_type(self, filename):
# 根据文件扩展名返回 MIME 类型
ext = os.path.splitext(filename)[1].lower()
return {
'.pdf': 'application/pdf',
'.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
}.get(ext, 'application/octet-stream')
# DeepSeek 管理类
class DeepSeekManager:
def __init__(self, api_key):
# 存储 DeepSeek API 密钥
self.api_key = api_key
# 设置请求头
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
def intelligent_qa(self, question):
# 清理问题中的特殊字符
question = re.sub(r'[^\w\s]', '', question)
# 构建请求体
data = {
"model": "deepseek-chat",
"messages": [
{"role": "user", "content": question}
],
"temperature": 0.7,
"max_tokens": 1024,
"top_p": 0.9# 新增可能的参数，根据实际情况调整
}
try:
# 发送 POST 请求到 DeepSeek API
response = requests.post(
"https://api.deepseek.com/chat/completions",
headers=self.headers,
json=data
)
# 检查响应状态码，如果不是 200，抛出异常
response.raise_for_status()
# 解析响应的 JSON 数据
result = response.json()
# 返回答案，如果没有找到答案，返回默认信息
return result.get("choices", [{}])[0].get("message", {}).get("content", "未找到相关答案，请补充更多信息。"
return answer
# Flask应用
app = Flask(__name__)
user_auth = UserAuth()
knowledge_base = OpsKnowledgeBase("dataset-V0TC4VIGfpfvmyrxxxV3hxxx", "sk-fc5c4a54ebfxxxxxxxc2cb6xxx25xx")
# 首页
@app.route('/')
@user_auth.auth.login_required
def index():
return render_template('index.html')
# 扫描文件夹并添加文档
@app.route('/scan_folder', methods=['POST'])
@user_auth.auth.login_required
def scan_folder():
if user_auth.auth.current_user() == 'admin':
folder_path = request.form.get('folder_path')
if folder_path:
knowledge_base.scan_folder_and_add_documents(folder_path)
return jsonify({'message': '文件夹扫描并添加文档完成'})
return jsonify({'error': '缺少文件夹路径参数'}), 400
return jsonify({'error': '只有管理员可以执行此操作'}), 403
# 搜索知识
@app.route('/search', methods=['GET'])
@user_auth.auth.login_required
def search():
query = request.args.get('query')
if query:
results = knowledge_base.search_knowledge(query)
return render_template('search_results.html', results=results)
return jsonify({'error': '缺少查询参数'}), 400
# 智能问答
@app.route('/qa', methods=['GET'])
@user_auth.auth.login_required
def qa():
question = request.args.get('question')
if question:
answer = knowledge_base.intelligent_qa(question)
return render_template('qa_result.html', question=question, answer=answer)
return jsonify({'error': '缺少问题参数'}), 400
if __name__ == '__main__':
app.run(debug=True)
在项目根目录下创建一个templates文件夹，里面包含index.html、login.html、qa_result.html、search_results.html四个文件。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>运维知识库</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
}
h1 {
color: #333;
}
form {
margin-bottom: 20px;
}
input[type="text"] {
padding: 8px;
width: 300px;
margin-right: 10px;
}
button {
padding: 8px 15px;
background-color: #007BFF;
color: white;
border: none;
cursor: pointer;
}
button:hover {
background-color: #0056b3;
}
</style>
</head>
<body>
<h1>运维知识库</h1>
<h2>扫描文件夹添加文档</h2>
<form action="/scan_folder" method="post">
<input type="text" name="folder_path" placeholder="输入文件夹路径">
<button type="submit">扫描并添加</button>
</form>
<h2>搜索知识</h2>
<form action="/search" method="get">
<input type="text" name="query" placeholder="输入搜索关键词">
<button type="submit">搜索</button>
</form>
<h2>智能问答</h2>
<form action="/qa" method="get">
<input type="text" name="question" placeholder="输入你的问题">
<button type="submit">提问</button>
</form>
</body>
</html>
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>登录</title>
</head>
<body>
<h1>登录</h1>
{% if error %}
<p style="color: red;">{{ error }}</p>
{% endif %}
<form action="{{ url_for('login') }}" method="post">
<label for="username">用户名:</label>
<input type="text" id="username" name="username" required><br>
<label for="password">密码:</label>
<input type="password" id="password" name="password" required><br>
<input type="submit" value="登录">
</form>
</body>
</html>
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>智能问答结果</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
}
h1 {
color: #333;
}
.question {
font-weight: bold;
}
.answer {
margin-top: 10px;
}
</style>
</head>
<body>
<h1>智能问答结果</h1>
<p class="question">问题: {{ question }}</p>
<p class="answer">答案: {{ answer }}</p>
<a href="/">返回首页</a>
</body>
</html>
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>搜索结果</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
}
h1 {
color: #333;
}
.result {
border: 1px solid #ccc;
padding: 10px;
margin-bottom: 10px;
}
.result h2 {
margin-top: 0;
}
</style>
</head>
<body>
<h1>搜索结果</h1>
{% if results %}
{% for result in results %}
<div class="result">
<h2>{{ result.title }}</h2>
<p>{{ result.content }}</p>
<p>文件路径: {{ result.file_path }}</p>
</div>
{% endfor %}
{% else %}
<p>未找到相关结果。</p>
{% endif %}
<a href="/">返回首页</a>
</body>
</html>
这个系统的本质，不是取代运维人员，而是把人类从重复劳动中解放出来——让工程师专注创新，让知识流动起来，让每一次深夜救急都变成“有备而战”。 正如Linux创始人Linus Torvalds所说：“技术之美，在于让复杂的事情变简单。”当每一个运维问题都能在10秒内找到答案时，我们便离“零故障焦虑”的运维乌托邦更近了一步。
也正如另外一位用户所说：“以前我觉得AI很遥远，直到它帮我保住了年终奖……”
代码仅做展示，非最终应用，革命尚未成功，同志仍需努力。
53AI，企业落地大模型首选服务商
产品：场景落地咨询+大模型应用平台+行业解决方案
承诺：免费场景POC验证，效果验证后签署服务协议。零风险落地应用大模型，已交付160+中大型企业
2025-04-30
日本写给小学生的生成式AI应用指南
2025-04-29
Flomo MCP：用自然语言对话打造你的第二大脑，效率提升100%的秘诀在这里！
2025-04-29
用「飞书知识问答」打造你的「超级AI教师助理」
2025-04-28
智能运维新时代：如何打造你的专属知识库
2025-04-28
在人工智能时代，需要重新审视企业显性知识、隐性知识以及“内隐”知识的管理范式
2025-04-27
用AI重构知识管理：重塑Obsidian工作流的三层进阶指南
2025-04-27
人人用AI ≠ 组织AI化，什么是智能组织？
2025-04-26
2025年，建议你一定要用 AI 搭建个人知识库
2024-09-14
2025-01-23
2024-07-10
2024-11-07
2025-02-17
2024-04-24
2024-08-04
2024-06-23
2024-05-15
2024-07-10
2025-04-27
2025-04-20
2025-04-17
2025-04-17
2025-04-13
2025-04-12
2025-04-11
2025-04-06
回到顶部