Azure Database for PostgreSQL 및 Python을 사용하여 RAG 애플리케이션 빌드

  • 7분

이제 데이터베이스에 포함 및 벡터 인덱스가 준비되었으므로 검색을 작업 애플리케이션으로 전환해야 합니다. 목표는 간단합니다. 사용자 질문을 하고, PostgreSQL에서 가장 관련성이 큰 청크를 검색하고, LangChain을 통해 Azure OpenAI를 사용하여 해당 청크에 근거한 답변을 생성합니다.

이 단원에서는 Python에서 RAG 애플리케이션을 빌드하는 기본 사항에 대해 알아봅니다. 애플리케이션은 데이터베이스에 연결하고, 데이터베이스에서 유사성 검색을 실행하고, 명확한 지침이 포함된 모델에 검색 결과를 전달하여 환각을 방지합니다. 모델은 데이터베이스 데이터를 기반으로 잘 정의된 컨텍스트를 수신하므로 해당 컨텍스트에 기반한 보다 정확하고 관련 있는 답변을 생성할 수 있습니다.

PostgreSQL 및 LangChain을 사용한 검색 증강 생성기술

이전 단원을 기억하다시피, RAG 흐름은 검색과 생성을 결합한 일련의 단계입니다. 자세히 살펴보겠습니다.

  1. 사용자가 질문을 합니다.
  2. 데이터베이스는 질문에 대한 포함을 계산하고 벡터 인덱스를 사용하여 데이터베이스에서 가장 가까운 일치 항목(상위 청크)을 찾습니다.
  3. 상위 청크는 다음과 같은 프롬프트와 함께 모델에 전달됩니다. "이 컨텍스트에서만 답변하십시오. 모르면 모른다고 말하세요."
  4. 모델은 제공된 컨텍스트에 따라 자연어 응답을 반환합니다.

LangChain 은 언어 모델을 사용하여 애플리케이션을 빌드하기 위한 프레임워크를 제공합니다. 다양한 데이터 원본에 연결하고, 프롬프트를 관리하고, 응답을 처리하는 프로세스를 간소화합니다. Azure Database for PostgreSQL과 Python 및 LangChain을 결합하여 관련 정보를 검색하고 정확한 응답을 생성하는 강력한 RAG 애플리케이션을 만들 수 있습니다. RAG 애플리케이션이 발전할 수 있지만 이러한 핵심 원칙은 개발을 안내합니다.

이 애플리케이션에서는 LangChain AzureChatOpenAI 래퍼를 사용하여 Azure OpenAI 서비스와 상호 작용합니다.

다음 Python 예제에서는 다음 테이블 구조를 사용합니다.

CREATE TABLE company_policies (
    id SERIAL PRIMARY KEY,
    title TEXT,
    policy_text TEXT,
    embedding VECTOR(1536)
);

이 표에서는 효율적인 유사성 검색을 사용하도록 열에 embedding 벡터 인덱스가 생성된다고 가정합니다.

각 단계에 대한 Python 코드 조각을 검토해 보겠습니다. 대부분의 RAG 애플리케이션은 비슷한 구조를 따릅니다.

데이터베이스에 연결

연결은 짧고 안전하게 유지하세요. 이 예제에서 비밀은 환경 변수에 있으며 연결에 전달됩니다.

import os, psycopg2
from contextlib import contextmanager

@contextmanager
def get_conn():
    conn = psycopg2.connect(
        host=os.getenv("PGHOST"),
        user=os.getenv("PGUSER"),
        password=os.getenv("PGPASSWORD"),
        dbname=os.getenv("PGDATABASE"),
        connect_timeout=10
    )
    try:
        yield conn
    finally:
        conn.close()

이 스크립트는 데이터베이스 연결에 대한 컨텍스트 관리자를 설정하여 사용 후 제대로 닫히도록 합니다. 실제 검색으로 이동할 시간입니다.

관련 청크 검색

사용자가 질문을 하기 때문에 해당 질문으로 데이터베이스를 쿼리하는 함수가 필요합니다. 해당 함수는 질문에 따라 관련 청크를 반환해야 합니다. 이 예제에서는 코사인 거리를 사용하여 벡터 인덱스의 순위를 지정한다고 가정하므로 쿼리에 해당 연산자 클래스(<=>)를 사용합니다.

def retrieve_chunks(question, top_k=5):
    sql = """
    WITH q AS (
        SELECT azure_openai.create_embeddings(%s, %s)::vector AS qvec
    )
    SELECT id, title, policy_text
    FROM company_policies, q
    ORDER BY embedding <=> q.qvec
    LIMIT %s;
    """
    params = (os.getenv("OPENAI_EMBED_DEPLOYMENT"), question, top_k)
    with get_conn() as conn, conn.cursor() as cur:
        cur.execute(sql, params)
        rows = cur.fetchall()
    return [{"id": r[0], "title": r[1], "text": r[2]} for r in rows]

