BigQuery와 하이퍼클로바X로 Text-To-SQL(NL2SQL) 구현하기

BigQuery는 Google Cloud에서 제공하는 데이터 웨어하우스로, 대량의 데이터를 빠르게 분석할 수 있다는 장점이 있습니다. BigQuery를 사용할 경우 별도의 DB 설치 필요 없이 클라우드 내에서 데이터베이스를 조작할 수 있고 언어별 SDK도 지원이 잘 되어 있기 때문에 구글 계정 인증만 한다면 코드단에서도 손쉽게 사용이 가능합니다.

 

이번 포스팅에서는 랭체인을 활용하여 BigQuery와 하이퍼클로바X를 연동해 보고 Text-to-SQL 프롬프트를 작성하는 방법에 대해 알아보도록 하겠습니다.

 

목차
1. BigQuery 사용방법 알아보기
2. BigQuery 연동하기
3. 하이퍼클로바X 연동하기
4. Text-to-SQL(NL2SQL) 프롬프트 작성하기
5. 정리하기

 

---

 

🔍 BigQuery 사용방법 알아보기

먼저 BigQuery를 사용하기 위해서는 다음과 같은 절차를 거치게 됩니다.

 

1. Google Cloud 프로젝트 생성
2. 데이터셋 및 테이블 생성
3. 데이터 조회 및 확인
4. BigQuery API 활성화

 

1. Google Cloud 프로젝트 생성하기

먼저 Google Cloud Platform으로 이동하여 BigQuery에 사용할 프로젝트를 하나 생성합니다. Google Cloud 계정이 없으시다면 계정을 먼저 생성해주셔야 합니다. 첫 회원가입 시 3달간 $300달러(약 43만 원) 무료 크레딧이 지급됩니다. 프로젝트를 생성하면 Google Cloud Console 접속 시 다음과 같은 화면이 나타납니다.

 

https://console.cloud.google.com/

 

2. 데이터셋 및 테이블 생성

다음은 BigQuery 스튜디오에 접속하여 새로운 데이터셋과 테이블을 생성하도록 하겠습니다. 구글 클라우드 콘솔의 왼쪽 네비게이션 바를 열어서 BigQuery 메뉴를 클릭합니다.

 

BigQuery 서비스 선택

 

BigQuery 메뉴를 클릭하면 다음과 같이 BigQuery Studio에 접속할 수 있습니다. 먼저 화면 왼쪽의 탐색기에서 프로젝트id 오른쪽의 [더보기 아이콘(:)]을 클릭하여 [데이터 세트 만들기] 메뉴를 클릭합니다.

 

 

 

[데이터 세트 만들기] 메뉴 클릭시 오른쪽에 메타 정보를 입력할 수 있는 Drawer가 열립니다. 원하는 데이터세트ID로 입력해 주신 뒤에 하단의 [데이터 세트 만들기] 버튼을 클릭합니다.

 

 

데이터셋이 정상적으로 만들어진다면 다음과 같이 프로젝트 내부에 새로 만든 데이터셋 ID가 표시됩니다.

 

 

 

다음은 테이블을 생성해보도록 하겠습니다. 아래의 예제 스키마는 간단한 사용자 이벤트를 추적할 수 있는 테이블 구조로 이루어져 있습니다.

 

CREATE TABLE datasets.events (
    event_name STRING,
    event_timestamp TIMESTAMP,
    user_id STRING,
    session_id STRING,
    event_params STRING
);

 

다음과 같이 상단의 [+] 버튼을 통해 쿼리 입력창을 열어서 위에서 정의한 DDL(Data Definition Language) 문을 복사 붙여넣기한 후 [실행] 버튼을 클릭합니다. 쿼리가 잘 실행된다면 하단 쿼리 결과창에 테이블이 생성되었다는 문구가 나타납니다.

 

 

다음으로는 데이터를 저장하도록 하겠습니다. 아래 DML(Data Manipulation Language) 쿼리를 복사해서 쿼리 입력창에 추가합니다. 

 

INSERT INTO datasets.events (event_name, event_timestamp, user_id, session_id, event_params)
VALUES
    ("page_view", TIMESTAMP("2025-02-23 12:00:00"), "user_001", "session_001", '{"page":"home","referrer":"google"}'),
    ("page_view", TIMESTAMP("2025-02-23 12:10:00"), "user_002", "session_002", '{"page":"donation","referrer":"facebook"}'),
    ("page_view", TIMESTAMP("2025-02-23 12:20:00"), "user_003", "session_003", '{"page":"home","referrer":"direct"}'),
    ("donation", TIMESTAMP("2025-02-23 12:25:00"), "user_004", "session_004", '{"campaign":"SaveTheEarth","donation_amount":50000}'),
    ("donation", TIMESTAMP("2025-02-23 13:00:00"), "user_005", "session_005", '{"campaign":"SaveTheEarth","donation_amount":75000}'),
    ("donation", TIMESTAMP("2025-02-23 14:30:00"), "user_006", "session_006", '{"campaign":"HelpTheChildren","donation_amount":100000}');

 

