AI 記憶與知識壓縮技術

from langchain.memory import ConversationSummaryBufferMemory
from langchain.chat_models import ChatOpenAI

# 建立摘要緩衝記憶體
# llm 參數指定用於生成摘要的大語言模型
summary_buffer_memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(),
    max_token_limit=1000  # 設定記憶體中 token 的最大限制
)

# 模擬一段長對話並儲存上下文
for i in range(10):
    summary_buffer_memory.save_context(
        {"input": f"這是使用者的第{i+1}個問題"},
        {"output": f"這是系統的第{i+1}個回答，提供了一些資訊"}
    )

# 檢視緩衝區中最近的對話內容
print(summary_buffer_memory.buffer)
# 檢視被壓縮的移動摘要
print("\n摘要:")
print(summary_buffer_memory.moving_summary_buffer)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainFilter
from langchain.chat_models import ChatOpenAI

# 假設 vectorstore 已被初始化
# 建立一個基礎的向量檢索器，先找出10個最相似的文件
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# 建立一個 LLM 過濾器，用於判斷文件是否與查詢相關
llm = ChatOpenAI(temperature=0)
filter_prompt = """判斷以下文字片段是否與問題相關:
問題: {query}
文字: {text}
如果相關，回答"相關"；如果不相關，回答"不相關"。
"""
llm_filter = LLMChainFilter.from_llm(
    llm=llm,
    prompt_template=filter_prompt
)

# 建立一個上下文壓縮檢索器
# 它會先使用 base_retriever 檢索文件，然後用 llm_filter 進行過濾
compression_retriever = ContextualCompressionRetriever(
    base_compressor=llm_filter,
    base_retriever=base_retriever
)

# 執行檢索與過濾
filtered_docs = compression_retriever.get_relevant_documents("什麼是向量資料庫的應用場景?")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMQueryTransformer
from langchain.chat_models import ChatOpenAI

# 建立查詢轉換器，用於將原始問題改寫為更有效的搜尋查詢
query_transformer_prompt = """給定以下使用者問題，請重新表述為一個更有效的搜尋查詢，以從向量資料庫中檢索相關文件。
原始問題: {query}
重寫的搜尋查詢:"""

query_transformer = LLMQueryTransformer(
    llm=ChatOpenAI(temperature=0),
    prompt_template=query_transformer_prompt
)

# 建立一個使用查詢轉換的增強檢索器
retriever_with_query_transformation = ContextualCompressionRetriever(
    base_compressor=query_transformer,
    base_retriever=vectorstore.as_retriever()
)

# 使用轉換後的查詢來檢索文件
transformed_docs = retriever_with_query_transformation.get_relevant_documents(
    "AI會取代人類工作嗎?"
)

from langchain.chat_models import ChatOpenAI, ChatAnthropic
from langchain.schema.runnable import RunnablePassthrough, RunnableLambda
from langchain.prompts import ChatPromptTemplate
from operator import itemgetter

# 假設 vectorstore 已被初始化
# 1. 定義檢索函式
def retrieve_documents(query):
    return vectorstore.similarity_search(query, k=5)

# 2. 定義提示範本
prompt_template = ChatPromptTemplate.from_template("""
根據以下資訊回答使用者問題。如果無法從提供的資訊中找到答案，請說明你不知道。

資訊:
{context}

使用者問題: {question}
""")

# 3. 建立不同的 LLM
# retrieval_llm = ChatOpenAI(model_name="gpt-3.5-turbo") # 可選：用於檢索或改寫查詢
generation_llm = ChatAnthropic(model_name="claude-2")   # 用於生成最終答案

# 4. 構建文件格式化函式
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 5. 使用 LangChain Expression Language (LCEL) 組裝 RAG 鏈
rag_chain = (
    # 將問題傳遞下去，同時執行檢索並格式化上下文
    {"question": RunnablePassthrough(), "context": RunnableLambda(retrieve_documents) | format_docs}
    # 將結果傳入提示範本
    | prompt_template
    # 將提示傳入生成模型
    | generation_llm
)

