一、投石問路
相信大家有沒有過這樣的經歷，急需某個操作流程，比如要找“給客户開發票”的相關信息，結果發現自己要在電腦裏翻箱倒櫃，從一堆命名混亂的Word、Excel裏尋找那份不知道有沒有存檔、也不知道更新沒更新的“開票説明.docx”。找到了，還得像做閲讀理解一樣，從十幾頁的文字裏摳出自己需要的那幾條信息。這個我們習以為常、普通的不能再普通的過程，時時刻刻都在我們的工作中經歷着，這種不僅效率低、體驗差，而且極易出錯。
想象一下，如果有一天，我們再需要什麼內容時，只需像現在的deepseek搜索引擎一樣，自然地輸入一句：“請問給客户開專票需要準備什麼？”一秒之內，一個清晰、準確、最新的答案列表就直接彈到你眼前。如今的AI時代已經達到了這種效果，然而企業發展過程中，有很多散落各處的本地文檔，日積月累的形成了企業的文檔庫，如果我們能整合這些手冊、文檔，形成一個真正意義上的本地知識庫，可以快速、方便的對這些知識手冊進行檢索查閲將是一件更加大快人心的事情，今天，時代精進，人工智能飛速發展，我們可以實現這件大快人心的事情了，接下來就讓我們一起聊聊，知識庫如何從一個靜態的檔案櫃，成長為我們身邊最得力的智慧大腦，將企業的核心檔案升級為一個會思考、能對話的智能數字知識庫。
二、知識庫的定義與重要性
知識庫是一個結構化的、經過組織的、可用於智能檢索和推理的知識集合，它並非一個簡單的文件倉庫或問答列表，它旨在捕獲、整理和存儲企業內部的顯性知識如規章制度、產品手冊、流程指南、技術文檔、常見問題解答等和部分隱性知識，如專家經驗、解決方案案例，並提供一個集中化的平台供員工檢索和使用。
在當今信息爆炸的時代，知識庫已成為企業不可或缺的核心戰略資產，日積月累已經形成了運營管理過程中不可或缺的部分：
企業賦能提升員工效率：
- 統一信息出口：確保員工獲取的信息是唯一、最新、準確的版本，避免因信息源混亂導致的決策失誤和操作錯誤。
- 自助式解決問題：員工可以隨時、隨地快速找到解決方案，減少對專家同事的重複性打擾，實現自助服務，極大提升個人工作效率和體驗。
- 加速新人成長：新員工通過知識庫可以快速學習業務流程、產品知識和公司制度，顯著縮短培訓週期和上崗時間。
企業運營降本增效：
- 降低運營成本：減少客服、技術支持等崗位的重複性諮詢壓力，讓人力資源聚焦於更復雜、高價值的問題。
- 沉澱企業智慧：將分散在個人大腦、郵件、聊天記錄中的知識系統化地沉澱下來，避免人才流失即知識流失的風險，形成企業的數字資產。
- 保障合規與標準：確保業務流程（如財務報銷、合規操作）的執行有據可依、標準統一，降低操作風險。
智能客服提升服務：
- 一個強大的知識庫是智能客服機器人的大腦，它能提供7x24小時即時、準確的自動應答，大幅提升客户滿意度和忠誠度。
三、傳統知識庫的變革
儘管知識庫理念先進，但傳統基於關鍵詞匹配和手動維護的知識庫系統在實踐中飽受詬病，存在很多痛點：
- 關鍵詞匹配的侷限：檢索不準，形成答非所問的困境，傳統搜索依賴純粹的關鍵詞匹配，如數據庫LIKE查詢或簡單搜索引擎，它無法理解語義。
- 完全依賴人工：知識的錄入、分類、更新、過期清理全部需要領域專家手動完成，是一項極其繁瑣、耗時的工作，導致心有餘而力不足。
- 更新滯後：業務流程一變，所有相關文檔都需要人工找出並修改，稍有遺漏就會導致知識庫內容過期，讓人不敢信。
- 投入產出比低：高昂的維護成本使得許多企業的知識庫項目最終淪為無人問津、內容陳舊的數字廢墟。
- 被動而非主動：知識庫是一個需要被維護知識的被動系統。它無法自動從企業日常運轉的如客服對話記錄、項目覆盤文檔、會議紀要中挖掘和提煉知識。
- 形成孤島：知識散落在各處，無法有效流入知識庫，使得知識庫與現實脱節，價值無法最大化。
近年來，大語言模型的和向量數據庫帶來的變革，相關技術的成熟與應用，兩者的協同效應為打造新一代智能知識庫提供了革命性的解決方案，兩者結合構成了現代智能知識庫的雙引擎。向量數據庫充當海馬體，負責快速、精準地記憶和召回知識片段；大模型則充當大腦皮層，負責理解、推理和組織語言，給出最終答案。二者的結合，使得知識庫終於從一個“靜態檔案櫃”進化成了一個能夠“聽懂人話、對答如流”的企業智慧大腦，真正成為了企業智能化的核心基石。
四、知識庫問題生成與檢索優化
1. 問題生成：解決語義鴻溝的理論基礎
傳統知識庫只有標準問題和答案，用户提問的方式和知識庫中標準問題的表述之間存在着巨大的語義鴻溝。為了解決這一問題，自動化問題生成成為關鍵。利用Qwen大模型強大的文本理解與生成能力，我們可以從已有的“答案”反向生成多種可能的用户問題，極大地擴充知識庫的入口，提高命中率。
- 理論依據：基於答案感知的問題生成。大語言模型（如Qwen）通過理解答案的語義內容，學習生成與之對應的、多種表述形式的問題。
- 生成策略：
- 同義改寫：生成與標準問題語義相同但表述不同的問題。（例如：“如何申請年假？” -> “年假的申請流程是怎樣的？”）
- 視角轉換：從不同角度提問。（例如：從員工角度“我怎麼請假？”到HR角度“請假制度是什麼？”）
- 難度分級：生成簡單、中等、複雜不同層次的問題，以匹配不同用户的認知水平。
- 價值：極大地擴展了知識庫的入口，提升了召回率，確保用户各種千奇百怪的問法都能被正確匹配，導向正確的答案。
2. 檢索優化：從關鍵詞到語義理解
2.1 方法介紹
檢索是知識庫的大門，其核心是從海量知識中快速、準確地找到最相關的內容。
傳統方法：稀疏檢索（如BM25）
- 理論：基於詞頻統計（TF）和逆文檔頻率（IDF）。BM25通過計算查詢詞條與文檔的加權匹配度來評分，擅長處理詞彙匹配。
- 優點：速度快、可解釋性強、無需訓練數據。
- 缺點：無法解決詞彙不匹配問題，即無法理解“電腦”和“計算機”是同一個意思。
現代方法：密集檢索與混合檢索
- 密集檢索理論：使用嵌入模型（如Sentence-BERT）將文本映射到高維向量空間。語義相似的文本，其向量在空間中的距離也更近。通過向量數據庫（如Faiss） 進行高效的最近鄰搜索（ANN）。
- 優點：具備語義匹配 能力，克服了詞彙不匹配的難題。
- 混合檢索理論：結合稀疏檢索（BM25）和密集檢索的優點。分別從兩個索引中檢索，然後使用重新排序（Re-ranking） 模型（如Cross-Encoder）對合並的結果進行精細打分，最終返回最優結果。這是一種“寬搜 + 精排”的策略，在保證召回率的同時極大提升了準確率。
- RAG架構：將檢索到的最相關知識片段作為上下文，與大語言模型的強大生成能力相結合，生成精準、可信且可追溯的答案，有效解決了大模型的幻覺問題。
2.2 案例演示
代碼結構
"""
知識庫問題生成與檢索優化系統
基於BM25算法和Qwen大模型，實現知識庫的智能化處理
主要功能：
1. 為知識內容自動生成多樣化問題
2. 使用BM25算法實現高效檢索
3. 比較原文檢索與問題檢索的性能差異
4. 支持多種知識主題的適配
"""
# 導入依賴庫
import os # 操作系統接口，用於環境變量和文件操作
import json # JSON格式處理
import numpy as np # 數值計算庫
from openai import OpenAI # OpenAI兼容API客户端
import pandas as pd # 數據分析庫
from datetime import datetime # 日期時間處理
from rank_bm25 import BM25Okapi # BM25算法實現
import jieba # 中文分詞庫
import re # 正則表達式庫
# 從環境變量中獲取 API Key
# 使用環境變量存儲敏感信息，避免硬編碼在代碼中
DASHSCOPE_API_KEY = os.getenv('DASHSCOPE_API_KEY')
# 初始化百鍊兼容的 OpenAI 客户端
# 配置API密鑰和基礎URL，使其兼容阿里雲百鍊平台
client = OpenAI(
api_key=DASHSCOPE_API_KEY, # 設置API密鑰
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 阿里雲百鍊兼容端點
)
# 預處理AI響應中的JSON格式
def preprocess_json_response(response):
"""
預處理AI響應，移除markdown代碼塊格式
確保從大模型獲取的響應能夠被正確解析為JSON格式
參數:
response (str): 從AI模型獲取的原始響應文本
返回:
str: 清理後的JSON字符串
"""
if not response: # 檢查響應是否為空
return ""
# 移除markdown代碼塊格式
# 處理以```json或```開頭的代碼塊標記
if response.startswith('```json'):
response = response[7:] # 移除 ```json (7個字符)
elif response.startswith('```'):
response = response[3:] # 移除 ``` (3個字符)
# 處理結尾的代碼塊標記
if response.endswith('```'):
response = response[:-3] # 移除結尾的 ```
return response.strip() # 移除首尾空白字符
# 基於 prompt 生成文本
def get_completion(prompt, model="qwen-turbo-latest"):
"""
使用Qwen大模型生成文本完成
參數:
prompt (str): 輸入提示文本
model (str): 使用的模型名稱，默認為"qwen-turbo-latest"
返回:
str: 模型生成的文本響應
"""
# 構建消息格式，符合OpenAI API要求
messages = [{"role": "user", "content": prompt}]
# 調用API生成文本
response = client.chat.completions.create(
model=model, # 指定模型
messages=messages, # 輸入消息
temperature=0.7, # 控制生成隨機性，0.7提供一定的創造性
)
# 返回生成的文本內容
return response.choices[0].message.content
# 文本預處理和分詞
def preprocess_text(text):
"""
文本預處理和分詞函數
對輸入文本進行清洗和分詞，為BM25檢索做準備
參數:
text (str): 待處理的文本
返回:
list: 分詞後的詞語列表
"""
if not text: # 檢查文本是否為空
return []
# 移除標點符號和特殊字符，只保留文字、數字和空白字符
text = re.sub(r'[^\w\s]', '', text)
# 使用jieba進行中文分詞
words = jieba.lcut(text)
# 定義停用詞集合，過濾常見無意義詞彙
stop_words = {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一個', '上', '也', '很', '到', '説', '要', '去', '你', '會', '着', '沒有', '看', '好', '自己', '這'}
# 過濾停用詞和短詞（長度小於2的詞）
words = [word for word in words if len(word) > 1 and word not in stop_words]
return words
class KnowledgeBaseOptimizer:
"""
知識庫優化器類
核心類，實現知識庫的問題生成、索引構建和檢索功能
"""
def __init__(self, model="qwen-turbo-latest"):
"""
初始化知識庫優化器
參數:
model (str): 使用的大模型名稱
"""
self.model = model # 大模型名稱
self.knowledge_base = [] # 存儲知識庫內容
self.content_bm25 = None # 原文內容的BM25索引
self.question_bm25 = None # 生成問題的BM25索引
self.content_documents = [] # 原文分詞後的文檔集合
self.question_documents = [] # 問題分詞後的文檔集合
self.content_metadata = [] # 原文元數據
self.question_metadata = [] # 問題元數據
def generate_questions_for_chunk(self, knowledge_chunk, num_questions=5):
"""
為單個知識切片生成多樣化問題
參數:
knowledge_chunk (str): 知識內容文本
num_questions (int): 要生成的問題數量，默認為5
返回:
list: 生成的問題字典列表
"""
# 指令模板，指導大模型生成多樣化問題
instruction = """
你是一個專業的問答系統專家。給定的知識內容能回答哪些多樣化的問題，這些問題可以：
1. 使用不同的問法（直接問、間接問、對比問等）
2. 避免重複和相似的問題
3. 確保問題不超出知識內容範圍
請返回JSON格式：
{
"questions": [
{
"question": "問題內容",
"question_type": "問題類型（直接問/間接問/對比問/條件問等）",
"difficulty": "難度等級（簡單/中等/困難）"
}
]
}
"""
# 構建提示詞，包含指令、知識內容和生成數量
prompt = f"""
### 指令 ###
{instruction}
### 知識內容 ###
{knowledge_chunk}
### 生成問題數量 ###
{num_questions}
### 生成結果 ###
"""
# 調用大模型生成問題
response = get_completion(prompt, self.model)
# 預處理響應，移除markdown代碼塊格式
response = preprocess_json_response(response)
try:
# 解析JSON響應
result = json.loads(response)
# 返回問題列表
return result.get('questions', [])
except json.JSONDecodeError as e:
# JSON解析失敗時的錯誤處理
print(f"JSON解析失敗: {e}")
print(f"AI返回內容: {response[:50]}...")
# 如果JSON解析失敗，返回簡單的問題列表
return [{"question": f"關於{knowledge_chunk[:50]}...的問題", "question_type": "直接問", "keywords": [], "difficulty": "中等"}]
def build_knowledge_index(self, knowledge_base):
"""
構建知識庫的BM25索引（包括原文和問題）
參數:
knowledge_base (list): 知識庫內容列表
"""
print("正在構建知識庫索引...")
self.knowledge_base = knowledge_base # 存儲知識庫
content_documents = [] # 初始化原文文檔列表
question_documents = [] # 初始化問題文檔列表
content_metadata = [] # 初始化原文元數據列表
question_metadata = [] # 初始化問題元數據列表
# 遍歷知識庫中的每個知識片段
for i, chunk in enumerate(knowledge_base):
# 獲取知識切片的內容
text = chunk.get('content', '')
if not text.strip(): # 跳過空內容
continue
# 處理原文文檔
content_words = preprocess_text(text) # 對原文進行分詞處理
if content_words: # 如果有有效分詞結果
content_documents.append(content_words) # 添加到原文文檔列表
# 存儲原文元數據
content_metadata.append({
"id": chunk.get('id', f"chunk_{i}"), # 文檔ID
"content": text, # 原文內容
"category": chunk.get('category', ''), # 分類信息
"chunk": chunk, # 原始知識片段
"type": "content" # 類型標記
})
# 處理問題文檔（如果存在生成的問題）
if 'generated_questions' in chunk and chunk['generated_questions']:
# 遍歷每個生成的問題
for j, question_data in enumerate(chunk['generated_questions']):
question = question_data.get('question', '')
if question.strip(): # 跳過空問題
# 拼接全文和問題，保持上下文
combined_text = f"內容：{text} 問題：{question}"
# 對組合文本進行分詞
question_words = preprocess_text(combined_text)
if question_words: # 如果有有效分詞結果
question_documents.append(question_words) # 添加到問題文檔列表
# 存儲問題元數據
question_metadata.append({
"id": f"{chunk.get('id', f'chunk_{i}')}_q{j}", # 問題ID
"content": question, # 問題內容
"combined_content": combined_text, # 組合內容
"category": chunk.get('category', ''), # 分類信息
"chunk": chunk, # 原始知識片段
"type": "question", # 類型標記
"question_data": question_data # 問題數據
})
# 創建BM25索引 - 原文索引
if content_documents:
self.content_bm25 = BM25Okapi(content_documents) # 創建BM25索引
self.content_documents = content_documents # 存儲原文文檔
self.content_metadata = content_metadata # 存儲原文元數據
print(f"原文索引構建完成，共索引 {len(content_documents)} 個知識切片")
# 創建BM25索引 - 問題索引
if question_documents:
self.question_bm25 = BM25Okapi(question_documents) # 創建BM25索引
self.question_documents = question_documents # 存儲問題文檔
self.question_metadata = question_metadata # 存儲問題元數據
print(f"問題索引構建完成，共索引 {len(question_documents)} 個問題")
# 檢查是否有有效內容
if not content_documents and not question_documents:
print("沒有有效的內容可以索引")
def search_similar_chunks(self, query, k=3, search_type="content"):
"""
使用BM25搜索相似的內容（原文或問題）
參數:
query (str): 查詢文本
k (int): 返回結果數量，默認為3
search_type (str): 搜索類型，"content"或"question"
返回:
list: 相似內容的結果列表
"""
# 根據搜索類型選擇相應的索引和元數據
if search_type == "content":
if not self.content_bm25: # 檢查索引是否存在
return []
bm25 = self.content_bm25 # 原文BM25索引
metadata_store = self.content_metadata # 原文元數據
elif search_type == "question":
if not self.question_bm25: # 檢查索引是否存在
return []
bm25 = self.question_bm25 # 問題BM25索引
metadata_store = self.question_metadata # 問題元數據
else:
return [] # 無效的搜索類型
try:
# 預處理查詢文本
query_words = preprocess_text(query)
if not query_words: # 檢查是否有有效分詞
return []
# 使用BM25計算相似度分數
scores = bm25.get_scores(query_words)
# 獲取top-k結果的索引（按分數降序排列）
top_indices = np.argsort(scores)[::-1][:k]
results = [] # 初始化結果列表
for idx in top_indices:
if scores[idx] > 0: # 只返回有相關性的結果（分數>0）
metadata = metadata_store[idx] # 獲取元數據
# 將BM25分數轉換為0-1範圍的相似度（歸一化）
similarity = min(1.0, scores[idx] / 10.0)
# 添加結果到列表
results.append({
"metadata": metadata, # 元數據
"score": scores[idx], # 原始分數
"similarity": similarity # 歸一化相似度
})
return results
except Exception as e:
# 異常處理
print(f"搜索失敗: {e}")
return []
def calculate_similarity(self, query, knowledge_chunk):
"""
計算查詢與知識切片的相似度（使用BM25）
參數:
query (str): 查詢文本
knowledge_chunk (str): 知識內容文本
返回:
float: 相似度分數（0-1範圍）
"""
try:
# 預處理查詢和知識內容
query_words = preprocess_text(query)
chunk_words = preprocess_text(knowledge_chunk)
if not query_words or not chunk_words: # 檢查是否有有效分詞
return 0.0
# 創建臨時BM25索引（僅包含知識內容）
temp_bm25 = BM25Okapi([chunk_words])
# 計算查詢與知識內容的相似度分數
scores = temp_bm25.get_scores(query_words)
# 返回最高分數並歸一化
max_score = max(scores) if scores else 0.0
return min(1.0, max_score / 10.0)
except Exception as e:
# 異常處理
print(f"相似度計算失敗: {e}")
return 0.0
def calculate_question_similarity(self, user_query, generated_questions):
"""
計算用户查詢與生成問題的相似度
參數:
user_query (str): 用户查詢文本
generated_questions (list): 生成的問題列表
返回:
float: 最大相似度分數
"""
similarities = [] # 初始化相似度列表
# 計算用户查詢與每個生成問題的相似度
for question_data in generated_questions:
question = question_data['question'] # 獲取問題文本
similarity = self.calculate_similarity(user_query, question) # 計算相似度
similarities.append(similarity) # 添加到列表
# 返回最大相似度
return max(similarities) if similarities else 0.0
def evaluate_retrieval_methods(self, knowledge_base, test_queries):
"""
評估兩種檢索方法的準確度
參數:
knowledge_base (list): 知識庫內容列表
test_queries (list): 測試查詢列表
返回:
dict: 評估結果字典
"""
# 首先構建知識庫索引（包括原文和問題）
self.build_knowledge_index(knowledge_base)
# 初始化結果字典
results = {
'content_similarity': [], # 原文檢索正確性
'question_similarity': [], # 問題檢索正確性
'improvement': [], # 問題檢索是否改進
'content_scores': [], # 原文檢索分數
'question_scores': [], # 問題檢索分數
'query_details': [] # 詳細查詢信息
}
# 遍歷測試查詢
for i, query_info in enumerate(test_queries):
user_query = query_info['query'] # 獲取查詢文本
correct_chunk = query_info['correct_chunk'] # 獲取正確答案對應的知識片段
# 方法1：BM25原文檢索
content_results = self.search_similar_chunks(user_query, k=1, search_type="content")
content_correct = False # 初始化正確性標誌
content_score = 0.0 # 初始化分數
content_chunk_id = None # 初始化知識片段ID
if content_results: # 如果有檢索結果
best_match = content_results[0]['metadata']['chunk'] # 獲取最佳匹配
# 檢查是否匹配正確答案
content_correct = best_match['content'] == correct_chunk
content_score = content_results[0]['similarity'] # 獲取相似度分數
content_chunk_id = best_match['id'] # 獲取知識片段ID
# 方法2：BM25問題檢索
question_results = self.search_similar_chunks(user_query, k=1, search_type="question")
question_correct = False # 初始化正確性標誌
question_score = 0.0 # 初始化分數
question_chunk_id = None # 初始化知識片段ID
if question_results: # 如果有檢索結果
best_match = question_results[0]['metadata']['chunk'] # 獲取最佳匹配
# 檢查是否匹配正確答案
question_correct = best_match['content'] == correct_chunk
question_score = question_results[0]['similarity'] # 獲取相似度分數
question_chunk_id = best_match['id'] # 獲取知識片段ID
# 記錄結果
results['content_similarity'].append(content_correct)
results['question_similarity'].append(question_correct)
results['improvement'].append(question_correct and not content_correct) # 問題檢索是否改進
results['content_scores'].append(content_score)
results['question_scores'].append(question_score)
# 記錄查詢詳情
results['query_details'].append({
'query': user_query, # 查詢文本
'content_score': content_score, # 原文檢索分數
'question_score': question_score, # 問題檢索分數
'content_correct': content_correct, # 原文檢索是否正確
'question_correct': question_correct, # 問題檢索是否正確
'score_diff': question_score - content_score, # 分數差異
'content_chunk_id': content_chunk_id, # 原文檢索匹配的知識ID
'question_chunk_id': question_chunk_id # 問題檢索匹配的知識ID
})
return results # 返回評估結果
def generate_diverse_questions(self, knowledge_chunk, num_questions=8):
"""
生成更多樣化的問題（更豐富）
參數:
knowledge_chunk (str): 知識內容文本
num_questions (int): 要生成的問題數量，默認為8
返回:
list: 生成的問題字典列表
"""
# 指令模板，指導大模型生成更多樣化的問題
instruction = """
你是一個專業的問答系統專家。請為給定的知識內容生成高度多樣化的問題，確保：
1. 問題類型多樣化：直接問、間接問、對比問、條件問、假設問、推理問等
2. 表達方式多樣化：使用不同的句式、詞彙、語氣
3. 難度層次多樣化：簡單、中等、困難的問題都要有
4. 角度多樣化：從不同角度和維度提問
5. 確保問題不超出知識內容範圍
請返回JSON格式：
{
"questions": [
{
"question": "問題內容",
"question_type": "問題類型",
"difficulty": "難度等級",
"perspective": "提問角度",
"is_answerable": "給出的知識能否回答該問題",
"answer": "基於該知識的回答"
}
]
}
"""
# 構建提示詞
prompt = f"""
### 指令 ###
{instruction}
### 知識內容 ###
{knowledge_chunk}
### 生成問題數量 ###
{num_questions}
### 生成結果 ###
"""
# 調用大模型生成問題
response = get_completion(prompt, self.model)
# 預處理響應，移除markdown代碼塊格式
response = preprocess_json_response(response)
try:
# 解析JSON響應
result = json.loads(response)
# 返回問題列表
return result.get('questions', [])
except json.JSONDecodeError as e:
# JSON解析失敗時的錯誤處理
print(f"多樣化問題生成JSON解析失敗: {e}")
print(f"AI返回內容: {response[:200]}...")
return []
def main():
"""
主函數 - 演示知識庫優化器的使用
"""
# 初始化知識庫優化器
optimizer = KnowledgeBaseOptimizer()
print("=== 知識庫問題生成與檢索優化示例（BM25版本）- 飲食與健康 ===\n")
# 示例知識庫 - 飲食與健康相關信息
knowledge_base = [
{
"id": "kb_001",
"content": "每天飲用足夠的水對維持身體健康至關重要。成年人每天應飲用約2升水，相當於8杯水。充足的水分攝入有助於新陳代謝、排毒和保持皮膚健康。",
"category": "健康飲水"
},
{
"id": "kb_002",
"content": "均衡飲食應包含五大類食物：穀物、蔬菜、水果、蛋白質和乳製品。建議每天攝入5份不同顏色的蔬菜和水果，以確保獲得各種維生素和礦物質。",
"category": "營養飲食"
},
{
"id": "kb_003",
"content": "成年人每週應進行至少150分鐘的中等強度有氧運動，如快走、游泳或騎自行車，或75分鐘的高強度有氧運動。此外，每週還應進行2次肌肉強化活動。",
"category": "運動健身"
},
{
"id": "kb_004",
"content": "成年人每天需要7-9小時的睡眠。良好的睡眠習慣包括保持固定的睡眠時間、創造舒適的睡眠環境、避免睡前使用電子設備以及限制咖啡因和酒精的攝入。",
"category": "睡眠健康"
}
]
# 示例1: 為知識切片生成問題
print("示例1: 為知識切片生成多樣化問題")
test_chunk = knowledge_base[0]['content'] # 獲取第一個知識片段
print(f"知識內容: {test_chunk}")
# 生成5個問題
questions = optimizer.generate_questions_for_chunk(test_chunk, num_questions=5)
print(f"\n生成的5個問題:")
for i, q in enumerate(questions, 1):
print(f" {i}. {q['question']} (類型: {q['question_type']}, 難度: {q['difficulty']})")
print("\n" + "="*60 + "\n")
# 示例2: 生成更多樣化的問題
print("示例2: 生成更多樣化的問題（8個）")
# 生成8個更多樣化的問題
diverse_questions = optimizer.generate_diverse_questions(test_chunk, num_questions=8)
print(f"\n生成的8個多樣化問題:")
for i, q in enumerate(diverse_questions, 1):
print(f" {i}. {q['question']}")
print(f" 類型: {q['question_type']}, 難度: {q['difficulty']}, 角度: {q['perspective']}, 能否回答: {q['is_answerable']}, 回答的答案：{q['answer']}")
print("\n" + "="*60 + "\n")
# 示例3: 評估檢索方法
print("示例3: 評估兩種檢索方法的準確度")
# 測試查詢 - 設計更有挑戰性的問題
test_queries = [
{
"query": "每天應該喝多少水？",
"correct_chunk": knowledge_base[0]['content']
},
{
"query": "什麼是均衡飲食？",
"correct_chunk": knowledge_base[1]['content']
},
{
"query": "每週需要運動多久？",
"correct_chunk": knowledge_base[2]['content']
}
]
# 為知識庫生成問題
print('正在為知識庫生成問題...')
for chunk in knowledge_base:
# 為每個知識片段生成問題
chunk['generated_questions'] = optimizer.generate_questions_for_chunk(chunk['content'])
print('為知識庫生成問題完畢')
# 評估檢索方法
results = optimizer.evaluate_retrieval_methods(knowledge_base, test_queries)
# 輸出評估結果
print(f"測試查詢數量: {len(test_queries)}")
print(f"BM25原文檢索準確率: {sum(results['content_similarity'])/len(results['content_similarity'])*100:.1f}%")
print(f"BM25問題檢索準確率: {sum(results['question_similarity'])/len(results['question_similarity'])*100:.1f}%")
print(f"問題檢索改進的查詢數量: {sum(results['improvement'])}")
# 詳細分析
print(f"\n=== 詳細分析 ===")
# 按相似度分數差異排序
sorted_details = sorted(results['query_details'], key=lambda x: x['score_diff'], reverse=True)
print(f"\n問題檢索方法表現更好的查詢（按分數差異排序）:")
for i, detail in enumerate(sorted_details[:5], 1):
if detail['score_diff'] > 0:
print(f" {i}. 查詢: {detail['query']}")
print(f" 原文檢索分數: {detail['content_score']:.3f}")
print(f" 問題檢索分數: {detail['question_score']:.3f}")
print(f" 分數差異: +{detail['score_diff']:.3f}")
print(f" 原文檢索: {'✓' if detail['content_correct'] else '✗'}")
print(f" 問題檢索: {'✓' if detail['question_correct'] else '✗'}")
print(f"\n原文檢索方法表現更好的查詢:")
for i, detail in enumerate(sorted_details[-5:], 1):
if detail['score_diff'] < 0:
print(f" {i}. 查詢: {detail['query']}")
print(f" 原文檢索分數: {detail['content_score']:.3f}")
print(f" 問題檢索分數: {detail['question_score']:.3f}")
print(f" 分數差異: {detail['score_diff']:.3f}")
print(f" 原文檢索: {'✓' if detail['content_correct'] else '✗'}")
print(f" 問題檢索: {'✓' if detail['question_correct'] else '✗'}")
# 程序入口點
if __name__ == "__main__":
main() # 執行主函數
執行結果
=== 知識庫問題生成與檢索優化示例（BM25版本）- 飲食與健康 ===
示例1: 為知識切片生成多樣化問題
知識內容: 每天飲用足夠的水對維持身體健康至關重要。成年人每天應飲用約2升水，相當於8杯水。充足的水分攝入有助於新陳代謝
、排毒和保持皮膚健康。
生成的5個問題:
1. 成年人每天應該喝多少水？ (類型: 直接問, 難度: 簡單)
2. 為什麼保持水分攝入對身體很重要？ (類型: 間接問, 難度: 中等)
3. 如果一個人每天只喝1升水，會對健康產生什麼影響？ (類型: 條件問, 難度: 中等)
4. 喝足夠的水和皮膚健康之間有什麼關係？ (類型: 對比問, 難度: 簡單)
5. 除了排毒之外，充足飲水還能幫助身體實現哪些功能？ (類型: 間接問, 難度: 中等)
============================================================
示例2: 生成更多樣化的問題（8個）
生成的8個多樣化問題:
1. 成年人每天應該喝多少水？
類型: 直接問, 難度: 簡單, 角度: 事實確認, 能否回答: True, 回答的答案：成年人每天應飲用約2升水，相當於8杯水。
2. 如果一個人不喝夠水，可能會出現什麼健康問題？
類型: 推理問, 難度: 中等, 角度: 後果分析, 能否回答: True, 回答的答案：缺乏足夠水分可能導致新陳代謝減慢、排毒功能
下降和皮膚健康受損。
3. 為什麼説喝水對皮膚好？
類型: 間接問, 難度: 簡單, 角度: 機制解釋, 能否回答: True, 回答的答案：充足的水分攝入有助於保持皮膚健康。
4. 假設一個人只喝1升水，是否還能維持正常的新陳代謝？
類型: 假設問, 難度: 困難, 角度: 條件推演, 能否回答: True, 回答的答案：可能無法維持正常的新陳代謝，因為知識指出充
足水分對新陳代謝至關重要。
5. 每天喝8杯水和喝2升水是同一個意思嗎？
類型: 對比問, 難度: 簡單, 角度: 單位換算, 能否回答: True, 回答的答案：是的，2升水相當於8杯水。
6. 有沒有可能比2升更多水才更有利於健康？
類型: 條件問, 難度: 中等, 角度: 優化建議, 能否回答: True, 回答的答案：知識未提及超過2升是否更有益，因此無法判斷
，但説明2升是推薦量。
7. 從科學角度看，喝水對身體的哪些系統有幫助？
類型: 推理問, 難度: 困難, 角度: 生理機制, 能否回答: True, 回答的答案：喝水有助於新陳代謝（消化/能量系統）、排毒
（泌尿/肝臟系統）和皮膚健康（表皮系統）。
8. 如果某人每天都喝足2升水，他是不是就一定健康？
類型: 假設問, 難度: 中等, 角度: 因果關係辨析, 能否回答: True, 回答的答案：不一定，雖然飲水充足是健康的重要因素，
但身體健康還取決於其他多種因素如飲食、運動等。
============================================================
示例3: 評估兩種檢索方法的準確度
正在為知識庫生成問題...
為知識庫生成問題完畢
正在構建知識庫索引...
Building prefix dict from the default dictionary ...
Loading model from cache C:\Users\du\AppData\Local\Temp\jieba.cache
Loading model cost 0.557 seconds.
Prefix dict has been built successfully.
原文索引構建完成，共索引 4 個知識切片
問題索引構建完成，共索引 20 個問題
測試查詢數量: 3
BM25原文檢索準確率: 100.0%
BM25問題檢索準確率: 100.0%
問題檢索改進的查詢數量: 0
=== 詳細分析 ===
問題檢索方法表現更好的查詢（按分數差異排序）:
1. 查詢: 每天應該喝多少水？
原文檢索分數: 0.028
問題檢索分數: 0.410
分數差異: +0.381
原文檢索: ✓
問題檢索: ✓
2. 查詢: 什麼是均衡飲食？
原文檢索分數: 0.178
問題檢索分數: 0.450
分數差異: +0.272
原文檢索: ✓
問題檢索: ✓
3. 查詢: 每週需要運動多久？
原文檢索分數: 0.237
問題檢索分數: 0.417
分數差異: +0.179
原文檢索: ✓
問題檢索: ✓
原文檢索方法表現更好的查詢:
實現功能
- 知識庫問題生成：使用Qwen大模型為知識內容自動生成多樣化的問題
- BM25索引構建：為知識庫的原文和生成的問題分別構建BM25索引
- 檢索功能：支持基於原文和基於問題的兩種檢索方式計算出檢索分數
- 性能評估：比較兩種檢索方法的準確度和效果差異
- 多樣化問題生成：生成更多樣化、多角度的問題
擴展介紹：BM25檢索方法
- BM25原文檢索：
- 檢索對象：知識庫中的原始內容片段，即 knowledge_base[i]['content']。
- 過程：使用BM25算法將用户的查詢直接與這些原始內容進行關鍵詞匹配和相似度計算，然後返回最相似的原始內容片段。
- 類比：像是在一整本書的正文裏搜索與你問題相關的段落。
- 優點：直接，沒有中間轉換，直接匹配內容。覆蓋全，可以匹配到答案中任何出現的關鍵詞。
- 缺點：詞彙不匹配，如果用户的問法和原文表述差異很大，就可能檢索失敗。由於關鍵詞不匹配，BM25可能無法建立有效關聯，導致檢索不到或排名很低。
- BM25問題檢索：
- 檢索對象：為每個知識塊預先生成的一系列問題，即 knowledge_base[i]['generated_questions'][j]['question']。
- 過程：將用户的查詢與這些預先生成的問題進行匹配。如果找到相似的問題，就返回該問題所對應的那個原始知識塊。
- 類比：像是在一本書末尾的索引或常見問題清單裏搜索。你先找到一個和你提問方式很像的標準問題，然後根據這個標準問題去翻到它對應的答案頁碼。
- 優點：解決詞彙不匹配，極大地豐富了問法的多樣性，用户的自然語言提問更容易與某個生成的問題匹配上。更符合用户習慣檢索的是“問題”，匹配的是“問題”，更貼近用户的意圖。
- 缺點：依賴生成問題的質量，如果AI生成的問題不好、不全面或不準確，會直接導致檢索失敗。增加預處理步驟，需要額外消耗資源（API調用、計算時間）來為所有知識塊生成問題。
- 準確率計算：
test_queries = [
{
"query": "可以帶食物進去嗎？", # 用户查詢
"correct_chunk": knowledge_base[5]['content'] # 唯一正確的知識塊內容
},
# ... 其他測試用例
]
在代碼的評估部分 (evaluate_retrieval_methods函數)，有一個測試集 test_queries。每個測試用例都明確指定了哪個知識塊是正確答案。
原文檢索準確率：系統用BM25原文檢索返回的 top-1 結果，即它認為最相關的那個知識塊，其內容是否等於 correct_chunk？如果是，則這個查詢上原文檢索“準確”，否則“不準確”。
準確次數 / 總查詢次數 = 原文檢索準確率
問題檢索準確率：系統用BM25問題檢索返回的 top-1 結果所對應的那個知識塊，其內容是否等於 correct_chunk？如果是，則這個查詢上問題檢索“準確”；否則“不準確”。
準確次數 / 總查詢次數 = 問題檢索準確率
BM25原文檢索準確率：衡量系統直接通過關鍵詞匹配找到正確答案的能力。
BM25問題檢索準確率：衡量系統通過“問題橋接”的方式找到正確答案的能力。這種方法的核心價值在於通過擴展問法來提升在詞彙不匹配場景下的檢索成功率。
五、對話知識沉澱：從交互中挖掘隱性知識
1. 基礎介紹
用户與客服、助理的對話是未被開發的知識金礦。對話知識沉澱旨在自動化地完成“挖掘”和“提煉”過程。
- 理論核心：持續學習和閉環反饋。
- 技術流程：
- 識別：利用模型自動識別對話中的“知識價值點”。例如：
- 未命中：模型未能回答的新問題。
- 新答案：針對已知問題，用户或專家提供了更優的答案。
- 知識確認：用户對現有答案的正面/負面反饋。
- 抽取：對於有價值的對話，使用大模型進行摘要總結和結構化抽取，將其轉化為標準的Q-A對或其他知識形態。
- 審核與入庫：抽取的結果經過人工或自動化規則審核後，方可併入正式知識庫，確保質量。
- 價值：使知識庫成為一個能夠從真實使用場景中學習併成長的有機體，實現了數據驅動的自我優化。
2. 案例演示
代碼結構
# 對話知識提取與沉澱
# 導入依賴庫
import dashscope
import os
import json
from datetime import datetime
from collections import Counter
# 從環境變量中獲取 API Key
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
# 預處理AI響應中的JSON格式
def preprocess_json_response(response):
"""預處理AI響應，移除markdown代碼塊格式"""
if not response:
return ""
# 移除markdown代碼塊格式
if response.startswith('```json'):
response = response[7:] # 移除 ```json
elif response.startswith('```'):
response = response[3:] # 移除 ```
if response.endswith('```'):
response = response[:-3] # 移除結尾的 ```
return response.strip() # 移除首尾空白
# 基於 prompt 生成文本
def get_completion(prompt, model="qwen-turbo-latest"):
messages = [{"role": "user", "content": prompt}]
response = dashscope.Generation.call(
model=model,
messages=messages,
result_format='message',
temperature=0.3,
)
return response.output.choices[0].message.content
class ConversationKnowledgeExtractor:
def __init__(self, model="qwen-turbo-latest"):
self.model = model
self.extracted_knowledge = []
self.knowledge_frequency = Counter()
def extract_knowledge_from_conversation(self, conversation):
"""從單次對話中提取知識"""
instruction = """
你是一個專業的知識提取專家。請從給定的對話中提取有價值的知識點，包括：
1. 事實性信息（地點、時間、價格、規則等）
2. 用户需求和偏好
3. 常見問題和解答
4. 操作流程和步驟
5. 注意事項和提醒
請返回JSON格式：
{
"extracted_knowledge": [
{
"knowledge_type": "知識類型（事實/需求/問題/流程/注意）",
"content": "知識內容",
"confidence": "置信度(0-1)",
"source": "來源（用户/AI/對話）",
"keywords": ["關鍵詞1", "關鍵詞2"],
"category": "分類"
}
],
"conversation_summary": "對話摘要",
"user_intent": "用户意圖"
}
"""
prompt = f"""
### 指令 ###
{instruction}
### 對話內容 ###
{conversation}
### 提取結果 ###
"""
response = get_completion(prompt, self.model)
# 預處理響應，移除markdown代碼塊格式
response = preprocess_json_response(response)
try:
result = json.loads(response)
return result
except json.JSONDecodeError as e:
print(f"對話知識提取JSON解析失敗: {e}")
print(f"AI返回內容: {response[:200]}...")
return {
"extracted_knowledge": [],
"conversation_summary": "無法解析對話",
"user_intent": "未知"
}
def batch_extract_knowledge(self, conversations):
"""批量提取知識"""
all_knowledge = []
for i, conversation in enumerate(conversations):
print(f"正在處理對話 {i+1}/{len(conversations)}...")
result = self.extract_knowledge_from_conversation(conversation)
all_knowledge.extend(result.get('extracted_knowledge', []))
# 更新頻率統計
for knowledge in result.get('extracted_knowledge', []):
key = f"{knowledge['knowledge_type']}:{knowledge['content'][:50]}"
self.knowledge_frequency[key] += 1
return all_knowledge
def merge_similar_knowledge(self, knowledge_list):
"""使用LLM合併相似的知識點，過濾掉需求和問題類型"""
# 過濾掉需求和問題類型的知識，因為它們是臨時的、個性化的
filtered_knowledge = [
knowledge for knowledge in knowledge_list
if knowledge.get('knowledge_type') not in ['需求', '問題']
]
print(f"過濾前知識點數量: {len(knowledge_list)}")
print(f"過濾後知識點數量: {len(filtered_knowledge)}")
print(f"過濾掉的'需求'和'問題'類型知識點: {len(knowledge_list) - len(filtered_knowledge)}")
# 按知識類型分組
knowledge_by_type = {}
for knowledge in filtered_knowledge:
knowledge_type = knowledge.get('knowledge_type', '其他')
if knowledge_type not in knowledge_by_type:
knowledge_by_type[knowledge_type] = []
knowledge_by_type[knowledge_type].append(knowledge)
merged_knowledge = []
# 對每個知識類型分別進行LLM合併
for knowledge_type, knowledge_group in knowledge_by_type.items():
if len(knowledge_group) == 1:
# 只有一個知識點，直接添加
merged_knowledge.append(knowledge_group[0])
else:
# 多個知識點，使用LLM合併
merged = self.merge_knowledge_with_llm(knowledge_group, knowledge_type)
merged_knowledge.append(merged)
return merged_knowledge
def merge_knowledge_with_llm(self, knowledge_group, knowledge_type):
"""使用LLM合併同類型的知識組"""
# 準備知識內容列表
knowledge_contents = []
all_keywords = set()
all_sources = []
for i, knowledge in enumerate(knowledge_group, 1):
content = knowledge.get('content', '')
confidence = knowledge.get('confidence', 0.5)
keywords = knowledge.get('keywords', [])
source = knowledge.get('source', '')
category = knowledge.get('category', '')
knowledge_contents.append(f"{i}. 內容: {content}")
knowledge_contents.append(f" 置信度: {confidence}")
knowledge_contents.append(f" 分類: {category}")
knowledge_contents.append(f" 來源: {source}")
knowledge_contents.append(f" 關鍵詞: {', '.join(keywords)}")
knowledge_contents.append("")
all_keywords.update(keywords)
if source and source not in all_sources:
all_sources.append(source)
# 構建LLM合併提示
prompt = f"""
你是一個專業的知識整理專家。請將以下{knowledge_type}類型的知識點進行智能合併，生成一個更完整、準確的知識點。
### 合併要求：
1. 保留所有重要信息，避免信息丟失
2. 消除重複內容，整合相似表述
3. 提高內容的準確性和完整性
4. 保持邏輯清晰，結構合理
5. 合併後的置信度取所有知識點中的最高值
### 待合併的知識點：
{chr(10).join(knowledge_contents)}
### 請返回JSON格式：
{{
"knowledge_type": "{knowledge_type}",
"content": "合併後的知識內容",
"confidence": 最高置信度值,
"keywords": ["合併後的關鍵詞列表"],
"category": "合併後的分類",
"sources": ["所有來源"],
"frequency": {len(knowledge_group)}
}}
### 合併結果：
"""
response = get_completion(prompt, self.model)
# 預處理響應
response = preprocess_json_response(response)
try:
result = json.loads(response)
return result
except json.JSONDecodeError as e:
print(f"知識合併JSON解析失敗: {e}")
print(f"AI返回內容: {response[:200]}...")
# 解析失敗時，使用簡單的合併策略
best_knowledge = max(knowledge_group, key=lambda x: x.get('confidence', 0))
return {
"knowledge_type": knowledge_type,
"content": best_knowledge['content'],
"confidence": best_knowledge.get('confidence', 0.5),
"frequency": len(knowledge_group),
"keywords": list(all_keywords),
"category": best_knowledge['category'],
"sources": all_sources
}
def main():
# 初始化對話知識提取器
extractor = ConversationKnowledgeExtractor()
print("=== 對話知識提取與沉澱示例（健康健身與營養諮詢） ===\n")
# 健康健身主題的示例對話數據
sample_conversations = [
# 對話1: 減脂諮詢
"""
用户: "我想減脂，應該怎麼安排運動和飲食？"
AI: "減脂需要創造熱量赤字，建議每週進行3-5次有氧運動（如跑步、游泳、騎行）和2-3次力量訓練。飲食上控制總熱量攝入，增加蛋白質比例，減少精製碳水和添加糖。"
用户: "有氧運動做多久有效果？"
AI: "有氧運動建議每次30-60分鐘，保持中等強度（心率在最大心率的60-70%）。最好結合高強度間歇訓練(HIIT)和穩態有氧(LISS)交替進行，效果更佳。"
用户: "減脂期間蛋白質吃多少合適？"
AI: "減脂期間建議每日蛋白質攝入量為每公斤體重1.6-2.2克。優質蛋白來源包括雞胸肉、魚、雞蛋、豆製品和乳清蛋白粉。"
""",
# 對話2: 增肌諮詢
"""
用户: "我想增加肌肉量，應該怎麼訓練和吃？"
AI: "增肌需要漸進式超負荷訓練和熱量盈餘。建議每週進行3-5次力量訓練，專注複合動作如深蹲、卧推、硬拉和引體向上。飲食上每日熱量盈餘300-500卡路里，蛋白質攝入每公斤體重1.6-2.2克。"
用户: "訓練後需要立即補充蛋白質嗎？"
AI: "訓練後30-90分鐘內是補充蛋白質的窗口期，建議攝入20-40克優質蛋白質，配合快速吸收的碳水化合物，有助於肌肉恢復和合成。乳清蛋白粉是方便的選擇。"
用户: "增肌需要做有氧運動嗎？"
AI: "增肌期間可以適量進行有氧運動，每週2-3次，每次20-30分鐘低強度有氧，有助於保持心肺健康和代謝功能，但不要過量以免影響力量訓練恢復。"
""",
# 對話3: 健康飲食諮詢
"""
用户: "什麼樣的飲食算健康均衡？"
AI: "健康均衡飲食應包含多樣化的食物：大量蔬菜和水果、優質蛋白質來源、全穀物碳水化合物、健康脂肪以及充足的水分。遵循'盤子法則'：一半盤子放蔬菜，1/4放蛋白質，1/4放全穀物。"
用户: "減脂期晚上可以吃碳水嗎？"
AI: "減脂期晚上可以適量攝入複合碳水化合物，如糙米、燕麥、紅薯等，但需控制總量並避免在臨睡前大量攝入。個體差異較大，建議根據自身代謝情況和訓練安排調整。"
用户: "健身補劑有必要嗎？哪些比較推薦？"
AI: "健身補劑不是必需品，但可以輔助訓練效果。基礎補劑包括乳清蛋白粉、肌酸和Omega-3脂肪酸。維生素D和鎂也可能對運動表現和恢復有益。選擇補劑前最好諮詢專業人士。"
""",
# 對話4: 運動損傷預防
"""
用户: "如何預防健身時的運動損傷？"
AI: "預防運動損傷的關鍵包括：充分熱身5-10分鐘、循序漸進增加負荷、保持正確動作形式、合理安排訓練和休息、以及運動後的拉伸和恢復。傾聽身體信號，避免過度訓練。"
用户: "肩膀疼痛還能繼續訓練嗎？"
AI: "如果出現持續性肩膀疼痛，應停止加重疼痛的訓練動作，並諮詢醫生或物理治療師。可以暫時進行不引起疼痛的其他部位訓練，但避免強行訓練疼痛部位。"
用户: "運動後如何加速恢復？"
AI: "運動後恢復包括：適當營養補充（蛋白質和碳水）、充足水分、7-9小時高質量睡眠、主動恢復（如散步、瑜伽）、拉伸和泡沫軸放鬆，以及必要時使用冰敷或熱敷。"
"""
]
# 示例1: 從單次對話中提取知識
print("示例1: 從單次對話中提取知識")
conversation = sample_conversations[0]
print(f"對話內容:\n{conversation}")
extracted = extractor.extract_knowledge_from_conversation(conversation)
print(f"\n提取的知識點:")
for i, knowledge in enumerate(extracted['extracted_knowledge'], 1):
print(f" {i}. 類型: {knowledge['knowledge_type']}")
print(f" 內容: {knowledge['content']}")
print(f" 置信度: {knowledge['confidence']}")
print(f" 分類: {knowledge['category']}")
print(f"\n對話摘要: {extracted['conversation_summary']}")
print(f"用户意圖: {extracted['user_intent']}")
print("\n" + "="*60 + "\n")
# 示例2: 批量提取知識
print("示例2: 批量提取知識")
all_knowledge = extractor.batch_extract_knowledge(sample_conversations)
print(f"總共提取了 {len(all_knowledge)} 個知識點")
# 顯示所有知識點
print(f"\n所有知識點:")
for key, count in extractor.knowledge_frequency.most_common():
print(f" {key}: {count}次")
print("\n" + "="*60 + "\n")
# 示例3: 合併相似知識
print("示例3: 合併相似知識")
merged_knowledge = extractor.merge_similar_knowledge(all_knowledge)
print(f"合併後剩餘 {len(merged_knowledge)} 個知識點")
print(f"\n合併後的知識點:")
for i, knowledge in enumerate(merged_knowledge, 1):
print(f" {i}. 類型: {knowledge.get('knowledge_type', '未知')}")
print(f" 內容: {knowledge['content']}")
print(f" 頻率: {knowledge.get('frequency', 1)}次")
print(f" 置信度: {knowledge.get('confidence', 0.5)}")
print(f" 分類: {knowledge.get('category', '未知')}")
print(f" 關鍵詞: {knowledge.get('keywords', [])}")
print(f" 來源: {knowledge.get('sources', [])}")
print()
print("\n" + "="*60 + "\n")
if __name__ == "__main__":
main()
執行結果
=== 對話知識提取與沉澱示例（健康健身與營養諮詢） ===
示例1: 從單次對話中提取知識
對話內容:
用户: "我想減脂，應該怎麼安排運動和飲食？"
AI: "減脂需要創造熱量赤字，建議每週進行3-5次有氧運動（如跑步、游泳、騎行）和2-3次力量訓練。飲食上控制總熱量攝入，增加
蛋白質比例，減少精製碳水和添加糖。"
用户: "有氧運動做多久有效果？"
AI: "有氧運動建議每次30-60分鐘，保持中等強度（心率在最大心率的60-70%）。最好結合高強度間歇訓練(HIIT)和穩態有氧(LISS)
交替進行，效果更佳。"
用户: "減脂期間蛋白質吃多少合適？"
AI: "減脂期間建議每日蛋白質攝入量為每公斤體重1.6-2.2克。優質蛋白來源包括雞胸肉、魚、雞蛋、豆製品和乳清蛋白粉。"
提取的知識點:
1. 類型: 流程
內容: 減脂需結合運動與飲食：每週3-5次有氧運動（如跑步、游泳、騎行）和2-3次力量訓練；飲食上控制總熱量，增加蛋白質
比例，減少精製碳水和添加糖。
置信度: 0.95
分類: 健身營養
2. 類型: 事實
內容: 有氧運動建議每次持續30-60分鐘，保持中等強度（心率在最大心率的60-70%），可結合HIIT和LISS交替進行以提升效果
。
置信度: 0.9
分類: 運動科學
3. 類型: 事實
內容: 減脂期間每日蛋白質攝入推薦量為每公斤體重1.6-2.2克，優質蛋白來源包括雞胸肉、魚、雞蛋、豆製品和乳清蛋白粉。
置信度: 0.95
分類: 營養學
4. 類型: 需求
內容: 用户希望瞭解如何通過運動和飲食安排實現減脂目標，關注具體執行方案（如運動時長、蛋白質攝入量）。
置信度: 0.9
分類: 用户目標
5. 類型: 問題
內容: 用户詢問有氧運動做多久才有效果，AI回答建議每次30-60分鐘，中等強度下效果更佳。
置信度: 0.9
分類: 常見疑問
對話摘要: 用户諮詢減脂期間的運動與飲食安排，AI提供了包含有氧運動頻率與時長、力量訓練建議、蛋白質攝入量及飲食結構優化
的具體方案，並回應了關於運動時長的有效性問題。
用户意圖: 獲取科學有效的減脂方案，包括運動頻率、時長、強度以及飲食中的蛋白質攝入建議。
============================================================
示例2: 批量提取知識
正在處理對話 1/4...
正在處理對話 2/4...
正在處理對話 3/4...
正在處理對話 4/4...
總共提取了 24 個知識點
所有知識點:
流程:減脂需結合運動與飲食：每週3-5次有氧運動（如跑步、游泳、騎行）和2-3次力量訓練；飲食上控制總熱量: 1次
事實:有氧運動建議每次持續30-60分鐘，保持中等強度（心率在最大心率的60-70%），可結合HIIT和L: 1次
事實:減脂期間每日蛋白質攝入推薦量為每公斤體重1.6-2.2克，優質蛋白來源包括雞胸肉、魚、雞蛋、豆製品和: 1次
需求:用户希望瞭解如何通過運動和飲食安排實現減脂目標，關注具體執行方案（如運動時長、蛋白質攝入量）。: 1次
問題:用户詢問有氧運動做多久有效果，AI回答建議每次30-60分鐘，保持中等強度，並推薦結合HIIT和LI: 1次
流程:增肌訓練建議每週進行3-5次力量訓練，專注複合動作如深蹲、卧推、硬拉和引體向上；飲食上每日熱量盈餘3: 1次
事實:訓練後30-90分鐘內是補充蛋白質的窗口期，建議攝入20-40克優質蛋白質，配合快速吸收的碳水化合物: 1次
事實:增肌期間可以適量進行有氧運動，每週2-3次，每次20-30分鐘低強度有氧，有助於保持心肺健康和代謝功: 1次
需求:用户希望瞭解如何通過訓練和飲食增加肌肉量，並關注訓練後的營養補充和有氧運動對增肌的影響。: 1次
問題:訓練後是否需要立即補充蛋白質？: 1次
問題:增肌期間是否需要做有氧運動？: 1次
注意:有氧運動不宜過量，以免影響力量訓練的恢復，應控制在每週2-3次、每次20-30分鐘低強度。: 1次
事實:健康均衡飲食應包含大量蔬菜和水果、優質蛋白質來源、全穀物碳水化合物、健康脂肪以及充足的水分。: 1次
事實:減脂期晚上可以適量攝入複合碳水化合物，如糙米、燕麥、紅薯等，但需控制總量並避免在臨睡前大量攝入。: 1次
事實:健身補劑不是必需品，但可以輔助訓練效果。基礎補劑包括乳清蛋白粉、肌酸和Omega-3脂肪酸；維生素D: 1次
需求:用户希望瞭解如何實現健康均衡飲食，以及在減脂期和健身場景下如何科學安排飲食和補劑。: 1次
問題:減脂期晚上是否可以吃碳水？: 1次
流程:遵循'盤子法則'：一半盤子放蔬菜，1/4放蛋白質，1/4放全穀物，以實現健康均衡飲食。: 1次
注意:選擇健身補劑前最好諮詢專業人士，避免盲目使用。: 1次
流程:預防運動損傷的步驟包括：充分熱身5-10分鐘、循序漸進增加負荷、保持正確動作形式、合理安排訓練和休息: 1次
問題:如果出現持續性肩膀疼痛，應停止加重疼痛的訓練動作，並諮詢醫生或物理治療師。可以暫時進行不引起疼痛的其: 1次
流程:運動後加速恢復的方法包括：適當營養補充（蛋白質和碳水）、充足水分、7-9小時高質量睡眠、主動恢復（如: 1次
事實:運動後建議保證7-9小時高質量睡眠以促進身體恢復。: 1次
注意:傾聽身體信號，避免過度訓練，是預防運動損傷的重要提醒。: 1次
============================================================
示例3: 合併相似知識
過濾前知識點數量: 24
過濾後知識點數量: 16
過濾掉的'需求'和'問題'類型知識點: 8
合併後剩餘 3 個知識點
合併後的知識點:
1. 類型: 流程
內容: 減脂與增肌均需科學結合運動與飲食：減脂時建議每週進行3-5次有氧運動（如跑步、游泳、騎行）和2-3次力量訓練，通
過控制總熱量攝入實現熱量赤字，同時增加蛋白質比例、減少精製碳水和添加糖；增肌則應每週進行3-5次力量訓練，專注深蹲、卧推
、硬拉、引體向上等複合動作，每日熱量盈餘300-500卡路里，並確保每公斤體重攝入1.6-2.2克蛋白質。無論目標是減脂還是增肌，
都應遵循均衡飲食結構，例如使用‘盤子法則’——一半盤子放蔬菜，四分之一放蛋白質，四分之一放全穀物。為預防運動損傷，需充分
熱身5-10分鐘、保持正確動作形式、循序漸進增加負荷、合理安排訓練與休息；運動後應注重恢復，包括適當營養補充（蛋白質+碳水
）、充足水分、7-9小時高質量睡眠、主動恢復（如散步、瑜伽）、拉伸及泡沫軸放鬆，必要時可採用冰敷或熱敷促進恢復。
頻率: 5次
置信度: 0.96
分類: 健身營養與訓練計劃
關鍵詞: ['減脂', '增肌', '運動', '飲食', '熱量赤字', '熱量盈餘', '蛋白質攝入', '複合動作', '盤子法則', '預防運動
損傷', '恢復', '熱身', '動作形式', '拉伸', '泡沫軸', '睡眠']
來源: ['AI', 'AI', 'AI', 'AI', 'AI']
2. 類型: 事實
內容: 在減脂或增肌過程中，應結合科學的營養策略與訓練安排以實現最佳效果。減脂期間每日蛋白質攝入推薦量為每公斤體重
1.6-2.2克，優質蛋白來源包括雞胸肉、魚、雞蛋、豆製品和乳清蛋白粉；訓練後30-90分鐘內是補充蛋白質的窗口期，建議攝入20-40克優質蛋白質，並配合快速吸收的碳水化合物，有助於肌肉恢復和合成。健康均衡飲食應包含大量蔬菜水果、優質蛋白質、全穀物碳
水化合物、健康脂肪及充足水分，同時晚間可適量攝入複合碳水（如糙米、燕麥、紅薯），但需控制總量並避免臨睡前過量攝入。有
氧運動方面，建議每次持續30-60分鐘、保持中等強度（心率在最大心率的60-70%），可結合HIIT和LISS交替進行提升效果；增肌期間
2. 類型: 事實
內容: 在減脂或增肌過程中，應結合科學的營養策略與訓練安排以實現最佳效果。減脂期間每日蛋白質攝入推薦量為每公斤體重
1.6-2.2克，優質蛋白來源包括雞胸肉、魚、雞蛋、豆製品和乳清蛋白粉；訓練後30-90分鐘內是補充蛋白質的窗口期，建議攝入20-40克優質蛋白質，並配合快速吸收的碳水化合物，有助於肌肉恢復和合成。健康均衡飲食應包含大量蔬菜水果、優質蛋白質、全穀物碳
水化合物、健康脂肪及充足水分，同時晚間可適量攝入複合碳水（如糙米、燕麥、紅薯），但需控制總量並避免臨睡前過量攝入。有
氧運動方面，建議每次持續30-60分鐘、保持中等強度（心率在最大心率的60-70%），可結合HIIT和LISS交替進行提升效果；增肌期間
克優質蛋白質，並配合快速吸收的碳水化合物，有助於肌肉恢復和合成。健康均衡飲食應包含大量蔬菜水果、優質蛋白質、全穀物碳
水化合物、健康脂肪及充足水分，同時晚間可適量攝入複合碳水（如糙米、燕麥、紅薯），但需控制總量並避免臨睡前過量攝入。有
氧運動方面，建議每次持續30-60分鐘、保持中等強度（心率在最大心率的60-70%），可結合HIIT和LISS交替進行提升效果；增肌期間
氧運動方面，建議每次持續30-60分鐘、保持中等強度（心率在最大心率的60-70%），可結合HIIT和LISS交替進行提升效果；增肌期間
每週可進行2-3次低強度有氧（每次20-30分鐘），有助於維持心肺健康和代謝功能，但不宜過量以免影響力量訓練恢復。此外，健身
補劑並非必需，但乳清蛋白粉、肌酸、Omega-3脂肪酸等基礎補劑可輔助訓練效果，維生素D和鎂也可能對運動表現和恢復有益。最後
，保證每天7-9小時高質量睡眠是促進身體恢復的關鍵環節。
頻率: 8次
置信度: 0.98
分類: 綜合運動與營養科學
關鍵詞: ['有氧運動', '蛋白質攝入', '減脂', '增肌', '訓練後補蛋白', '肌肉恢復', '碳水化合物', '健康飲食', '睡眠', '健身補劑', '訓練強度', '時間管理']
來源: ['AI', 'AI', 'AI', 'AI', 'AI', 'AI', 'AI', 'AI']
3. 類型: 注意
內容: 進行健身訓練時，應合理安排有氧運動頻率與強度，避免其干擾力量訓練的恢復，建議每週2-3次、每次20-30分鐘低強度
；同時，傾聽身體信號，識別疲勞或不適，防止過度訓練導致損傷；此外，使用健身補劑前務必諮詢專業人士，以確保安全並考慮個
體差異，避免盲目使用。
頻率: 3次
置信度: 0.95
，保證每天7-9小時高質量睡眠是促進身體恢復的關鍵環節。
頻率: 8次
置信度: 0.98
分類: 綜合運動與營養科學
關鍵詞: ['有氧運動', '蛋白質攝入', '減脂', '增肌', '訓練後補蛋白', '肌肉恢復', '碳水化合物', '健康飲食', '睡眠', '健身補劑', '訓練強度', '時間管理']
來源: ['AI', 'AI', 'AI', 'AI', 'AI', 'AI', 'AI', 'AI']
3. 類型: 注意
內容: 進行健身訓練時，應合理安排有氧運動頻率與強度，避免其干擾力量訓練的恢復，建議每週2-3次、每次20-30分鐘低強度
；同時，傾聽身體信號，識別疲勞或不適，防止過度訓練導致損傷；此外，使用健身補劑前務必諮詢專業人士，以確保安全並考慮個
體差異，避免盲目使用。
頻率: 3次
置信度: 0.95
分類: 綜合運動與營養科學
關鍵詞: ['有氧運動', '蛋白質攝入', '減脂', '增肌', '訓練後補蛋白', '肌肉恢復', '碳水化合物', '健康飲食', '睡眠', '健身補劑', '訓練強度', '時間管理']
來源: ['AI', 'AI', 'AI', 'AI', 'AI', 'AI', 'AI', 'AI']
3. 類型: 注意
內容: 進行健身訓練時，應合理安排有氧運動頻率與強度，避免其干擾力量訓練的恢復，建議每週2-3次、每次20-30分鐘低強度
；同時，傾聽身體信號，識別疲勞或不適，防止過度訓練導致損傷；此外，使用健身補劑前務必諮詢專業人士，以確保安全並考慮個
體差異，避免盲目使用。
頻率: 3次
置信度: 0.95
；同時，傾聽身體信號，識別疲勞或不適，防止過度訓練導致損傷；此外，使用健身補劑前務必諮詢專業人士，以確保安全並考慮個
體差異，避免盲目使用。
頻率: 3次
置信度: 0.95
置信度: 0.95
分類: 訓練注意事項
關鍵詞: ['有氧運動', '恢復干擾', '強度控制', '補劑安全', '專業建議', '個體差異', '身體信號', '過度訓練', '預防', '健身安全']
來源: ['AI', 'AI', 'AI']
核心函數
- extract_knowledge_from_conversation() : 從單次對話中提取知識
- batch_extract_knowledge() : 批量提取知識
- merge_similar_knowledge()：使用大模型合併相似知識點
六、知識庫健康度檢查：量化評估與科學治理
1. 基礎介紹
知識庫不能只建不管，需要一套可量化的指標體系來持續監控其健康狀況，需要定期評估知識庫的質量，否則它會逐漸失效。可以通過自動化腳本計算關鍵指標。
- 核心指標：
- 覆蓋度：知識庫能回答的問題佔所有可能問題的比例。可通過日誌分析未命中查詢來評估。
- 準確率：返回的答案正確的比例。需通過人工抽檢或用户反饋（如“是否解決”）來估算。
- 解決率：用户在看到答案後，不再進一步追問或轉人工的比例。是衡量知識有效性的黃金指標。
- 時效性：知識的新舊程度。定期掃描並標記過期內容（如包含“去年”、“2022年”等時間敏感信息的條款）。
- 使用度：各知識條目的被訪問頻率。低頻知識可能價值低或難以查找，高頻知識則需要重點保障其質量。
- 理論支撐：數據驅動決策。通過dashboard監控這些指標，可以發現知識庫的薄弱環節，從而有針對性地進行優化，分配有限的人力資源。
2. 案例演示
代碼結構
# 知識庫健康度檢查
# 導入依賴庫
import dashscope
import os
import json
import re
from datetime import datetime
# 從環境變量中獲取 API Key
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
# 基於 prompt 生成文本
def get_completion(prompt, model="qwen-turbo-latest"):
messages = [{"role": "user", "content": prompt}]
response = dashscope.Generation.call(
model=model,
messages=messages,
result_format='message',
temperature=0.3,
)
return response.output.choices[0].message.content
class KnowledgeBaseHealthChecker:
def __init__(self, model="qwen-turbo-latest"):
self.model = model
self.health_report = {}
def check_missing_knowledge(self, knowledge_base, test_queries):
"""使用LLM檢查缺少的知識"""
instruction = """
你是一個知識庫完整性檢查專家。請分析給定的測試查詢和知識庫內容，判斷知識庫中是否缺少相關的知識。
檢查標準：
1. 查詢是否能在知識庫中找到相關答案
2. 知識是否完整、準確
3. 是否覆蓋了用户的主要需求
4. 是否存在知識空白
請返回JSON格式：
{
"missing_knowledge": [
{
"query": "測試查詢",
"missing_aspect": "缺少的知識方面",
"importance": "重要性（高/中/低）",
"suggested_content": "建議的知識內容",
"category": "知識分類"
}
],
"coverage_score": "覆蓋率評分(0-1)",
"completeness_analysis": "完整性分析"
}
"""
# 構建知識庫內容摘要
knowledge_summary = []
for chunk in knowledge_base:
knowledge_summary.append(f"ID: {chunk.get('id', 'unknown')} - {chunk.get('content', '')}")
knowledge_text = "\n".join(knowledge_summary)
# 構建測試查詢列表
queries_text = []
for query_info in test_queries:
query = query_info['query']
expected = query_info.get('expected_answer', '')
queries_text.append(f"查詢: {query} | 期望答案: {expected}")
queries_text = "\n".join(queries_text)
prompt = f"""
### 指令 ###
{instruction}
### 知識庫內容 ###
{knowledge_text}
### 測試查詢 ###
{queries_text}
### 分析結果 ###
"""
try:
response = get_completion(prompt, self.model)
# 預處理響應，移除markdown代碼塊格式
if response.startswith('```json'):
response = response[7:]
elif response.startswith('```'):
response = response[3:]
if response.endswith('```'):
response = response[:-3]
result = json.loads(response.strip())
return result
except Exception as e:
print(f"LLM檢查缺少知識失敗: {e}")
return None
def check_outdated_knowledge(self, knowledge_base):
"""使用LLM檢查過期的知識"""
instruction = """
你是一個知識時效性檢查專家。請分析給定的知識內容，判斷是否存在過期或需要更新的信息。
檢查標準：
1. 時間相關信息是否過期（年份、日期、時間範圍）
2. 價格信息是否最新（價格、費用、票價等）
3. 政策規則是否更新（政策、規定、規則等）
4. 活動信息是否有效（活動、節日、特殊安排等）
5. 聯繫方式是否準確（電話、地址、網址等）
6. 技術信息是否過時（版本、技術標準等）
請返回JSON格式：
{
"outdated_knowledge": [
{
"chunk_id": "知識切片ID",
"content": "知識內容",
"outdated_aspect": "過期方面",
"severity": "嚴重程度（高/中/低）",
"suggested_update": "建議更新內容",
"last_verified": "最後驗證時間"
}
],
"freshness_score": "新鮮度評分(0-1)",
"update_recommendations": "更新建議"
}
"""
# 構建知識庫內容
knowledge_text = []
for chunk in knowledge_base:
content = chunk.get('content', '')
chunk_id = chunk.get('id', 'unknown')
last_updated = chunk.get('last_updated', 'unknown')
knowledge_text.append(f"ID: {chunk_id} | 更新時間: {last_updated} | 內容: {content}")
knowledge_text = "\n".join(knowledge_text)
prompt = f"""
### 指令 ###
{instruction}
### 知識庫內容 ###
{knowledge_text}
### 當前時間 ###
{datetime.now().strftime('%Y年%m月%d日')}
### 分析結果 ###
"""
try:
response = get_completion(prompt, self.model)
# 預處理響應，移除markdown代碼塊格式
if response.startswith('```json'):
response = response[7:]
elif response.startswith('```'):
response = response[3:]
if response.endswith('```'):
response = response[:-3]
result = json.loads(response.strip())
return result
except Exception as e:
print(f"LLM檢查過期知識失敗: {e}")
return None
def check_conflicting_knowledge(self, knowledge_base):
"""使用LLM檢查衝突的知識"""
instruction = """
你是一個知識一致性檢查專家。請分析給定的知識庫，找出可能存在衝突或矛盾的信息。
檢查標準：
1. 同一主題的不同説法（地點、名稱、描述等）
2. 價格信息的差異（價格、費用、收費標準等）
3. 時間信息的不一致（營業時間、開放時間、活動時間等）
4. 規則政策的衝突（規定、政策、要求等）
5. 操作流程的差異（步驟、方法、流程等）
6. 聯繫方式的差異（地址、電話、網址等）
請返回JSON格式：
{
"conflicting_knowledge": [
{
"conflict_type": "衝突類型",
"chunk_ids": ["相關切片ID"],
"conflicting_content": ["衝突內容"],
"severity": "嚴重程度（高/中/低）",
"resolution_suggestion": "解決建議"
}
],
"consistency_score": "一致性評分(0-1)",
"conflict_analysis": "衝突分析"
}
"""
# 構建知識庫內容
knowledge_text = []
for chunk in knowledge_base:
content = chunk.get('content', '')
chunk_id = chunk.get('id', 'unknown')
knowledge_text.append(f"ID: {chunk_id} | 內容: {content}")
knowledge_text = "\n".join(knowledge_text)
prompt = f"""
### 指令 ###
{instruction}
### 知識庫內容 ###
{knowledge_text}
### 分析結果 ###
"""
try:
response = get_completion(prompt, self.model)
# 預處理響應，移除markdown代碼塊格式
if response.startswith('```json'):
response = response[7:]
elif response.startswith('```'):
response = response[3:]
if response.endswith('```'):
response = response[:-3]
result = json.loads(response.strip())
return result
except Exception as e:
print(f"LLM檢查衝突知識失敗: {e}")
return None
def calculate_overall_health_score(self, missing_result, outdated_result, conflicting_result):
"""計算整體健康度評分"""
coverage_score = missing_result.get('coverage_score', 0)
freshness_score = outdated_result.get('freshness_score', 0)
consistency_score = conflicting_result.get('consistency_score', 0)
# 加權計算
overall_score = (
coverage_score * 0.4 + # 覆蓋率權重40%
freshness_score * 0.3 + # 新鮮度權重30%
consistency_score * 0.3 # 一致性權重30%
)
return overall_score
def generate_health_report(self, knowledge_base, test_queries):
"""生成完整的健康度報告"""
print("正在檢查知識庫健康度...")
# 1. 檢查缺少的知識
print("1. 檢查缺少的知識...")
missing_result = self.check_missing_knowledge(knowledge_base, test_queries)
# 2. 檢查過期的知識
print("2. 檢查過期的知識...")
outdated_result = self.check_outdated_knowledge(knowledge_base)
# 3. 檢查衝突的知識
print("3. 檢查衝突的知識...")
conflicting_result = self.check_conflicting_knowledge(knowledge_base)
# 4. 計算整體健康度
overall_score = self.calculate_overall_health_score(missing_result, outdated_result, conflicting_result)
# 5. 生成報告
report = {
"overall_health_score": overall_score,
"health_level": self.get_health_level(overall_score),
"missing_knowledge": missing_result,
"outdated_knowledge": outdated_result,
"conflicting_knowledge": conflicting_result,
"recommendations": self.generate_recommendations(missing_result, outdated_result, conflicting_result),
"check_date": datetime.now().isoformat()
}
return report
def get_health_level(self, score):
"""根據評分確定健康等級"""
if score >= 0.8:
return "優秀"
elif score >= 0.6:
return "良好"
elif score >= 0.4:
return "一般"
else:
return "需要改進"
def generate_recommendations(self, missing_result, outdated_result, conflicting_result):
"""生成改進建議"""
recommendations = []
# 基於缺少知識的建議
missing_count = len(missing_result.get('missing_knowledge', []))
if missing_count > 0:
recommendations.append(f"補充{missing_count}個缺少的知識點，提高覆蓋率")
# 基於過期知識的建議
outdated_count = len(outdated_result.get('outdated_knowledge', []))
if outdated_count > 0:
recommendations.append(f"更新{outdated_count}個過期知識點，確保信息時效性")
# 基於衝突知識的建議
conflicting_count = len(conflicting_result.get('conflicting_knowledge', []))
if conflicting_count > 0:
recommendations.append(f"解決{conflicting_count}個知識衝突，提高一致性")
if not recommendations:
recommendations.append("知識庫狀態良好，建議定期維護")
return recommendations
def main():
# 初始化知識庫健康度檢查器
checker = KnowledgeBaseHealthChecker()
print("=== 知識庫健康度檢查示例（健康健身主題） ===\n")
# 示例知識庫（包含一些故意的問題）
knowledge_base = [
{
"id": "kb_001",
"content": "減脂需要創造熱量赤字，建議每週進行3-5次有氧運動和2-3次力量訓練。飲食上控制總熱量攝入，增加蛋白質比例。",
"last_updated": "2024-01-15"
},
{
"id": "kb_002",
"content": "增肌需要漸進式超負荷訓練和熱量盈餘。建議每週進行3-5次力量訓練，專注複合動作如深蹲、卧推、硬拉。",
"last_updated": "2024-01-10"
},
{
"id": "kb_003",
"content": "蛋白質攝入建議：減脂期每公斤體重1.6-2.2克，增肌期每公斤體重1.6-2.2克。", # 故意設置相同數值，可能存在問題
"last_updated": "2023-06-01" # 故意設置為較舊的時間
},
{
"id": "kb_004",
"content": "訓練後30-90分鐘內是補充蛋白質的窗口期，建議攝入20-40克優質蛋白質。",
"last_updated": "2024-01-20"
},
{
"id": "kb_005",
"content": "有氧運動建議每次30-60分鐘，保持中等強度（心率在最大心率的60-70%）。",
"last_updated": "2024-01-10"
},
{
"id": "kb_006",
"content": "健康均衡飲食應包含大量蔬菜和水果、優質蛋白質、全穀物碳水化合物和健康脂肪。",
"last_updated": "2024-02-01"
},
{
"id": "kb_007",
"content": "乳清蛋白粉是方便的訓練後補充選擇，但並非必需品。基礎補劑還包括肌酸和Omega-3脂肪酸。",
"last_updated": "2023-11-01" # 較舊的時間
},
{
"id": "kb_008",
"content": "預防運動損傷的關鍵包括：充分熱身、循序漸進增加負荷、保持正確動作形式、合理安排訓練和休息。",
"last_updated": "2024-01-05"
},
{
"id": "kb_009",
"content": "減脂期間晚上不應攝入碳水化合物，以免轉化為脂肪儲存。", # 故意設置可能過時或不準確的觀點
"last_updated": "2024-01-15"
}
]
# 測試查詢
test_queries = [
{
"query": "如何有效減脂？",
"expected_answer": "熱量赤字、有氧和力量訓練結合"
},
{
"query": "增肌需要怎麼做？",
"expected_answer": "漸進超負荷、熱量盈餘、足夠蛋白質"
},
{
"query": "蛋白質應該吃多少？",
"expected_answer": "根據目標每公斤體重1.6-2.2克"
},
{
"query": "訓練後應該補充什麼？",
"expected_answer": "蛋白質和快速吸收碳水"
},
{
"query": "有氧運動做多久？",
"expected_answer": "30-60分鐘中等強度"
},
{
"query": "什麼是健康飲食？", # 知識庫中有相關信息
"expected_answer": "多樣化、蔬菜水果、優質蛋白、全穀物"
},
{
"query": "健身補劑有必要嗎？", # 知識庫中有相關信息
"expected_answer": "不是必需品但可以輔助"
},
{
"query": "如何預防運動損傷？", # 知識庫中有相關信息
"expected_answer": "熱身、正確形式、循序漸進"
},
{
"query": "運動後肌肉痠痛怎麼辦？", # 知識庫中沒有相關信息
"expected_answer": "恢復方法"
},
{
"query": "生酮飲食適合減脂嗎？", # 知識庫中沒有相關信息
"expected_answer": "生酮飲食評價"
},
{
"query": "瑜伽對健身有什麼好處？", # 知識庫中沒有相關信息
"expected_answer": "瑜伽益處"
}
]
# 生成健康度報告
health_report = checker.generate_health_report(knowledge_base, test_queries)
# 顯示報告
print("=== 知識庫健康度報告 ===\n")
print(f"整體健康度評分: {health_report['overall_health_score']:.2f}")
print(f"健康等級: {health_report['health_level']}")
print(f"檢查時間: {health_report['check_date']}")
print("\n" + "="*60 + "\n")
# 詳細分析
print("=== 詳細分析 ===\n")
# 1. 缺少的知識
print("1. 缺少的知識分析:")
missing = health_report['missing_knowledge']
print(f" 覆蓋率: {health_report['missing_knowledge']['coverage_score']*100:.1f}%")
print(f" 缺少知識點數量: {len(missing['missing_knowledge'])}")
for i, item in enumerate(missing['missing_knowledge'][:3], 1):
print(f" {i}. 查詢: {item['query']}")
print(f" 缺少方面: {item['missing_aspect']}")
print(f" 重要性: {item['importance']}")
print("\n" + "-"*40 + "\n")
# 2. 過期的知識
print("2. 過期的知識分析:")
outdated = health_report['outdated_knowledge']
print(f" 新鮮度評分: {outdated['freshness_score']:.2f}")
print(f" 過期知識點數量: {len(outdated['outdated_knowledge'])}")
for i, item in enumerate(outdated['outdated_knowledge'][:3], 1):
print(f" {i}. 切片ID: {item['chunk_id']}")
print(f" 過期方面: {item['outdated_aspect']}")
print(f" 嚴重程度: {item['severity']}")
print("\n" + "-"*40 + "\n")
# 3. 衝突的知識
print("3. 衝突的知識分析:")
conflicting = health_report['conflicting_knowledge']
print(f" 一致性評分: {conflicting['consistency_score']:.2f}")
print(f" 衝突數量: {len(conflicting['conflicting_knowledge'])}")
for i, item in enumerate(conflicting['conflicting_knowledge'][:3], 1):
print(f" {i}. 衝突類型: {item['conflict_type']}")
print(f" 相關切片: {item['chunk_ids']}")
print(f" 嚴重程度: {item['severity']}")
print("\n" + "="*60 + "\n")
# 改進建議
print("=== 改進建議 ===\n")
for i, recommendation in enumerate(health_report['recommendations'], 1):
print(f"{i}. {recommendation}")
if __name__ == "__main__":
main()
輸出結果
=== 知識庫健康度檢查示例（健康健身主題） ===
正在檢查知識庫健康度...
1. 檢查缺少的知識...
2. 檢查過期的知識...
3. 檢查衝突的知識...
=== 知識庫健康度報告 ===
整體健康度評分: 0.78
健康等級: 良好
檢查時間: 2025-09-06T00:54:31.765164
============================================================
=== 詳細分析 ===
1. 缺少的知識分析:
覆蓋率: 70.0%
缺少知識點數量: 3
1. 查詢: 運動後肌肉痠痛怎麼辦？
缺少方面: 恢復方法（如休息、拉伸、冷熱交替、營養補充等）
重要性: 高
2. 查詢: 生酮飲食適合減脂嗎？
缺少方面: 生酮飲食對減脂的科學評價（機制、適用人羣、潛在風險）
重要性: 中
3. 查詢: 瑜伽對健身有什麼好處？
缺少方面: 瑜伽對健身的具體益處（柔韌性、核心穩定性、心理放鬆、損傷預防等）
重要性: 中
----------------------------------------
2. 過期的知識分析:
新鮮度評分: 0.87
過期知識點數量: 1
1. 切片ID: kb_009
過期方面: 飲食建議
嚴重程度: 高
----------------------------------------
3. 衝突的知識分析:
一致性評分: 0.80
衝突數量: 1
1. 衝突類型: 規則政策的衝突
相關切片: ['kb_001', 'kb_009']
嚴重程度: 中
============================================================
=== 改進建議 ===
1. 補充3個缺少的知識點，提高覆蓋率
2. 更新1個過期知識點，確保信息時效性
3. 解決1個知識衝突，提高一致性
七、知識庫版本管理與性能比較
1. 基礎介紹
知識庫的迭代和優化需要像管理代碼一樣科學，避免改錯無法回溯、效果無法衡量。
版本管理理論：
- 每一次更改（增、刪、改） 都對應一個提交，記錄更改人、時間和原因。
- 建立分支（Branch） 進行重大修改或實驗，穩定後再合併（Merge）到主分支（Master）。
- 允許回滾（Rollback） 到任何一個歷史版本。
性能比較（A/B Testing）：
- 理論：將用户流量隨機分為兩組（A組和B組）。A組使用舊版本知識庫（對照組），B組使用新版本知識庫（實驗組）。
- 對比指標：在實驗期間，統計並比較兩組的關鍵指標，如解決率、用户滿意度、平均會話時長等。
- 統計顯著性：使用統計檢驗方法（如t-test）確認指標差異不是由隨機波動引起的，從而科學地判斷新版本是否真正帶來了提升。
價值：實現了知識庫優化的可度量、可比較、可回溯，使知識庫的演進過程從“藝術”變為“科學”。
2. 案例演示
代碼結構
# 知識庫版本管理與性能比較
# 導入依賴庫
import dashscope
import os
import json
import re
from datetime import datetime, timedelta
from collections import defaultdict, Counter
import pandas as pd
import numpy as np
import faiss
from openai import OpenAI
# 從環境變量中獲取 API Key
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
# 初始化百鍊兼容的 OpenAI 客户端
client = OpenAI(
api_key=dashscope.api_key,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 全局配置
TEXT_EMBEDDING_MODEL = "text-embedding-v4"
TEXT_EMBEDDING_DIM = 1024
# 基於 prompt 生成文本
def get_completion(prompt, model="qwen-turbo-latest"):
messages = [{"role": "user", "content": prompt}]
response = dashscope.Generation.call(
model=model,
messages=messages,
result_format='message',
temperature=0.3,
)
return response.output.choices[0].message.content
def get_text_embedding(text):
"""獲取文本的 Embedding"""
response = client.embeddings.create(
model=TEXT_EMBEDDING_MODEL,
input=text,
dimensions=TEXT_EMBEDDING_DIM
)
return response.data[0].embedding
class KnowledgeBaseVersionManager:
def __init__(self, model="qwen-turbo-latest"):
self.model = model
self.versions = {}
def create_version(self, knowledge_base, version_name, description=""):
"""創建知識庫版本"""
# 構建向量索引
metadata_store, text_index = self.build_vector_index(knowledge_base)
version_info = {
"version_name": version_name,
"description": description,
"created_date": datetime.now().isoformat(),
"knowledge_base": knowledge_base,
"metadata_store": metadata_store,
"text_index": text_index,
"statistics": self.calculate_version_statistics(knowledge_base)
}
self.versions[version_name] = version_info
return version_info
def build_vector_index(self, knowledge_base):
"""構建向量索引"""
metadata_store = []
text_vectors = []
for i, chunk in enumerate(knowledge_base):
content = chunk.get('content', '')
if not content.strip():
continue
metadata = {
"id": i,
"content": content,
"chunk_id": chunk.get('id', f'chunk_{i}')
}
# 獲取文本embedding
vector = get_text_embedding(content)
text_vectors.append(vector)
metadata_store.append(metadata)
# 創建FAISS索引
text_index = faiss.IndexFlatL2(TEXT_EMBEDDING_DIM)
text_index_map = faiss.IndexIDMap(text_index)
if text_vectors:
text_ids = [m["id"] for m in metadata_store]
text_index_map.add_with_ids(np.array(text_vectors).astype('float32'), np.array(text_ids))
return metadata_store, text_index_map
def calculate_version_statistics(self, knowledge_base):
"""計算版本統計信息"""
total_chunks = len(knowledge_base)
total_content_length = sum(len(chunk.get('content', '')) for chunk in knowledge_base)
return {
"total_chunks": total_chunks,
"total_content_length": total_content_length,
"average_chunk_length": total_content_length / total_chunks if total_chunks > 0 else 0
}
def compare_versions(self, version1_name, version2_name):
"""比較兩個版本的差異"""
if version1_name not in self.versions or version2_name not in self.versions:
return {"error": "版本不存在"}
v1 = self.versions[version1_name]
v2 = self.versions[version2_name]
kb1 = v1['knowledge_base']
kb2 = v2['knowledge_base']
comparison = {
"version1": version1_name,
"version2": version2_name,
"comparison_date": datetime.now().isoformat(),
"changes": self.detect_changes(kb1, kb2),
"statistics_comparison": self.compare_statistics(v1['statistics'], v2['statistics'])
}
return comparison
def detect_changes(self, kb1, kb2):
"""檢測知識庫變化"""
changes = {
"added_chunks": [],
"removed_chunks": [],
"modified_chunks": [],
"unchanged_chunks": []
}
# 創建ID映射
kb1_dict = {chunk.get('id'): chunk for chunk in kb1}
kb2_dict = {chunk.get('id'): chunk for chunk in kb2}
# 檢測新增和刪除
kb1_ids = set(kb1_dict.keys())
kb2_ids = set(kb2_dict.keys())
added_ids = kb2_ids - kb1_ids
removed_ids = kb1_ids - kb2_ids
common_ids = kb1_ids & kb2_ids
# 記錄新增的知識切片
for chunk_id in added_ids:
changes["added_chunks"].append({
"id": chunk_id,
"content": kb2_dict[chunk_id].get('content', '')
})
# 記錄刪除的知識切片
for chunk_id in removed_ids:
changes["removed_chunks"].append({
"id": chunk_id,
"content": kb1_dict[chunk_id].get('content', '')
})
# 檢測修改的知識切片
for chunk_id in common_ids:
chunk1 = kb1_dict[chunk_id]
chunk2 = kb2_dict[chunk_id]
if chunk1.get('content') != chunk2.get('content'):
changes["modified_chunks"].append({
"id": chunk_id,
"old_content": chunk1.get('content', ''),
"new_content": chunk2.get('content', '')
})
else:
changes["unchanged_chunks"].append(chunk_id)
return changes
def compare_statistics(self, stats1, stats2):
"""比較統計信息"""
comparison = {}
for key in stats1.keys():
if key in stats2:
if isinstance(stats1[key], (int, float)):
comparison[key] = {
"version1": stats1[key],
"version2": stats2[key],
"difference": stats2[key] - stats1[key],
"percentage_change": ((stats2[key] - stats1[key]) / stats1[key] * 100) if stats1[key] != 0 else 0
}
elif isinstance(stats1[key], dict):
comparison[key] = self.compare_dict_statistics(stats1[key], stats2[key])
return comparison
def compare_dict_statistics(self, dict1, dict2):
"""比較字典類型的統計信息"""
comparison = {}
all_keys = set(dict1.keys()) | set(dict2.keys())
for key in all_keys:
val1 = dict1.get(key, 0)
val2 = dict2.get(key, 0)
comparison[key] = {
"version1": val1,
"version2": val2,
"difference": val2 - val1
}
return comparison
def evaluate_version_performance(self, version_name, test_queries):
"""評估版本性能"""
if version_name not in self.versions:
return {"error": "版本不存在"}
performance_metrics = {
"version_name": version_name,
"evaluation_date": datetime.now().isoformat(),
"query_results": [],
"overall_metrics": {}
}
total_queries = len(test_queries)
correct_answers = 0
response_times = []
for query_info in test_queries:
query = query_info['query']
expected_answer = query_info.get('expected_answer', '')
# 使用embedding檢索
start_time = datetime.now()
retrieved_chunks = self.retrieve_relevant_chunks(query, version_name)
end_time = datetime.now()
response_time = (end_time - start_time).total_seconds()
response_times.append(response_time)
# 評估檢索質量
is_correct = self.evaluate_retrieval_quality(query, retrieved_chunks, expected_answer)
if is_correct:
correct_answers += 1
performance_metrics["query_results"].append({
"query": query,
"retrieved_chunks": len(retrieved_chunks),
"response_time": response_time,
"is_correct": is_correct
})
# 計算整體指標
accuracy = correct_answers / total_queries if total_queries > 0 else 0
avg_response_time = sum(response_times) / len(response_times) if response_times else 0
performance_metrics["overall_metrics"] = {
"accuracy": accuracy,
"avg_response_time": avg_response_time,
"total_queries": total_queries,
"correct_answers": correct_answers
}
return performance_metrics
def retrieve_relevant_chunks(self, query, version_name, k=3):
"""使用embedding和faiss檢索相關知識切片"""
if version_name not in self.versions:
return []
version_info = self.versions[version_name]
metadata_store = version_info['metadata_store']
text_index = version_info['text_index']
# 獲取查詢的embedding
query_vector = np.array([get_text_embedding(query)]).astype('float32')
# 使用faiss進行檢索
distances, indices = text_index.search(query_vector, k)
relevant_chunks = []
for i, doc_id in enumerate(indices[0]):
if doc_id != -1: # faiss返回-1表示沒有找到匹配
# 通過ID在元數據中查找
match = next((item for item in metadata_store if item["id"] == doc_id), None)
if match:
# 構造返回的知識切片格式
chunk = {
"id": match["chunk_id"],
"content": match["content"],
"similarity_score": 1.0 / (1.0 + distances[0][i]) # 將距離轉換為相似度
}
relevant_chunks.append(chunk)
return relevant_chunks
def evaluate_retrieval_quality(self, query, retrieved_chunks, expected_answer):
"""評估檢索質量"""
if not retrieved_chunks:
return False
# 簡化的質量評估
for chunk in retrieved_chunks:
content = chunk.get('content', '').lower()
if expected_answer.lower() in content:
return True
return False
def compare_version_performance(self, version1_name, version2_name, test_queries):
"""比較兩個版本的性能"""
perf1 = self.evaluate_version_performance(version1_name, test_queries)
perf2 = self.evaluate_version_performance(version2_name, test_queries)
if "error" in perf1 or "error" in perf2:
return {"error": "版本評估失敗"}
comparison = {
"version1": version1_name,
"version2": version2_name,
"comparison_date": datetime.now().isoformat(),
"performance_comparison": {
"accuracy": {
"version1": perf1["overall_metrics"]["accuracy"],
"version2": perf2["overall_metrics"]["accuracy"],
"improvement": perf2["overall_metrics"]["accuracy"] - perf1["overall_metrics"]["accuracy"]
},
"response_time": {
"version1": perf1["overall_metrics"]["avg_response_time"],
"version2": perf2["overall_metrics"]["avg_response_time"],
"improvement": perf1["overall_metrics"]["avg_response_time"] - perf2["overall_metrics"]["avg_response_time"]
}
},
"recommendation": self.generate_performance_recommendation(perf1, perf2)
}
return comparison
def generate_performance_recommendation(self, perf1, perf2):
"""生成性能建議"""
acc1 = perf1["overall_metrics"]["accuracy"]
acc2 = perf2["overall_metrics"]["accuracy"]
time1 = perf1["overall_metrics"]["avg_response_time"]
time2 = perf2["overall_metrics"]["avg_response_time"]
if acc2 > acc1 and time2 <= time1:
return f"推薦使用版本2，準確率提升{(acc2-acc1)*100:.1f}%，響應時間{'提升' if time2 < time1 else '相當'}"
elif acc2 > acc1 and time2 > time1:
return f"版本2準確率更高但響應時間較長，需要權衡"
elif acc2 < acc1 and time2 < time1:
return f"版本2響應更快但準確率較低，需要權衡"
else:
return f"推薦使用版本1，性能更優"
def generate_regression_test(self, version_name, test_queries):
"""生成迴歸測試"""
if version_name not in self.versions:
return {"error": "版本不存在"}
regression_results = {
"version_name": version_name,
"test_date": datetime.now().isoformat(),
"test_results": [],
"pass_rate": 0
}
passed_tests = 0
total_tests = len(test_queries)
for query_info in test_queries:
query = query_info['query']
expected_answer = query_info.get('expected_answer', '')
# 執行測試
retrieved_chunks = self.retrieve_relevant_chunks(query, version_name)
is_passed = self.evaluate_retrieval_quality(query, retrieved_chunks, expected_answer)
if is_passed:
passed_tests += 1
regression_results["test_results"].append({
"query": query,
"expected": expected_answer,
"retrieved": len(retrieved_chunks),
"passed": is_passed
})
regression_results["pass_rate"] = passed_tests / total_tests if total_tests > 0 else 0
return regression_results
def main():
# 初始化版本管理器
version_manager = KnowledgeBaseVersionManager()
print("=== 知識庫版本管理與性能比較示例（健康健身知識） ===\n")
# 創建版本1（基礎版本）
knowledge_base_v1 = [
{
"id": "kb_001",
"content": "減脂需要創造熱量赤字，建議每週進行3-5次有氧運動和2-3次力量訓練。",
"category": "減脂"
},
{
"id": "kb_002",
"content": "增肌需要漸進式超負荷訓練和熱量盈餘，建議每週進行3-5次力量訓練。",
"category": "增肌"
},
{
"id": "kb_003",
"content": "蛋白質攝入建議：減脂期每公斤體重1.6-2.2克，增肌期每公斤體重1.6-2.2克。",
"category": "營養"
}
]
# 創建版本2（增強版本）
knowledge_base_v2 = [
{
"id": "kb_001",
"content": "減脂需要創造熱量赤字，建議每週進行3-5次有氧運動（如跑步、游泳）和2-3次力量訓練。飲食上控制總熱量攝入，增加蛋白質比例，減少精製碳水和添加糖。",
"category": "減脂"
},
{
"id": "kb_002",
"content": "增肌需要漸進式超負荷訓練和熱量盈餘。建議每週進行3-5次力量訓練，專注複合動作如深蹲、卧推、硬拉和引體向上。飲食上每日熱量盈餘300-500卡路里。",
"category": "增肌"
},
{
"id": "kb_003",
"content": "蛋白質攝入建議：減脂期每公斤體重1.6-2.2克，增肌期每公斤體重1.6-2.2克。優質蛋白來源包括雞胸肉、魚、雞蛋、豆製品和乳清蛋白粉。",
"category": "營養"
},
{
"id": "kb_004",
"content": "訓練後30-90分鐘內是補充蛋白質的窗口期，建議攝入20-40克優質蛋白質，配合快速吸收的碳水化合物，有助於肌肉恢復和合成。",
"category": "訓練"
},
{
"id": "kb_005",
"content": "有氧運動建議每次30-60分鐘，保持中等強度（心率在最大心率的60-70%）。最好結合高強度間歇訓練(HIIT)和穩態有氧(LISS)交替進行。",
"category": "訓練"
},
{
"id": "kb_006",
"content": "健康均衡飲食應包含多樣化的食物：大量蔬菜和水果、優質蛋白質來源、全穀物碳水化合物、健康脂肪以及充足的水分。",
"category": "營養"
}
]
# 功能1: 創建版本
print("功能1: 創建知識庫版本")
v1_info = version_manager.create_version(knowledge_base_v1, "v1.0", "基礎版本")
v2_info = version_manager.create_version(knowledge_base_v2, "v2.0", "增強版本")
print(f"版本1信息:")
print(f" 版本名: {v1_info['version_name']}")
print(f" 描述: {v1_info['description']}")
print(f" 知識切片數量: {v1_info['statistics']['total_chunks']}")
print(f" 平均切片長度: {v1_info['statistics']['average_chunk_length']:.0f}字符")
print(f"\n版本2信息:")
print(f" 版本名: {v2_info['version_name']}")
print(f" 描述: {v2_info['description']}")
print(f" 知識切片數量: {v2_info['statistics']['total_chunks']}")
print(f" 平均切片長度: {v2_info['statistics']['average_chunk_length']:.0f}字符")
print("\n" + "="*60 + "\n")
# 功能示例2: 版本比較
print("功能2: 版本差異比較")
comparison = version_manager.compare_versions("v1.0", "v2.0")
print(f"版本比較結果:")
changes = comparison['changes']
print(f" 新增知識切片: {len(changes['added_chunks'])}個")
print(f" 刪除知識切片: {len(changes['removed_chunks'])}個")
print(f" 修改知識切片: {len(changes['modified_chunks'])}個")
print(f"\n新增的知識切片:")
for i, chunk in enumerate(changes['added_chunks'], 1):
print(f" {i}. ID: {chunk['id']}")
print(f" 內容: {chunk['content']}")
print(f"\n修改的知識切片:")
for i, chunk in enumerate(changes['modified_chunks'], 1):
print(f" {i}. ID: {chunk['id']}")
print(f" 舊內容: {chunk['old_content']}")
print(f" 新內容: {chunk['new_content']}")
print("\n" + "="*60 + "\n")
# 功能3: 性能評估
print("功能3: 版本性能評估")
test_queries = [
{"query": "如何有效減脂？", "expected_answer": "熱量赤字"},
{"query": "增肌需要怎麼做？", "expected_answer": "漸進超負荷"},
{"query": "蛋白質應該吃多少？", "expected_answer": "每公斤體重"},
{"query": "訓練後應該補充什麼？", "expected_answer": "蛋白質"},
{"query": "有氧運動做多久？", "expected_answer": "30-60分鐘"},
{"query": "什麼是健康飲食？", "expected_answer": "多樣化"},
{"query": "如何預防運動損傷？", "expected_answer": "熱身"}
]
perf_v1 = version_manager.evaluate_version_performance("v1.0", test_queries)
perf_v2 = version_manager.evaluate_version_performance("v2.0", test_queries)
print(f"版本1性能:")
print(f" 準確率: {perf_v1['overall_metrics']['accuracy']*100:.1f}%")
print(f" 平均響應時間: {perf_v1['overall_metrics']['avg_response_time']*1000:.1f}ms")
print(f"\n版本2性能:")
print(f" 準確率: {perf_v2['overall_metrics']['accuracy']*100:.1f}%")
print(f" 平均響應時間: {perf_v2['overall_metrics']['avg_response_time']*1000:.1f}ms")
print("\n" + "="*60 + "\n")
# 功能4: 性能比較
print("功能4: 性能比較與建議")
perf_comparison = version_manager.compare_version_performance("v1.0", "v2.0", test_queries)
print(f"性能比較結果:")
comp = perf_comparison['performance_comparison']
print(f" 準確率提升: {comp['accuracy']['improvement']*100:.1f}%")
print(f" 響應時間變化: {comp['response_time']['improvement']*1000:.1f}ms")
print(f" 建議: {perf_comparison['recommendation']}")
print("\n" + "="*60 + "\n")
# 功能5: 迴歸測試
print("功能5: 迴歸測試")
regression_v2 = version_manager.generate_regression_test("v2.0", test_queries)
print(f"迴歸測試結果:")
print(f" 測試通過率: {regression_v2['pass_rate']*100:.1f}%")
print(f" 測試用例數量: {len(regression_v2['test_results'])}")
print(f"\n詳細測試結果:")
for i, result in enumerate(regression_v2['test_results'], 1):
status = "✓" if result['passed'] else "✗"
print(f" {i}. {result['query']} {status}")
if __name__ == "__main__":
main()
輸出結果
=== 知識庫版本管理與性能比較示例（健康健身知識） ===
功能1: 創建知識庫版本
版本1信息:
版本名: v1.0
描述: 基礎版本
知識切片數量: 3
平均切片長度: 37字符
版本2信息:
版本名: v2.0
描述: 增強版本
知識切片數量: 6
平均切片長度: 67字符
============================================================
功能2: 版本差異比較
版本比較結果:
新增知識切片: 3個
刪除知識切片: 0個
修改知識切片: 3個
新增的知識切片:
1. ID: kb_004
內容: 訓練後30-90分鐘內是補充蛋白質的窗口期，建議攝入20-40克優質蛋白質，配合快速吸收的碳水化合物，有助於肌肉恢復
和合成。
2. ID: kb_005
內容: 有氧運動建議每次30-60分鐘，保持中等強度（心率在最大心率的60-70%）。最好結合高強度間歇訓練(HIIT)和穩態有氧(LISS)交替進行。
3. ID: kb_006
內容: 健康均衡飲食應包含多樣化的食物：大量蔬菜和水果、優質蛋白質來源、全穀物碳水化合物、健康脂肪以及充足的水分。
修改的知識切片:
1. ID: kb_002
舊內容: 增肌需要漸進式超負荷訓練和熱量盈餘，建議每週進行3-5次力量訓練。
新內容: 增肌需要漸進式超負荷訓練和熱量盈餘。建議每週進行3-5次力量訓練，專注複合動作如深蹲、卧推、硬拉和引體向上
。飲食上每日熱量盈餘300-500卡路里。
2. ID: kb_001
舊內容: 減脂需要創造熱量赤字，建議每週進行3-5次有氧運動和2-3次力量訓練。
新內容: 減脂需要創造熱量赤字，建議每週進行3-5次有氧運動（如跑步、游泳）和2-3次力量訓練。飲食上控制總熱量攝入，增
加蛋白質比例，減少精製碳水和添加糖。
3. ID: kb_003
舊內容: 蛋白質攝入建議：減脂期每公斤體重1.6-2.2克，增肌期每公斤體重1.6-2.2克。
新內容: 蛋白質攝入建議：減脂期每公斤體重1.6-2.2克，增肌期每公斤體重1.6-2.2克。優質蛋白來源包括雞胸肉、魚、雞蛋、
豆製品和乳清蛋白粉。
============================================================
功能3: 版本性能評估
版本1性能:
準確率: 42.9%
平均響應時間: 134.8ms
版本2性能:
準確率: 71.4%
平均響應時間: 128.3ms
============================================================
功能4: 性能比較與建議
性能比較結果:
準確率提升: 28.6%
響應時間變化: -4.9ms
建議: 版本2準確率更高但響應時間較長，需要權衡
============================================================
功能5: 迴歸測試
迴歸測試結果:
測試通過率: 71.4%
測試用例數量: 7
詳細測試結果:
1. 如何有效減脂？ ✓
2. 增肌需要怎麼做？ ✗
3. 蛋白質應該吃多少？ ✓
4. 訓練後應該補充什麼？ ✓
5. 有氧運動做多久？ ✓
6. 什麼是健康飲食？ ✓
7. 如何預防運動損傷？ ✗
八、總結
現代知識庫處理是一個融合了信息檢索、自然語言處理、機器學習和大數據技術的綜合性工程。它不再是後台的靜態數據，而是走向前台的、驅動企業智能化的核心生產力工具。
未來的知識庫將向着更智能的方向演進：
- 多模態知識庫：融合文本、圖片、表格、視頻等多種形式的知識，並提供統一的多模態檢索與生成能力。
- 個性化知識推薦：根據用户的身份、歷史行為、當前上下文，提供動態組合和定製的個性化答案。
- 自動化與自主進化：進一步減少人工干預，實現從知識挖掘、質量評估到優化入庫的更高程度的自動化，最終形成一個能夠自主演進的“知識大腦”。
通過系統性地應用上述理論與技術，企業可以構建出不僅能説會道，更善解人意、並能持續成長的下一代知識基礎設施，最終在激烈的競爭中贏得智能化的優勢。
簡單來説，智能時代的知識庫管理，就是利用AI技術，讓知識庫變成一個能聽、會説、善思考、會學習的活的系統。它的管理重心從“維護內容”轉向了“設計流程和優化體驗”，最終目的是讓知識這個核心資產，真正高效地流動起來，賦能企業的每一個員工和每一次客户交互。