[Python/Langchain 기초] Text To SQL 간단하게 구현하기

chloe·2025년 3월 13일
LLMlangChainpythonquerysqldatabasetexttosql

Langchain library 활용 기초편
목표 : 데이터 조회 요청을 자연어로 작성하여, 관련 데이터를 반환하는 기능

기본 정보

  • 테스트 환경 : local PostgreSQL, vscode 활용, 테스트는 jupyter notebook 이용
  • python 버전

코드

필요 라이브러리 설치

!pip install psycopg2-binary sqlalchemy
!pip install -U langchain langchain-community langchain-openai
import langchain
langchain.__version__
# '0.3.20'

Postgres DB 데이터 가져오기

  • 이번 테스트는 select 쿼리를 대상으로 하기 때문에 테이블 정보 확인 겸 DB 상태 점검
    • DB 접근은 정상적으로 되는지
    • 어떤 테이블과 데이터가 있는지 등 기본 데이터 확인 진행
from sqlalchemy import create_engine
import pandas as pd

hostname='localhost'
username='xxx'
password='xxx'
port='xxx'
database='xxx'

connect_url = f"postgresql://{username}:{password}@{hostname}:{port}/{database}"
engine = create_engine(connect_url)
table = "employees"

query = f"SELECT * FROM {table}"
df = pd.read_sql(query, engine)
df.head()

Langchain SQLDatabase 라이브러리

  • db가 내가 연결한 DB의 테이블 내역과 테이블의 meta정보를 확보
  • SQLDatabase 클래스 : 테이블의 열 정보와 샘플 데이터를 가져오는 데 사용할 수 있는 get_table_info 메서드 제공
from langchain.sql_database import SQLDatabase

db = SQLDatabase.from_uri(connect_url)

# DB의 테이블 목록 출력
print(db.get_table_names())
print(db.get_usable_table_names())
print(db.get_table_info(table_names=['employees']))
  • output
['employees', 'employees22','test']
['employees', 'employees22','test']

CREATE TABLE employees (
	id SERIAL NOT NULL, 
	name TEXT NOT NULL, 
	age INTEGER NOT NULL, 
	lon NUMERIC(9, 6) NOT NULL, 
	lat NUMERIC(9, 6) NOT NULL, 
	check_time TIMESTAMP WITHOUT TIME ZONE NOT NULL, 
	age_group TEXT NOT NULL, 
	en_name TEXT NOT NULL, 
	department TEXT NOT NULL, 
	position TEXT NOT NULL, 
	residence TEXT NOT NULL, 
	CONSTRAINT employees_pkey PRIMARY KEY (id)
)
3 rows from employees table:
id	name	age	lon	lat	check_time	age_group	en_name	department	position	residence
1	홍길동	25	126.978300	37.566500	2025-02-20 10:00:00	20대	Ace	IT	사원	강남구
2	김철수	30	127.034300	37.565100	2025-02-20 10:05:00	30대	Blaze	마케팅	대리	서초구
3	이영희	22	126.955400	37.558500	2025-02-20 10:10:00	20대	Cloud	영업	사원	송파구
*/

이해

  • 갖고있는 테이블 이름을 가져올 수 있고, get_table_info를 통해서 테이블의 기본 정보를 가지고 있는 것을 볼 수 있음
  • 우리가 어떤 테이블에 query를 날리려고 하면 원하는 칼럼에 대한 정보를 가지고 있어야함
    • 왜냐면 모든 칼럼에 대해 칼럼명과 관련 설명을 다 정확하게 질문하지 않음
    • 내가 원하는 질의에 대해 llm이 테이블의 칼럼을 적절히 적용해서 조건절을 처리하거나 필요한 데이터만 조회할 수 있는 쿼리를 생성해야함
  • 쿼리 대상 테이블에 대한 정보(칼럼명, 타입, 무슨 정보를 갖고있는지 등)가 없으면 아무런 소용 없는 임의의 쿼리문만 생성될 뿐
  • 근데 이런 정보를 이미 db를 통해 가지고 있으므로, llm에서 자연어로 입력한 질의를 SQL문으로 변환할 때 테이블 정보를 이해하고 답을 도출함

질문에 대한 Query 생성

  • create_sql_query_chain : 입력한 DB에 대한 SQL Query를 생성하는 Chain
  • 한글로 작성된 자연어 문장을 입력