복사 붙여넣기 후 다음과 같이 [실행] 버튼을 클릭하여 데이터를 저장합니다.

 

 

3. 데이터 조회 및 확인

다음은 BigQuery 콘솔에서 SQL을 통해 데이터를 조회해보도록 하겠습니다. 저는 앞서 Mockup 데이터를 실제 비영리단체에서 사용할 법한 데이터를 기반으로 만들어보았습니다. 따라서 사용하실 수 있는 쿼리문 예제는 다음과 같습니다.

 

예제 쿼리: 일별 후원 발생 건수 조회

SELECT
    DATE(event_timestamp) AS donation_date,
    COUNT(*) AS donation_count
FROM datasets.events
WHERE event_name = 'donation'
GROUP BY donation_date
ORDER BY donation_date;

 

위 쿼리를 복사해서 쿼리 입력창에서 실행해보고 결괏값을 확인해 보도록 하겠습니다.

 

 

마찬가지로 다음 예제도 실행해보도록 하겠습니다.

 

예제 쿼리: 특정 캠페인 후원 금액 합산

SELECT
    JSON_VALUE(event_params, '$.campaign') AS campaign_name,
    SUM(CAST(JSON_VALUE(event_params, '$.donation_amount') AS FLOAT)) AS total_donation
FROM datasets.events
WHERE event_name = 'donation'
GROUP BY campaign_name
ORDER BY total_donation DESC;

 

 

위의 예시들과 같이 간단한 SQL을 활용하여 데이터가 정상적으로 저장되었는지 검토할 수 있습니다.

 

4. BigQuery API 활성화

다음으로는 LLM을 활용하기 위해 Python 코드 단에서 데이터를 저장하고 분석할 수 있도록 BigQuery API를 활성화하도록 하겠습니다. 왼쪽 상단의 탐색 메뉴를 클릭합니다.

 

 

다음으로는 [API 및 서비스] 메뉴를 클릭합니다.

 

 

다음으로는 BigQuery API 항목을 클릭합니다.

 

 

오른쪽 상단의 [사용자 인증 정보 만들기] 버튼을 클릭합니다.

 

 

사용할 API와 애플리케이션 데이터를 선택한 후에 [다음] 버튼을 클릭합니다.

 

 

다음으로는 서비스 계정 이름 입력 후 [완료] 버튼을 클릭합니다.

 

 

서비스 계정이 만들어지면 [사용자 인증 정보] 탭을 클릭하여 하단에 생성된 서비스 계정을 확인 후 계정을 클릭합니다.

 

 

서비스 계정 내에서는 다음과 같이 [키] 탭으로 이동하여 [키 추가] 버튼을 클릭한 후 JSON 비공개 키를 생성합니다.

 

 

JSON 파일이 다운로드 되면서 다음과 같이 Key가 활성화되면 API를 사용할 수 있게 됩니다. 다운로드된 JSON파일은 안전한 곳에 잘 보관해 놓습니다.

 

 

다음은 서비스 계정에 빅쿼리 액세스 권한을 부여하도록 하겠습니다. 왼쪽 내비게이션바의 IAM 메뉴(사람 모양의 아이콘)로 접속하여 [액세스 권한 부여] 버튼을 클릭합니다. 다음으로는 주 구성원으로 새로 생성한 서비스 계정을 추가한 뒤에 역할에 BigQuery 관리자를 지정한 후 [저장] 버튼을 클릭합니다.

 

 

다음과 같이 서비스 계정에 BigQuery 관리자 역할이 생성된 것이 확인되었다면 이제 빅쿼리 클라이언트 SDK를 활용하여 코드단에서도 API 호출이 가능합니다.

 

 

---

 

🚀 BigQuery 연동하기

이번에는 앞서 Google BigQuery 스튜디오에서 만들어놓은 데이터셋을 실제 파이썬 코드에서 호출하는 방법에 대해 알아보도록 하겠습니다. 저는 Google에서 제공하고 있는 Colab 환경에서 가이드드리도록 하겠습니다. 먼저 환경변수 셋팅을 위한 dotenv 라이브러리를 설치합니다.

 

