LangChain 없이 구현하는 RAG 예제

앞서 살펴본 LangChain은 편리한 프레임워크이지만 유연성이 떨어집니다. 이 페이지에서는 LangChain을 사용하지 않고, AkasicDB에 직접 접근하여 RAG 응용을 개발하는 예제를 설명합니다. 임베딩과 생성은 외부 LLM/임베딩 API(OpenAI, etc.)를 호출한다고 가정합니다.

데이터베이스 준비

데이터베이스 생성

createdb rag

확장 로드

새로 생성한 rag 라는 데이터베이스에 접속합니다.

psql rag

아래 명령으로 AkasicDB 확장을 로드합니다.

CREATE EXTENSION IF NOT EXISTS akasicdb;

SELECT akasicdb_admin.initialize();

• AkasicDB의 모든 타입과 함수를 데이터베이스에 등록합니다.
• 이 과정은 관리자 권한을 가진 계정으로만 수행할 수 있습니다.

테이블 스키마

아래와 같이 테이블과 벡터 인덱스를 생성합니다.

CREATE TABLE IF NOT EXISTS documents (
  id SERIAL PRIMARY KEY,
  content TEXT NOT NULL,
  embedding vectoron.vector(1536) -- 예: OpenAI text-embedding-3-small 차원
);

CREATE INDEX IF NOT EXISTS idx_documents_embedding
ON documents
USING vectoron (embedding vectoron.vector_l2_ops)
WITH (options = $$
    [indexing.vamana]
    alpha    = 1.2
    r        = 32
    l_search = 64
    phase    = 1
$$);

벡터 인덱스 생성에 대해 자세한 내용은 Indexing을 참고하세요.

---

Python 코드

이 예제에서는 psycopg2를 통해 AkasicDB에 직접 접근합니다.

실행 전 준비

이 예제를 실행하기 위해 Python의 가상환경을 만들어 사용하는 것이 좋습니다.

mkdir rag-without-langchain/
cd rag-without-langchain/
python3 -m venv .venv
source .venv/bin/activate

새로운 가상 환경에 예제를 실행하기 위해 필요한 라이브러리를 설치합니다.

pip install openai psycopg2-binary requests beautifulsoup4

환경 변수 OPENAI_API_KEY를 설정합니다. API Key는 OpenAI 에서 만들 수 있습니다.

export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>

환경변수 DATABASE_URL을 적절한 값으로 변경하거나 아래 코드에서 기본값을 변경합니다.

import os
import textwrap
from typing import Iterable

import argparse
import psycopg2
import requests
from bs4 import BeautifulSoup
from openai import OpenAI
from psycopg2.extras import execute_values

"""
1) AkasicDB 연결 설정: PostgreSQL 과 동일합니다.
"""
DB_DSN = os.getenv("DATABASE_URL", "dbname=rag user=postgres password=postgres host=localhost")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
EMBED_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-5-nano-2025-08-07"

client = OpenAI(api_key=OPENAI_API_KEY)


def get_conn():
    return psycopg2.connect(DB_DSN)

# -----------------------
# 2) 임베딩/생성 모델 호출 (예시 함수)
# -----------------------

def embed_text(text: str) -> list[float]:
    """
    OpenAI Embeddings API 호출로부터 벡터를 반환.

    Args:
        text (str): 임베딩할 텍스트.

    Returns:
        list[float]: 임베딩된 벡터.
    """
    response = client.embeddings.create(
        model=EMBED_MODEL,
        input=text,
    )
    return response.data[0].embedding


def generate_answer(prompt: str) -> str:
    """
    OpenAI Chat Completions API 호출로부터 답변 텍스트를 반환.

    Args:
        prompt (str): 답변을 생성할 프롬프트.

    Returns:
        str: 생성된 답변.
    """
    response = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": prompt},
        ],
        temperature=1,
    )
    return response.choices[0].message.content.strip()


# -----------------------
# 3) 블로그 수집 + 청킹
# -----------------------

def fetch_blog_text(url: str) -> str:
    """
    URL에 주어진 블로그 혹은 웹사이트에서 텍스트를 수집.

    Args:
        url (str): 수집할 블로그 또는 웹사이트의 URL.

    Returns:
        str: 수집된 텍스트.
    """
    response = requests.get(
        url,
        timeout=10,
        headers={"User-Agent": "Mozilla/5.0 (rag-example)"},
    )
    response.raise_for_status()

    soup = BeautifulSoup(response.text, "html.parser")
    for tag in soup(["script", "style", "noscript"]):
        tag.decompose()

    text = soup.get_text(separator=" ", strip=True)
    return " ".join(text.split())