따라서 이 함수는 사용자의 질문에 따라 데이터베이스에서 관련 청크를 검색합니다. 그런 다음, 이러한 청크를 사용하여 후속 단계에서 컨텍스트 인식 답변을 생성합니다.

LangChain을 사용하여 답변 생성

이제 관련 청크를 검색하는 함수가 있으므로 해당 청크를 사용하여 답변을 생성해야 합니다. 모델은 검색된 컨텍스트만 사용하고 정책 제목을 인용해야 합니다. 응답 생성 함수를 만드는 시간입니다.

from langchain_openai import AzureChatOpenAI

SYSTEM_PROMPT = """
You are a helpful assistant. Answer using ONLY the provided context.
If the answer is not in the context, say you don’t have enough information.
Cite policy titles in square brackets, e.g., [Vacation policy].
"""

def format_context(chunks):
    return "\n\n".join([f"[{c['title']}] {c['text']}" for c in chunks])

def generate_answer(question, chunks):
    llm = AzureChatOpenAI(
        azure_deployment=os.getenv("OPENAI_CHAT_DEPLOYMENT"),
        api_key=os.getenv("AZURE_OPENAI_API_KEY"),
        azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        temperature=0
    )
    context = format_context(chunks)
    messages = [
        ("system", SYSTEM_PROMPT),
        ("human", f"Question: {question}\nContext:\n{context}")
    ]
    return llm.invoke(messages).content

비고

AzureChatOpenAI를 사용하려면 다음을 수행합니다.

  • 메시지를 (역할, 콘텐츠) 튜플(또는 메시지 개체) 목록으로 전달합니다.
  • 환경 변수 및 생성자를 통해 Azure 설정을 AZURE_OPENAI_API_KEYAZURE_OPENAI_ENDPOINTazure_deploymentapi_version 제공합니다.
  • 사실 기반 답변에는 temperature=0을 유지합니다. 값이 클수록 창의성이 높아지지만 정확도가 떨어질 수 있습니다.

이 함수는 언어 모델과의 상호 작용을 처리하는 RAG 애플리케이션의 핵심입니다. 검색된 청크가 형식을 지정하고 컨텍스트에 포함되는 방식을 확인합니다. 또한 시스템 프롬프트는 모델이 제공된 컨텍스트를 준수하고 잘못된 AI 생성 응답을 줄이도록 설계되었습니다. 마지막으로, 메시지는 언어 모델에서 처리되어 LangChain호출 메서드를 사용하여 응답을 생성합니다. invoke 메서드는 형식이 지정된 메시지로 호출되고 모델의 응답은 자연어 텍스트로 반환됩니다.

함께 묶기

마지막으로 필요한 것은 전체 흐름을 실행하는 간단한 함수입니다.

def answer_question(question):
    chunks = retrieve_chunks(question)
    if not chunks:
        return "I couldn’t find relevant content in the policy store."
    return generate_answer(question, chunks)

# Quick test
print(answer_question("How many vacation days do employees get?"))

이 예제에서는 데이터베이스에서 관련 청크를 검색하고 이를 사용하여 컨텍스트 인식 답변을 생성합니다. 이 함수 호출은 질문에서 답변 생성으로의 전체 흐름을 보여 줍니다. 먼저 함수를 retrieve_chunks 호출하여 관련 컨텍스트(기본적으로 데이터베이스에서 반환된 행)를 가져옵니다. 그런 다음, 이 함수는 언어 모델과 상호 작용하는 함수에 해당 컨텍스트 generate_answer 를 전달하여 자연어 응답으로 최종 답변(컨텍스트의 일부로 행 사용)을 생성합니다. 전체 흐름은 답변이 검색된 데이터에 근거하도록 하여 보다 정확하고 신뢰할 수 있는 응답을 제공합니다.

주요 내용

실용적인 RAG 애플리케이션은 사용자 질문을 받고, SQL에 포함을 만들고, 벡터 인덱스를 사용하여 데이터베이스에서 가장 가까운 구절을 가져오고, 해당 컨텍스트만 모델에 전달합니다. 환각을 줄이기 위해 모델은 제공된 컨텍스트 내에 머물며 정보가 누락된 경우를 인정하도록 지시됩니다. 연결을 수명이 짧은 상태로 유지하고, 쿼리를 매개 변수화하고, 환경 변수를 통해 비밀을 전달합니다. 응답을 추적할 수 있도록 사실 답변에 낮은 온도를 사용하고 간단한 인용(제목 또는 ID)을 포함합니다.

이제 Azure Database for PostgreSQL 및 Python을 사용하여 RAG(검색 증강 세대) 애플리케이션을 빌드하는 방법을 확실하게 이해해야 합니다. 실제 시나리오에서는 RAG 애플리케이션이 훨씬 더 복잡할 수 있지만 핵심 원칙은 동일하게 유지됩니다.