前言

今天，我們要介紹一個強大的技術：RAG (Retrieval-Augmented Generation)，它能讓 LLM 基於您的專有知識庫來回答問題，

簡單的介紹 RAG

RAG 是一種結合「檢索」和「生成」的技術架構

• 檢索 (Retrieval)：從知識庫中找出與問題相關的文檔片段
• 增強 (Augmented)：將檢索到的內容加入到提示詞中
• 生成 (Generation)：讓 LLM 基於這些上下文生成答案

好處

• 即時更新：不需要重新訓練模型就能使用最新資料
• 可追溯性：可以知道答案來自哪些文檔
• 降低幻覺：基於實際文檔回答，減少模型編造內容
• 成本效益：比 Fine-tuning 更經濟實惠

RAG 架構

使用者問題
    ↓
問題向量化 (Embedding)
    ↓
向量資料庫搜尋 (Vector Search)
    ↓
檢索相關文檔
    ↓
構建 Prompt (問題 + 檔)
    ↓
LLM 生成答案
    ↓
回傳給使用者

安裝

uv add boto3 langchain langchain-aws faiss-cpu pypdf

AWS 服務選擇

Amazon Bedrock：提供 LLM 和 Embedding 模型
FAISS：本地向量資料庫（也可以使用 Amazon OpenSearch 或 Pinecone）

開始實作

step 1:

準備知識庫文檔

# 創建示例文檔
documents = [
    "Amazon S3 是一個物件儲存服務，提供產業領先的可擴展性、資料可用性、安全性及效能。",
    "Amazon EC2 提供安全且可調整大小的運算容量，讓您可以在雲端中執行應用程式。",
    "Amazon RDS 讓您能夠在雲端輕鬆設定、操作和擴展關聯式資料庫。",
    "AWS Lambda 是一項無伺服器運算服務，可讓您執行程式碼而無需佈建或管理伺服器。",
    "Amazon Bedrock 是一項全受管服務，透過單一 API 提供來自領先 AI 公司的高效能基礎模型。"
]

step 2:

將文檔轉換成向量

import boto3
from langchain_aws import BedrockEmbeddings
from langchain.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 初始化 Bedrock 客戶端
bedrock_runtime = boto3.client(
    service_name='bedrock-runtime',
    region_name='ap-northeast-1'
)

# 使用 Amazon Titan Embeddings 模型
embeddings = BedrockEmbeddings(
    client=bedrock_runtime,
    model_id="amazon.titan-embed-text-v1"
)

# 將文檔分割成較小的區塊（如果文檔很長）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)

# 創建向量資料庫
vectorstore = FAISS.from_texts(
    texts=documents,
    embedding=embeddings
)

print("✅ 向量資料庫建立完成！")

step 3:

實作檢索功能

def retrieve_relevant_docs(query, k=3):
    """
    根據查詢檢索最相關的文檔
    
    Args:
        query: 使用者問題
        k: 返回前 k 個最相關的文檔
    """
    results = vectorstore.similarity_search(query, k=k)
    return results

# 測試檢索
query = "什麼是無伺服器運算？"
relevant_docs = retrieve_relevant_docs(query)

print(f"問題：{query}\n")
print("檢索到的相關文檔：")
for i, doc in enumerate(relevant_docs, 1):
    print(f"{i}. {doc.page_content}")

step 4 :

整合 LLM 生成答案

import json

def generate_rag_response(query):
    """
    使用 RAG 生成答案
    """
    # 1. 檢索相關文檔
    relevant_docs = retrieve_relevant_docs(query, k=3)
    
    # 2. 構建上下文
    context = "\n\n".join([doc.page_content for doc in relevant_docs])
    
    # 3. 構建 Prompt
    prompt = f"""請根據以下參考資料回答問題。如果參考資料中沒有相關資訊，請誠實地說不知道。

參考資料：
{context}

問題：{query}

答案："""
    
    # 4. 呼叫 Bedrock LLM
    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
        "messages": [
            {
                "role": "user",
                "content": prompt
            }
        ]
    })
    
    response = bedrock_runtime.invoke_model(
        modelId='anthropic.claude-3-sonnet-20240229-v1:0',
        body=body
    )
    
    response_body = json.loads(response['body'].read())
    answer = response_body['content'][0]['text']
    
    return answer, relevant_docs

# 測試 RAG 系統
query = "AWS 有哪些無伺服器服務？"
answer, sources = generate_rag_response(query)

print(f"問題：{query}\n")
print(f"答案：{answer}\n")
print("參考來源：")
for i, doc in enumerate(sources, 1):
    print(f"{i}. {doc.page_content}")

step 5:

使用 LangChain 簡化流程 ::
LangChain 提供了更簡潔的方式來實作 RAG

from langchain.chains import RetrievalQA
from langchain_aws import ChatBedrock

# 初始化 LLM
llm = ChatBedrock(
    client=bedrock_runtime,
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs={"max_tokens": 1000}
)

# 創建 RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    return_source_documents=True
)

# 執行查詢
result = qa_chain.invoke({"query": "Amazon S3 的主要特點是什麼？"})

print(f"答案：{result['result']}\n")
print("參考文檔：")
for doc in result['source_documents']:
    print(f"- {doc.page_content}")