# colab에서는 pip앞에 느낌표(!)를 붙여서 모듈을 설치합니다.
!pip install dotenv

 

.env 파일에는 다음과 같이 하이퍼클로바X 호출에 필요한 API 키 값과 BigQuery에서 생성한 테이블 ID를 저장해놓으면 좋습니다. colab에서는 다음과 같이 왼쪽 네비게이션 바에서 폴더(📁) 선택 후 마우스 우 클릭하여 .env 파일을 생성하시면 됩니다.

 

 

.env 파일에는 발급받은 빅쿼리에서 발급받은 프로젝트 아이디를 다음과 같이 입력해 주시면 됩니다.

 

BIGQUERY_PROJECT_ID=빅쿼리에서 발급받은 프로젝트 아이디 입력

 

이후에는 아래와 같이 환경변수를 로드하여 사용하실 수 있습니다. 

load_dotenv()
project_id = os.getenv('BIGQUERY_PROJECT_ID')
table_id = f"{project_id}.datasets.events"

 

다음은 아래와 같이 필요한 라이브러리들을 import 합니다.

import os
from dotenv import load_dotenv
from google.cloud import bigquery

 

다음으로는 BigQueryClient 클래스 파일을 다음과 같이 생성합니다.

class BigQueryClient:
    """Google BigQuery 클라이언트

    BigQuery API를 사용하여 쿼리를 실행하고 결과를 반환하는 클라이언트 클래스입니다.
    """

    def __init__(self):
        try:
            # Colab에서 인증된 계정의 기본 자격 증명 가져오기
            credentials, project_id = google.auth.default()

            # BigQuery 클라이언트 초기화
            self.client = bigquery.Client(
                credentials=credentials,
                project=os.getenv('BIGQUERY_PROJECT_ID')
            )
            print(f"BigQuery 클라이언트 초기화 성공")

        except Exception as e:
            print(f"BigQuery 클라이언트 초기화 실패: {str(e)}")
            raise

    async def execute_query(self, sql: str) -> Dict[str, Any]:
        """SQL 쿼리를 실행하고 결과를 반환합니다.

        Args:
            sql (str): 실행할 SQL 쿼리문.

        Returns:
            Dict[str, Any]: 쿼리 실행 결과를 포함하는 딕셔너리

        Raises:
            Exception: 쿼리 실행 중 오류 발생시
        """
        try:
            print(f"Executing query : {sql}")

            # 쿼리 작업 실행
            query_job = self.client.query(sql)

            # 결과 대기 및 가져오기
            results = query_job.result()

            # 결과를 딕셔너리 리스트로 변환
            rows = [dict(row.items()) for row in results]

            response = {
                "results": rows,
                "total_rows": results.total_rows,
                "job_id": query_job.job_id
            }

            print(f"Query executed successfully. Total rows: {results.total_rows}")
            return response

        except Exception as e:
            error_msg = f"BigQuery 쿼리 실행 실패: {str(e)}"
            print(error_msg)
            raise Exception(error_msg)
            

# ✅ Colab 환경에서 사용하기 위해 Google 계정 인증 먼저 실행
from google.colab import auth
auth.authenticate_user()

# BigQueryClient 인스턴스 생성
bigquery_client = BigQueryClient()

 

colab에서는 Google의 SSO (Single Sign On)로 인해 colab의 계정 인증을 사용하시면 BigQuery 인증까지 가능합니다. 만약 Colab이 아니라 로컬 PC에서 연동할 때는 별도로 auth 라이브러리를 사용해주셔야 합니다. 로컬 PC에서는 다음과 같이 연동하실 수 있습니다.

 

pip install google-cloud-bigquery

 

from google.cloud import bigquery
from google.oauth2 import service_account

from app.core.settings import settings
from app.utils.logger import make_logger

logger = make_logger(__name__)

...

class BigQueryClient:
    def __init__(self):
        try:
            # 서비스 계정 키 파일 경로로부터 인증 정보 생성
            credentials = service_account.Credentials.from_service_account_file(
                settings.GOOGLE_APPLICATION_CREDENTIALS
            )

            # BigQuery 클라이언트 초기화
            self.client = bigquery.Client(
                credentials=credentials,
                project=settings.BIGQUERY_PROJECT_ID
            )
            logger.info("BigQuery 클라이언트 초기화 성공")

        except Exception as e:
            logger.error(f"BigQuery 클라이언트 초기화 실패: {str(e)}")
            raise

 

