SQL 에이전트 만들기
SQL 에이전트 만들기 (Build a SQL agent)
데이터가 데이터베이스에 있다면, 사용자가 자연어로 질문하고 에이전트가 그걸 SQL로 바꿔 실행해 주면 얼마나 좋을까요. 이 튜토리얼에서는 LangChain 에이전트를 이용해 SQL 데이터베이스에 대한 질문에 답할 수 있는 에이전트를 만드는 방법을 배울 거예요. 고수준에서 에이전트는 이렇게 동작합니다.
개요 (Overview)
에이전트가 수행하는 단계는 다음과 같아요.
- 데이터베이스에서 사용 가능한 테이블과 스키마를 가져온다.
- 질문에 관련된 테이블을 결정한다.
- 관련 테이블의 스키마를 가져온다.
- 질문과 스키마 정보를 바탕으로 쿼리를 생성한다.
- LLM을 사용해 흔한 실수가 없는지 쿼리를 이중으로 확인한다.
- 쿼리를 실행하고 결과를 반환한다.
- 데이터베이스 엔진이 드러낸 실수를 쿼리가 성공할 때까지 수정한다.
- 결과를 바탕으로 응답을 구성한다.
SQL 데이터베이스에 대한 Q&A 시스템을 만들려면 모델이 생성한 SQL 쿼리를 실행해야 해요. 여기에는 본질적인 위험이 있습니다. 에이전트의 요구에 맞게 데이터베이스 연결 권한을 항상 가능한 한 좁게 스코프하세요. 이렇게 하면 모델 기반 시스템을 만들 때의 위험을 줄일 수 있지만, 완전히 없애지는 못해요.
다룰 개념 (Concepts)
- 도구(Tools) — SQL 데이터베이스에서 읽는 데 사용
- LangChain 에이전트(agents)
- 휴먼-인-더-루프(human-in-the-loop) 프로세스
설정 (Setup)
1. 의존성 설치
pip install langchain langgraph
2. LangSmith 설정
LangSmith를 설정해 체인이나 에이전트 안에서 무슨 일이 일어나는지 확인하세요. 다음 환경 변수를 설정합니다.
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
SQL 에이전트 만들기 (Build your SQL agent)
1. LLM 선택
tool-calling을 지원하는 모델을 선택하세요. 예를 들어 OpenAI라면:
pip install -U "langchain[openai]"
import os
from langchain.chat_models import init_chat_model
os.environ["OPENAI_API_KEY"] = "sk-..."
model = init_chat_model("gpt-5.5")
Anthropic, Azure, Google Gemini, AWS Bedrock, HuggingFace, OpenRouter 등 다른 제공자의 설치·모델 클래스 예시는 원문 문서의 탭을 참고하세요. 아래 예시에서 보여주는 출력은 OpenAI를 사용한 결과예요.
2. 데이터베이스 구성
튜토리얼용으로 SQLite 데이터베이스를 만들 거예요. SQLite는 설정과 사용이 쉬운 가벼운 데이터베이스죠. 디지털 미디어 스토어를 나타내는 샘플 데이터베이스인 chinook을 불러오겠습니다. 편의를 위해 Chinook.db를 공개 GCS 버킷에 호스팅해 뒀어요.
import pathlib
import requests
url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
local_path = pathlib.Path("Chinook.db")
if local_path.exists():
print(f"{local_path} already exists, skipping download.")
else:
response = requests.get(url, timeout=60)
if response.status_code == 200:
local_path.write_bytes(response.content)
print(f"File downloaded and saved as {local_path}")
else:
print(f"Failed to download the file. Status code: {response.status_code}")
Python 내장 sqlite3 모듈로 데이터베이스와 상호작용할게요.
import sqlite3
con = sqlite3.connect("Chinook.db")
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")]
print("Dialect: sqlite")
print(f"Available tables: {tables}")
cursor.execute("SELECT * FROM Artist LIMIT 5;")
print(f"Sample output: {cursor.fetchall()}")
con.close()
Dialect: sqlite
Available tables: ['Album', 'Artist', 'Customer', 'Employee', 'Genre', 'Invoice', 'InvoiceLine', 'MediaType', 'Playlist', 'PlaylistTrack', 'Track']
Sample output: [(1, 'AC/DC'), (2, 'Accept'), (3, 'Aerosmith'), (4, 'Alanis Morissette'), (5, 'Alice In Chains')]
3. 데이터베이스 상호작용 도구 추가
아래 데이터베이스 도구들은 데모용 최소 래퍼에요. 보안을 고려하지 않았고 운영용이 아니므로, 좁은 범위의 DB 권한을 쓰고 모델이 생성한 SQL을 실행하기 전에 애플리케이션별 검증을 추가해야 합니다.
langchain.tools의 @tool 데코레이터로 얇은 래퍼를 구현할 수 있어요.
import sqlite3
from langchain.tools import tool
# Below are minimal tools for demonstration purposes.
# They are not intended to be secure or for production use.
@tool
def sql_db_list_tables() -> str:
"""Input is an empty string, output is a comma-separated list of tables in the database."""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")]
return ", ".join(tables)
finally:
con.close()
@tool
def sql_db_schema(table_names: str) -> str:
"""Input to this tool is a comma-separated list of tables, output is the schema and sample rows for those tables.
Be sure that the tables actually exist by calling sql_db_list_tables first!
Example Input: table1, table2, table3"""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
valid_tables = {row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")}
results = []
for table in table_names.split(","):
table = table.strip()
if table not in valid_tables:
results.append(f"Error: table_names {table!r} not found in database")
continue
cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name=?;", (table,))
schema_row = cursor.fetchone()
if schema_row:
results.append(schema_row[0])
try:
quoted_table = '"' + table.replace('"', '""') + '"'
cursor.execute(f"SELECT * FROM {quoted_table} LIMIT 3;")
rows = cursor.fetchall()
if rows:
col_names = [description[0] for description in cursor.description]
results.append(
f"/*\n3 rows from {table} table:\n"
+ "\t".join(col_names)
+ "\n"
+ "\n".join("\t".join(str(x) for x in row) for row in rows)
+ "\n*/"
)
except Exception as e:
results.append(f"Error fetching sample rows: {e}")
return "\n\n".join(results)
finally:
con.close()
@tool
def sql_db_query(query: str) -> str:
"""Input to this tool is a detailed and correct SQL query, output is a result from the database.
If the query is not correct, an error message will be returned.
If an error is returned, rewrite the query, check the query, and try again.
If you encounter an issue with Unknown column 'xxxx' in 'field list', use sql_db_schema to query the correct table fields."""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute(query)
res = cursor.fetchall()
return str(res)
except Exception as e:
return f"Error: {e}"
finally:
con.close()
@tool
def sql_db_query_checker(query: str) -> str:
"""Use this tool to double check if your query is correct before executing it.
Always use this tool before executing a query with sql_db_query!"""
trigger_prompt = """{query}
Double check the sqlite query above for common mistakes, including:
- Using NOT IN with NULL values
- Using UNION when UNION ALL should have been used
- Using BETWEEN for exclusive ranges
- Data type mismatch in predicates
- Properly quoting identifiers
- Using the correct number of arguments for functions
- Casting to the correct data type
- Using the proper columns for joins
If there are any of the above mistakes, rewrite the query. If there are no mistakes, just reproduce the original query.
Output the final SQL query only.
SQL Query: """.format(query=query)
response = model.invoke(trigger_prompt)
return response.text.strip()
tools = [sql_db_list_tables, sql_db_schema, sql_db_query, sql_db_query_checker]
# Use a distinct loop variable so it does not shadow the `tool` decorator.
for t in tools:
print(f"{t.name}: {t.description}\n")
sql_db_query_checker가 흥미로운 부분이에요. 쿼리를 실행하기 전에 LLM으로 쿼리를 이중 확인해 흔한 실수(NOT IN with NULL, UNION vs UNION ALL, BETWEEN exclusive ranges 등)를 잡아내죠.
4. 에이전트 만들기
create_agent로 최소 코드의 ReAct 에이전트를 만들어요. 에이전트는 요청을 해석해 SQL 커맨드를 생성하고, 도구가 이를 실행해요. 커맨드에 오류가 있으면 오류 메시지가 모델로 반환되고, 모델은 원래 요청과 새 오류 메시지를 보고 새 커맨드를 생성할 수 있어요. 이 과정은 LLM이 커맨드를 성공적으로 생성하거나 종료 카운트에 도달할 때까지 계속됩니다. 모델에 피드백(여기서는 오류 메시지)을 제공하는 이 패턴은 매우 강력해요.
행동을 커스터마이즈할 서술적인 시스템 프롬프트로 에이전트를 초기화해요.
system_prompt = """
You are an agent designed to interact with a SQL database.
Given an input question, create a syntactically correct {dialect} query to run,
then look at the results of the query and return the answer. Unless the user
specifies a specific number of examples they wish to obtain, always limit your
query to at most {top_k} results.
You can order the results by a relevant column to return the most interesting
examples in the database. Never query for all the columns from a specific table,
only ask for the relevant columns given the question.
You MUST double check your query before executing it. If you get an error while
executing a query, rewrite the query and try again.
DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the
database.
To start you should ALWAYS look at the tables in the database to see what you
can query. Do NOT skip this step.
Then you should query the schema of the most relevant tables.
""".format(
dialect="sqlite",
top_k=5,
)
이제 모델, 도구, 프롬프트로 에이전트를 만들어요.
from langchain.agents import create_agent
agent = create_agent(
model,
tools,
system_prompt=system_prompt,
)
5. 에이전트 실행
샘플 쿼리로 에이전트를 실행하고 동작을 관찰해 볼게요.
question = "Which genre on average has the longest tracks?"
stream = agent.stream_events(
{"messages": [{"role": "user", "content": question}]},
version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
if kind == "messages":
for token in item.text:
print(token, end="", flush=True)
elif kind == "tool_calls":
print(f"\nTool call: {item.tool_name}({item.input})")
for delta in item.output_deltas:
print(delta, end="", flush=True)
print(f"\nTool result: {item.output}")
final_state = stream.output
에이전트는 테이블 목록을 조회하고(sql_db_list_tables), 관련 테이블(Track, Genre)의 스키마를 가져오고(sql_db_schema), 쿼리를 이중 확인한 뒤(sql_db_query_checker) 실제로 실행합니다(sql_db_query). 그리고 결과를 바탕으로 최종 응답을 만들어요 — "Sci Fi & Fantasy"가 평균 트랙 길이가 가장 긴 장르라고요. 위 실행의 모든 측면(취한 단계, 호출한 도구, LLM이 본 프롬프트 등)은 LangSmith trace에서 확인할 수 있어요.
6. (선택) Studio 사용
Studio는 "클라이언트 사이드" 루프와 메모리를 제공해 채팅 인터페이스로 이걸 실행하고 데이터베이스에 질문할 수 있어요. "데이터베이스 스키마를 알려줘"나 "상위 5명 고객의 인보이스를 보여줘" 같은 질문을 할 수 있고, 생성된 SQL 커맨드와 결과 출력을 볼 수 있습니다.
Studio에서 에이전트를 실행하려면 앞서 언급한 패키지 외에 다음이 필요해요.
pip install -U langgraph-cli[inmem]>=0.4.0
실행할 디렉터리에 다음 내용의 langgraph.json 파일이 필요해요.
{
"dependencies": ["."],
"graphs": {
"agent": "./agent.py:agent"
},
"env": ".env"
}
휴먼-인-더-루프 (Human-in-the-loop)
sql_db_query 도구를 실행하기 전에 에이전트가 리뷰를 위해 잠시 멈추게 할 수도 있어요. 이렇게 하면 위험한 SQL 실행 전에 사용자에게 승인을 받을 수 있습니다.
question = "Which genre on average has the longest tracks?"
config = {"configurable": {"thread_id": "1"}}
stream = agent.stream_events(
{"messages": [{"role": "user", "content": question}]},
config,
version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
if kind == "messages":
for token in item.text:
print(token, end="", flush=True)
elif kind == "tool_calls":
print(f"\nTool call: {item.tool_name}({item.input})")
if stream.interrupted:
print("INTERRUPTED:")
interrupt = stream.interrupts[0]
for request in interrupt.value["action_requests"]:
print(request["description"])
그러면 에이전트가 sql_db_query 실행 전에 리뷰를 위해 멈춰요.
...
INTERRUPTED:
Tool execution pending approval
Tool: sql_db_query
Args: {'query': 'SELECT g.Name AS Genre, AVG(t.Milliseconds) AS AvgTrackLength FROM Track t JOIN Genre g ON t.GenreId = g.GenreId GROUP BY g.Name ORDER BY AvgTrackLength DESC LIMIT 1;'}
Command로 실행을 재개하고, 이 경우에는 쿼리를 승인하면 됩니다.
from langgraph.types import Command
stream = agent.stream_events(
Command(resume={"decisions": [{"type": "approve"}]}),
config,
version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
if kind == "messages":
for token in item.text:
print(token, end="", flush=True)
elif kind == "tool_calls":
print(f"\nTool call: {item.tool_name}({item.input})")
if stream.interrupted:
print("INTERRUPTED:")
interrupt = stream.interrupts[0]
for request in interrupt.value["action_requests"]:
print(request["description"])
이렇게 하면 에이전트가 쿼리를 실행하고 "Sci Fi & Fantasy"가 가장 긴 평균 트랙 길이를 가진 장르라는 응답을 내놓아요. 자세한 내용은 휴먼-인-더-루프 가이드를 참고하세요.
다음 단계 (Next steps)
더 깊은 커스터마이즈를 원한다면, LangGraph 원시 요소를 사용해 SQL 에이전트를 직접 구현하는 이 튜토리얼을 확인해 보세요.