# 6. 執行 RAG 鏈
response = rag_chain.invoke("區塊鏈技術有哪些實際應用?")
print(response)

import numpy as np
from langchain.embeddings import OpenAIEmbeddings

class SemanticCache:
    """一個簡單的語義快取實現"""
    def __init__(self, embedding_model, similarity_threshold=0.95):
        self.embedding_model = embedding_model
        self.cache = {}  # 儲存查詢與結果的對應
        self.query_embeddings = {}  # 儲存查詢的向量
        self.similarity_threshold = similarity_threshold
    
    def get(self, query: str):
        """嘗試從快取中獲取結果"""
        if not self.query_embeddings:
            return None
            
        query_embedding = self.embedding_model.embed_query(query)
        
        # 找出最相似的已快取查詢
        cached_embeddings = np.array(list(self.query_embeddings.values()))
        similarities = self._cosine_similarity_matrix([query_embedding], cached_embeddings)[0]
        
        most_similar_idx = np.argmax(similarities)
        max_similarity = similarities[most_similar_idx]
        
        if max_similarity > self.similarity_threshold:
            most_similar_query = list(self.query_embeddings.keys())[most_similar_idx]
            print(f"快取命中! 相似度: {max_similarity:.4f}, 原查詢: '{most_similar_query}'")
            return self.cache[most_similar_query]
        
        return None
    
    def set(self, query: str, result: str):
        """將新的查詢結果存入快取"""
        self.cache[query] = result
        self.query_embeddings[query] = self.embedding_model.embed_query(query)
    
    def _cosine_similarity_matrix(self, matrix_a, matrix_b):
        """計算兩個矩陣間的餘弦相似度"""
        return np.dot(matrix_a, matrix_b.T) / (np.linalg.norm(matrix_a, axis=1)[:, np.newaxis] * np.linalg.norm(matrix_b, axis=1))

# 模擬 RAG 查詢函式
def query_rag_system(query, cache):
    cached_result = cache.get(query)
    if cached_result:
        return cached_result
    
    print(f"執行完整 RAG 查詢: '{query}'")
    result = f"這是 '{query}' 的查詢結果"
    cache.set(query, result)
    return result

# 使用範例
embeddings = OpenAIEmbeddings()
semantic_cache = SemanticCache(embeddings)

print(query_rag_system("什麼是向量資料庫?", semantic_cache))
print(query_rag_system("向量資料庫是什麼?", semantic_cache))
print(query_rag_system("機器學習模型如何工作?", semantic_cache))

import plotly.graph_objects as go
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

# 檔案樣本
documents = [
    "The sky is blue and beautiful.",
    "Love this blue and beautiful sky!",
    "The quick brown fox jumps over the lazy dog.",
    "A king's breakfast has sausages, ham, bacon, eggs, toast, and beans",
    "I love green eggs, ham, sausages and bacon!",
    "The brown fox is quick and the blue dog is lazy!",
    "The sky is very blue and the sky is very beautiful today",
    "The dog is lazy but the brown fox is quick!"
]

# 使用TF-IDF進行向量化
vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(documents)

# 計算所有文件對之間的餘弦相似度
cosine_similarities = cosine_similarity(X)

# 建立一個互動迴圈，讓使用者查詢文件相似度
while True:
    try:
        # 提示使用者輸入
        user_input = input(f"請輸入文件編號 (0-{len(documents)-1}) 或輸入 'exit' 離開: ").strip()
        
        if user_input.lower() == 'exit':
            break
            
        selected_index = int(user_input)
        if not 0 <= selected_index < len(documents):
            raise ValueError("索引超出範圍")
            
        # 獲取並顯示相似度結果
        similarities = cosine_similarities[selected_index]
        print(f"\n文件 '{documents[selected_index]}' 與其他文件的相似度:")
        for i, score in enumerate(similarities):
            print(f"  - 與文件 {i}: {score:.4f}")
            
    except ValueError as e:
        print(f"輸入無效: {e}。請輸入有效的數字。")