주의하실 점은 GOOGLE_APPLICATION_CREDENTIALS 환경변수가 추가되었는데요. 이 환경변수에는 이전에 서비스 계정을 만들면서 발급받은 JSON 키 파일이 저장된 경로를 입력해주시면 됩니다. 위 예시 코드에서는 pydantic_settings 모듈을 사용했는데 해당 부분은 os.getenv('GOOGLE_APPLICATION_CREDENTIALS')를 사용해 주셔도 됩니다.

 

다음은 빅쿼리가 실제로 실행되는지 아래와 같이 테스트해보실 수 있습니다.

query = f"""
SELECT
    DATE(event_timestamp) AS donation_date,
    COUNT(*) AS donation_count
FROM {table_id}
WHERE event_name = 'donation'
GROUP BY donation_date
ORDER BY donation_date;
"""

result = await bigquery_client.execute_query(query)
print(result)

 

정상적으로 수행된다면 다음과 같이 결과값이 조회됩니다.

 

---

 

🍀 하이퍼클로바X 연동하기

다음은 하이퍼클로바X 연동방법에 대해 가이드 드리도록 하겠습니다. 하이퍼클로바X의 API 키는 클로바 스튜디오에서 발급받을 수 있는데요. 2025년 1월 부로 API 연동방식의 변화가 생겼습니다.

 

 

기존에는 클로바 스튜디오 플레이그라운드에 접속하여 테스트 앱을 발행한 후 API 키를 발급받았다면 이제는 Bearer 인증 방식으로 조금 더 안전하게 관리가 가능해졌고 불 필요하게 API KEY를 두 개씩이나 입력할 필요가 없어졌습니다.

 

 

물론 기존에 쓰고 있던 방식도 아직까지는 지원이 되고 있습니다. 언제 중단될지 모르는 상황이므로 안전하게 최신 방식으로 변경하시는 게 좋습니다. 

 

그러면 우선, 먼저 새로 변경된 방식으로 API KEY를 발급받아보도록 하겠습니다. 클로바 스튜디오에 접속하여 오른쪽 상단의 계정 메뉴에서 API 키를 선택합니다.

 

clovastudio.ncloud.com

 

오른쪽 상단의 [테스트 API 키 발급] 버튼을 클릭한 후 API 키를 발급하실 수 있습니다. 발급받은 API키는 향후 재확인이 불가능하다고 하니 안전한 곳에 저장해 놓도록 합니다.

 

CLOVA Studio API 발급 방법

 

발급 후에는 다음과 같이 API 발급 이력이 남게 됩니다. 

 

API키 발급 이력 조회

 

새로 발급받은 키는 다시 .env 파일에 다음과 같이 추가합니다.

NCP_CLOVASTUDIO_API_KEY=클로바스튜디오에서 발급받은 키값 입력

 

이후에는 빅쿼리 때와 마찬가지로 아래와 같이 환경변수를 로드하여 사용하실 수 있습니다. 

load_dotenv()

 

다음은 랭체인을 활용하여 하이퍼클로바 X를 호출할 수 있도록 코드를 구현해보도록 하겠습니다. 먼저 필요한 라이브러리를 install 합니다. 참고로 langchain-community 0.3.15 버전부터 새로운 발급방식으로 발행한 키를 적용하실 수 있습니다. 따라서 0.3.15이상 버전을 설치합니다.

 

# colab에서는 pip앞에 느낌표(!)를 붙여서 모듈을 설치합니다.
!pip install langchain-core==0.3.37
!pip install langchain-community==0.3.18

 

랭체인에서 하이퍼클로바X를 호출할 때는 다음과 같이 간단하게 사용하실 수 있습니다.

 

 

---

 

📜 Text-to-SQL(NL2SQL) 프롬프트 작성하기

이제 BigQuery와 하이퍼클로바X 준비를 모두 마쳤으니 실제로 Text-to-SQL 구동을 위한 프롬프트를 작성해 보도록 하겠습니다. 먼저, Text-to-SQL이라는 것은 자연어를 SQL로 변환하여 SQL 지식 없이도 원하는 데이터를 출력해서 사용하는 기법을 말하는데요. 이것이 가능하려면 다음과 같은 템플릿 구조로 프롬프트를 작성해야 합니다.

 

prompt = """
다음은 데이터베이스 스키마 정보입니다:
{schema}

위 스키마에 따라 SQL쿼리를 작성해주세요. 주의:
{filter}

다음은 몇 가지 예시 SQL 쿼리입니다.
{example}

요청 : {query}
SQL 쿼리:
"""

 

위 프롬프트 템플릿에서 각 변수들이 의미하는 것은 다음과 같습니다.

 