def chunk_text(text: str, chunk_size: int = 800, chunk_overlap: int = 100) -> list[str]:
    """
    텍스트를 청킹.

    Args:
        text (str): 청킹할 텍스트.
        chunk_size (int): 청크 하나의 크기.
        chunk_overlap (int): 청크마다 겹치는 문자 수.

    Returns:
        list[str]: 청크된 텍스트 리스트.
    """
    if chunk_overlap >= chunk_size:
        raise ValueError("chunk_overlap must be smaller than chunk_size")

    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk)
        start += chunk_size - chunk_overlap
    return chunks


# -----------------------
# 4) 문서 저장
# -----------------------

def upsert_documents(docs: list[str]):
    """
    텍스트 청크와 그 임베딩을 데이터베이스에 저장

    Args:
        docs (list[str]): 저장할 텍스트 청크 리스트.
    """
    rows = []
    for doc in docs:
        embedding = embed_text(doc)
        rows.append((doc, embedding))

    with get_conn() as conn:
        with conn.cursor() as cur:
            execute_values(
                cur,
                """
                INSERT INTO documents (content, embedding)
                VALUES %s
                """,
                rows,
                template="(%s, %s::vector)",
            )
        conn.commit()


# -----------------------
# 5) 유사도 검색
# -----------------------

def search_similar(query: str, top_k: int = 3) -> list[str]:
    """
    유사도 검색

    Args:
        query (str): 
        top_k (int): 

    Returns:
        list[str]: 
    """
    query_embedding = embed_text(query)
    query_embedding_literal = "[" + ",".join(str(x) for x in query_embedding) + "]"

    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT content
                FROM documents
                ORDER BY embedding <-> %s::vector
                LIMIT %s
                """,
                (query_embedding_literal, top_k)
            )
            results = cur.fetchall()

    return [row[0] for row in results]


# -----------------------
# 6) RAG 파이프라인
# -----------------------

def rag_answer(query: str) -> str:
    """
    주어진 질문에 대해 RAG 파이프라인을 실행하여 답변을 생성.

    Args:
        query (str): 질문

    Returns:
        str: RAG 파이프라인에서 생성된 답변.
    """
    retrieved_docs = search_similar(query, top_k=3)
    context = "\n\n".join(retrieved_docs)

    prompt = f"""
    다음 문맥을 참고하여 질문에 답하세요.

    [문맥]
    {context}

    [질문]
    {query}

    [답변]
    """

    return generate_answer(prompt)


# -----------------------
# 7) 사용 예시
# -----------------------

def main():
    parser = argparse.ArgumentParser(description="Simple RAG without LangChain")
    parser.add_argument(
        "--ingest",
        action="store_true",
        help="Fetch, chunk, and store documents before answering",
    )
    parser.add_argument(
        "--url",
        default="https://lilianweng.github.io/posts/2023-06-23-agent/",
        help="Source URL to ingest when --ingest is set",
    )
    parser.add_argument(
        "--question",
        default=None,
        help="Question to ask the RAG pipeline (omit to only ingest)",
    )
    parser.add_argument(
        "--chunk-size",
        type=int,
        default=1000,
        help="Chunk size to use during ingestion",
    )
    parser.add_argument(
        "--chunk-overlap",
        type=int,
        default=200,
        help="Chunk overlap to use during ingestion",
    )
    args = parser.parse_args()

    if not args.ingest and not args.question:
        parser.error("No action requested; use --ingest and/or --question.")

    if args.ingest:
        blog_text = fetch_blog_text(args.url)
        chunks = chunk_text(blog_text, chunk_size=args.chunk_size, chunk_overlap=args.chunk_overlap)
        upsert_documents(chunks)

    if args.question:
        answer = rag_answer(args.question)
        print(answer)


if __name__ == "__main__":
    main()

---

---

마무리

• LangChain 없이도 임베딩 저장 + 벡터 검색 + 프롬프트 구성 + 생성을 직접 연결하면 RAG를 구현할 수 있습니다.
• embedding <-> %s는 VectorOn의 제곱 유클리드 거리 연산자입니다.
• 임베딩/생성 모델 호출은 원하는 API로 바꾸면 됩니다.