Hi大家好，
這是我參加 iT 邦幫忙鐵人賽的第 1 次挑戰，這次的主題聚焦在結合 Python 爬蟲、RAG（檢索增強生成）與 AI，打造一套 PTT 文章智慧問答系統。在過程中，我會依照每天進度上傳程式碼到 GitHub ，方便大家參考學習。也歡迎留言或來信討論，我的信箱是 gerryearth@gmail.com。
今天我們將實作 Retrieval-Augmented Generation (RAG) 的最後一步: 利用 Gemini 大語言模型回應使用者問題！
我們將實作一個簡單的 API：
以下介紹幾個重點步驟：
prompt_template = """
你是一個根據文章回答問題的助理。以下是PTT相關文章內容:
---
{merge_text}
---
根據以上內容，請用純文字回答這個問題：
{question}
"""
merge_textmerge_text = "\n".join([f"Title:{a.title} - Content:{a.content}" for a in related_articles])
我們使用 LangChain 搭配 Gemini 的 GPT-4o 模型來建立一個「問答鏈」（Question Answering Chain），並用來針對指定文本回答問題：
model = ChatGoogleGenerativeAI(
model="gemini-2.0-flash",
temperature=0,
google_api_key=env_settings.google_api_key
)
ptt_template = PromptTemplate(
input_variables=["merge_text", "question"],
template="""
根據以下PTT的文章內容以純文字回答問題：
---
{merge_text}
---
問題：{question}
"""
)
chain = ptt_template | model
answer = chain.invoke({"merge_text": merge_text, "question": question}).content
最後我們整理並組合成完整的 API 在 article/views.py：
class SearchAPIView(APIView):
@extend_schema(
methods=("POST",),
description="輸入question(問題)與top_k(想查詢的文章片段數)，藉由LLM與向量資料庫得到question、answer(相關回答)、related_articles(相關文章)。",
request=QueryRequestSerializer,
responses=QueryRequestSerializer
)
def post(self, request):
import asyncio
query_request_serializer = QueryRequestSerializer(data=request.data)
if not query_request_serializer.is_valid():
Log.objects.create(level='ERROR', category='user-search', message='查詢參數不合法', )
return Response(query_request_serializer.errors, status=status.HTTP_400_BAD_REQUEST)
question = query_request_serializer.validated_data.get("question")
top_k = query_request_serializer.validated_data.get("top_k")
# 查詢Pinecone embeddings內容
try:
try:
asyncio.get_running_loop()
except RuntimeError:
asyncio.set_event_loop(asyncio.new_event_loop())
vector_store = PineconeVectorStore(
index=Pinecone(
api_key=env_settings.PINECONE_API_KEY
).Index(env_settings.PINECONE_INDEX_NAME),
embedding=GoogleGenerativeAIEmbeddings(model=env_settings.GOOGLE_EMBEDDINGS_MODEL, google_api_key=SecretStr(env_settings.GOOGLE_API_KEY))
)
top_k_results = vector_store.similarity_search_with_score(question, k=top_k, )
except Exception as e:
Log.objects.create(level='ERROR', category='user-search', message=f'查詢Pinecone embeddings內容發生錯誤: {e}',
traceback=traceback.format_exc())
return Response({"error": f"查詢Pinecone embeddings內容發生錯誤: {str(e)}"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR)
# 從資料庫找出文章內容並合併
try:
match_ids = [match[0].metadata['article_id'] for match in top_k_results]
query_request_serializer.related_articles = Article.objects.filter(id__in=match_ids)
merge_text = "\n".join(
[f"Title:{a.title} - Content:{a.content}" for a in query_request_serializer.related_articles])
if len(merge_text) > 128000:
Log.objects.create(level='ERROR', category='user-search', message='回傳文章總字數過長，請嘗試減少top_k')
return Response(
{"error": "回傳文章總字數過長，請嘗試減少top_k"},
status=status.HTTP_400_BAD_REQUEST)
except (KeyError, TypeError) as e:
Log.objects.create(level='ERROR', category='user-search', message=f'從資料庫找出文章內容發生錯誤: {e}',
traceback=traceback.format_exc())
return Response(
{"error": f"從資料庫找出文章內容發生錯誤: {str(e)}"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR)
# 請求 ChatGPT 回答問題
try:
model = ChatGoogleGenerativeAI(
model="gemini-2.0-flash",
temperature=0,
google_api_key=env_settings.GOOGLE_API_KEY,
)
ptt_template = PromptTemplate(
input_variables=["merge_text", "question"],
template="""
根據以下PTT的文章內容以純文字回答問題：
---
{merge_text}
---
問題：{question}
"""
)
chain = ptt_template | model
answer = chain.invoke({"merge_text": merge_text, "question": question}).content
except Exception as e:
Log.objects.create(level='ERROR', category='user-search', message=f'請求ChatGPT回答發生錯誤: {e}',
traceback=traceback.format_exc())
return Response({"error": f"請求ChatGPT回答問題發生錯誤: {str(e)}"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR)
try:
serializer = QueryRequestSerializer(instance={
"question": question,
"answer": answer,
"related_articles": Article.objects.filter(id__in=match_ids),
})
return Response(serializer.data, status=status.HTTP_200_OK)
except Exception as e:
Log.objects.create(level='ERROR', category='user-search', message=f'序列化輸出資料失敗: {e}',
traceback=traceback.format_exc())
return Response({'error': '序列化輸出資料失敗'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
相應套件:
from langchain_google_genai import ChatGoogleGenerativeAI,GoogleGenerativeAIEmbeddings from env_settings import EnvSettings from langchain_pinecone import PineconeVectorStore from pinecone import Pinecone from langchain_core.prompts import PromptTemplate from pydantic import SecretStr from .serializers import QueryRequestSerializer env_settings = EnvSettings()
Serializer在 article/serializers.py 新增:
class QueryRequestSerializer(serializers.Serializer):
question = serializers.CharField(help_text="查詢內容", required=True, max_length=100, min_length=1)
top_k = serializers.IntegerField(help_text="控制段落的查詢數量 (預設 3)", default=3, write_only=True, min_value=1, max_value=10)
answer = serializers.CharField(required=False, read_only=True)
related_articles = ArticleSerializer(many=True, read_only=True)
在 urls.py 新增 search API 端點:
urlpatterns = [
...
path('search/', views.SearchAPIView.as_view(), name='search'),
]
使用者輸入問題：
輝達股票
Gemini 回覆：
根據文章內容，關於輝達股票：\n\n* 川普持有輝達股票，截至去年底持有61.5萬至130萬美元的輝達股票。\n* 川普允許輝達恢復對中國銷售人工智慧（AI）晶片，條件是美國政府可抽取15％營收分成。\n* 輝達因政策轉向預計可獲數十億美元收益。\n* 7月15日商務部宣布恢復對華銷售當天，輝達股價首次突破170美元，市值超過4兆美元。\n* 川普近期多次稱讚輝達執行長黃仁勳。\n* 過去兩年他似乎加碼輝達部位。
結果如下:
在明天**【Day 20】封裝語意查詢流程 - 讓程式碼更乾淨且可維護**，我們會把語意查詢與生成回答的流程 封裝成一個服務，讓 API 只需要單純呼叫這個函式就能取得結果。這樣可以讓程式碼更好維護。
iThome鐵人賽
看影片追技術