템플릿 변수명	설명
schema	데이터베이스 스키마 설명
filter	BigQuery 사용시 주의사항, 제한사항
example	AI가 답변해줬으면 하는 예시 쿼리 (few-shot)
query	사용자가 실제로 카카오톡으로 물어볼것 같은 예상 질문(또는 대화)

 

반드시 위의 템플릿 구조를 따를 필요는 없지만 통상적으로 필요한 정보들이 위와 같은 형태라고 보시면 좋을 것 같습니다. 따라서 위의 템플릿 구조를 바탕으로 코드를 구현하면 다음과 같은 형태가 나타납니다.

 

from langchain_core.runnables import Runnable
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

def sql_agent() -> Runnable:
    """SQL 생성을 위한 LangChain 체인 생성"""

    sql_prompt = """
    당신은 비영리단체에 속한 수준급의 BigQuery 및 SQL 전문가 겸 데이터 분석가 입니다.
    아래 스키마를 참고하여 사용자의 질문에 대해 적절한 SQL문을 생성해주세요.
    당신이 사용할 테이블명은 {table_id} 입니다.

    ** 스키마 **
    | 필드명 | 데이터 유형 | 설명 |
    | ------ | ----------- | ---- |
    | event_name | STRING | 이벤트 이름 (예: page_view, donation) |
    | event_timestamp | TIMESTAMP | 이벤트 발생 시간 |
    | user_id | STRING | 사용자 고유 ID |
    | session_id | STRING | 세션 ID |
    | event_params | STRING | 추가적인 이벤트 정보 (JSON 형식) |

    사용자 질문 : {input}

    ** 주의사항 **
    - 답변은 SQL 쿼리문으로만 대답합니다.
    - 마크다운 형식으로 답변하지 마세요.
    """

    llm = ChatClovaX()
    sql_chain = (lambda x: {"input": x, "table_id": f"{table_id}"}) | PromptTemplate.from_template(sql_prompt) | llm | StrOutputParser()

    return sql_chain

def report_agent() -> Runnable:
    """보고서 생성을 위한 LangChain 체인 생성"""

    report_prompt = """
    사용자 질문 : {input}
    반환된 SQL : {sql}
    쿼리 실행 결과 : {result}

    위 데이터를 바탕으로 보고서를 작성하세요.
    """

    llm = ChatClovaX()
    report_chain = PromptTemplate.from_template(report_prompt) | llm | StrOutputParser()

    return report_chain

async def execute_full_chain(question: str) -> str:
    """SQL 생성부터 보고서 생성까지 전체 실행"""

    # SQL 생성
    sql = await sql_agent().ainvoke(question)

    # 쿼리 실행
    sql_result = await bigquery_client.execute_query(sql)
    print(sql_result)
    result = {
        "input":question,
        "sql":sql,
        "result": str(sql_result.get("results", []))
    }

    # 보고서 생성
    report = await report_agent().ainvoke(result)

    return report

 

위 코드는 사용자 질문이 들어오면 쿼리로 반환하여 BigQuery에서 데이터를 조회해 오고 분석하여 보고서 생성까지 이어지는 간단한 Text-to-SQL 에이전트입니다. 이 코드를 토대로 아래와 같이 실제로 테스트해보면 다음과 같은 결과값이 나타나는 것을 확인하실 수 있습니다.

 

 

---

 

🌠 정리하기

지금까지 간단하게 BigQuery와 하이퍼클로바X를 활용하여 Text-to-SQL 구현하는 방법에 대해 알아보았는데요. 쓰다 보니 조금 긴 글이 되었지만, 최대한 핸즈온하면서 따라 해볼 수 있도록 자세하게 작성해 보았습니다.

 

사실, 조금 더 복잡한 테이블 구조이면 테이블을 나눠야 할 수도 있고 그에 따른 테이블 분류 및 서치 하는 과정도 필요합니다. 이전 포스팅에서 작성하였지만, 이 때는 Vector Database를 활용하여 적절한 테이블과 스키마, few-shot 예제들을 선정해 오는 것이 중요합니다.

 

Text-to-SQL을 위한 랭체인 Vector DBLess 환경 구축하기(ClovaXEmbeddings 활용)

 

Text-to-SQL 프롬프트 엔지니어링에는 여러 가지 기법들이 있겠지만, 가장 중요한 것은 결국 스키마와 설명, 예제들을 어떻게 작성하느냐에 따라 답변 품질이 달라지기 때문에 보유하고 있는 테이블의 상황에 맞게 적절한 프롬프트를 작성하는 것이 중요합니다.