from langchain.sql_database import SQLDatabase
from sqlalchemy import create_engine,text
from langchain_openai import ChatOpenAI
from langchain.chains import create_sql_query_chain


db = SQLDatabase.from_uri(connect_url)
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
chain = create_sql_query_chain(llm, db)
query = chain.invoke({"question": "employees의 테이블에서 20대 사람들의 데이터를 모두 가져와줘"})

print(query)
  • output
    • 어떤 칼럼인지 정확하게 작성하지 않아도, meta 정보를 가지고 llm에서 해석하여 연관된 칼럼으로 조건절을 생성
SELECT "id", "name", "age", "lon", "lat", "check_time", "age_group", "en_name", "department", "position", "residence" 
FROM employees 
WHERE "age_group" = '20대'
LIMIT 5;

Query 기반 dataframe 반환 : 예시1 조건절 쿼리

query = chain.invoke({"question": "employees의 테이블에서 20대 사람들의 데이터를 모두 가져와줘"})
df = pd.read_sql(query, engine)

print(query)
df

Query 기반 dataframe 반환 : 예시2 테이블명 미지정

  • 만약 테이블명을 명시적으로 질문하지 않는다면 어떻게 될까?
    • 알아서 해당 db에 있는 테이블을 Union으로 병합해서 반환
query = chain.invoke({"question": "서울 사는 사람 데이터 모두 가져와줘"})
df = pd.read_sql(query, engine)

print(query)
df
  • output
    • 유사한 스키마에 다른 임의 데이터 넣은 테이블 2개에 대해 UNION 쿼리 결과를 반환
      SELECT "id", "name", "age", "lon", "lat", "check_time", "age_group", "en_name", "residence"
FROM employees
WHERE "residence" = '서울'
UNION
SELECT "id", "name", "age", "lon", "lat", "check_time", "age_group", "en_name", "residence"
FROM employees22
WHERE "residence" = '서울'
LIMIT 5;

Query 기반 dataframe 반환 : 예시3 공간쿼리

  • 사용한 DB가 Postgresql이기 때문에, 공간 연산 쿼리 생성을 위한 질의 입력
query = chain.invoke(
    {
        "question": "employees 테이블에서 위도(lon), 경도(lat) 좌표를 사용하여 point geometry 데이터를 생성해줘. 그걸 다시 wkt로 바꿔서 보여줘"
    }
)
df = pd.read_sql(text(query), engine)

print(query)
df
  • output
    • postgreSQL에서 지원하는 st_function을 적절히 사용하여 원하는 공간쿼리문 반환
    • wkt로 바꿔달라는 요청에 알아서 st_asText 변환 함수를 쓰고, st_setSRID로 epsg까지 설정해줌(놀랍다)
    • 결과로는 point geometry를 반환받음

All code & comments

전체 테스트 코드

from sqlalchemy import create_engine, text
import pandas as pd
from langchain.sql_database import SQLDatabase
from langchain_openai import ChatOpenAI
from langchain.chains import create_sql_query_chain

hostname='localhost'
username='xxx'
password='xxx'
port='xxx'
database='xxx'

connect_url = f"postgresql://{username}:{password}@{hostname}:{port}/{database}"
engine = create_engine(connect_url)

db = SQLDatabase.from_uri(connect_url)
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
chain = create_sql_query_chain(llm, db)

# query test
questions=[
	"employees의 테이블에서 20대 사람들의 데이터를 모두 가져와줘",
    "서울 사는 사람 데이터 모두 가져와줘",
    "employees 테이블에서 위도(lon), 경도(lat) 좌표를 사용하여 point geometry 데이터를 생성해줘. 그걸 다시 wkt로 바꿔서 보여줘"
 ]
 
for question in questions:
	query = chain.invoke({"question" : f"{question}"})
    df=pd.read_sql(text(query), engine)
    print(df.head())

comment

  • 데이터에 관심이 많지만 query를 작성하는 것에 어려움이 있는 일반 사용자/고객에게 간편하게 질의를 하는 서비스로 제공하기에 좋아 보임
  • 복잡한 쿼리, 더 많은 데이터를 처리하는 용으로 테스트 추가해보고 실제 활용에 어떻게 적용할 수 있을지 시나리오 구체화 필요
profile
chloe
꾸준하게 기록하기
다음 포스트

Github Gist 사용하기(vscode 연동 및 git clone)

0개의 댓글