Hi大家好，
這是我參加 iT 邦幫忙鐵人賽的第 1 次挑戰，這次的主題聚焦在結合 Python 爬蟲、RAG（檢索增強生成）與 AI，打造一套 PTT 文章智慧問答系統。在過程中，我會依照每天進度上傳程式碼到 GitHub ，方便大家參考學習。也歡迎留言或來信討論，我的信箱是 gerryearth@gmail.com。

在昨天，我們已經成功建立了一個智慧問答 API，能夠把使用者的問題送進 RAG 流程，並由 LLM 回答。

今天，我們要將這個流程 封裝成獨立函式，將查詢邏輯簡化為直接調用 run_rag_query，方便管理。

今日目標

• 將 檢索 + LLM 回答 封裝成 run_rag_query
• 讓程式碼更乾淨且可維護

設計 RAG Query Service

我們在 article 建立一個服務類別， rag_query.py，專責處理語意查詢:

import traceback
from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone
from langchain.prompts import PromptTemplate
from langchain_google_genai import GoogleGenerativeAIEmbeddings, ChatGoogleGenerativeAI
from pydantic import SecretStr
import asyncio
from article.models import Article
from log_app.models import Log
from env_settings import EnvSettings

env_settings = EnvSettings()


def run_rag_query(question, top_k):
    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 {"error": f"查詢Pinecone embeddings內容發生錯誤: {str(e)}"}
    try:
        match_ids = [match[0].metadata['article_id'] for match in top_k_results]
        related_articles = Article.objects.filter(id__in=match_ids)
        merge_text = "\n".join(
            [f"Title:{a.title} - Content:{a.content}" for a in related_articles])
        if len(merge_text) > 128000:
            Log.objects.create(level='ERROR', category='user-search', message='回傳文章總字數過長，請嘗試減少top_k')
            return {"error": "回傳文章總字數過長，請嘗試減少top_k"}
    except (KeyError, TypeError) as e:
        Log.objects.create(level='ERROR', category='user-search', message=f'從資料庫找出文章內容發生錯誤: {e}', traceback=traceback.format_exc())
        return {"error": f"從資料庫找出文章內容發生錯誤: {str(e)}"}
    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})
        return {
            "question": question,
            "answer": answer,
            "related_articles": [
                {"id": a.id, "title": a.title, "content": a.content} for a in related_articles
            ]
        }
    except Exception as e:
        Log.objects.create(level='ERROR', category='user-search', message=f'LLM生成回答發生錯誤: {e}', traceback=traceback.format_exc())
        return {"error": f"LLM生成回答發生錯誤: {str(e)}"}

在 View 中使用 RAG Service

views.py

將智慧問答 API 改為以下更為簡潔的內容:

from .rag_query import run_rag_query

class SearchAPIView(APIView):
    @extend_schema(
        methods=("POST",),
        description="輸入question(問題)與top_k(想查詢的文章片段數)，藉由LLM與向量資料庫得到question、answer(相關回答)、related_articles(相關文章)。",
        request=QueryRequestSerializer,
        responses=QueryRequestSerializer
    )
    def post(self, request):
        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")
        result = run_rag_query(question, top_k)
        if "error" in result:
            return Response(result, status=status.HTTP_400_BAD_REQUEST)
        return Response(result)

修改

這裡順便把其他 API 的 OpenApiParameter.QUERY 修改成 'query' 變為更簡潔的程式碼。

下一篇**【Day 21】RAG 的文章切割策略 - 切割長度應該如何決定**，讓系統提供更完整的語意查詢服務，包括返回「相關文章來源」。