커스텀 SQL 에이전트 만들기
커스텀 SQL 에이전트 만들기 (Build a custom SQL agent)
이 튜토리얼에서는 LangGraph로 SQL 데이터베이스에 대한 질문에 답할 수 있는 커스텀 에이전트를 만들어 볼게요.
LangChain에는 LangGraph 프리미티브로 구현된 내장 에이전트 구현체가 있어요. 더 깊은 커스터마이징이 필요하다면 에이전트를 LangGraph로 직접 구현할 수 있어요. 이 가이드는 SQL 에이전트의 예시 구현을 보여줘요. 실용적인 입문은 더 높은 수준의 LangChain 추상화로 SQL 에이전트 만들기를 참고해요.
SQL 데이터베이스에 대한 Q&A 시스템을 만드는 것은 모델이 생성한 SQL 쿼리를 실행하는 것을 요구해요. 이는 본질적인 위험이 있어요. 데이터베이스 연결 권한이 에이전트의 요구에 맞게 항상 가능한 한 좁게 범위가 지정되도록 하세요. 그래야 모델 주도 시스템을 만드는 위험이 완화되지만, 제거되지는 않아요.
prebuilt 에이전트 덕에 빠르게 시작할 수 있었지만, 우리는 그 동작을 제약하는 데 시스템 프롬프트에 의존했어요 — 예를 들어 에이전트가 항상 "list tables" 도구로 시작하고, 쿼리를 실행하기 전에 항상 쿼리 검사 도구를 실행하도록 지시했죠.
LangGraph에서 에이전트를 커스터마이즈하면 더 높은 수준의 제어를 강제할 수 있어요. 여기서는 특정 도구 호출용 전용 노드를 가진 간단한 ReAct 에이전트 설정을 구현해요. prebuilt 에이전트와 같은 [상태]를 사용할 거예요.
출처: 문서
본문
개념 (Concepts)
다음 개념을 다룰 거예요.
- SQL 데이터베이스에서 읽는 도구
- 상태, 노드, 엣지, 조건부 엣지를 포함한 LangGraph Graph API
- Human-in-the-loop 프로세스
설정 (Setup)
설치 (Installation)
pip install langchain langgraph
LangSmith
LangSmith를 설정해 체인이나 에이전트 안에서 무슨 일이 일어나는지 살펴봐요. 그런 다음 아래 환경 변수를 설정해요.
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
1. LLM 선택 (Select an LLM)
도구 호출(tool-calling)을 지원하는 모델을 선택해요.
아래 예시들의 출력은 OpenAI를 사용했어요.
OpenAI
👉 OpenAI 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain[openai]"
uv를 쓴다면:
uv add "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")
또는 모델 클래스를 직접:
import os
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = "sk-..."
model = ChatOpenAI(model="gpt-5.5")
Anthropic
👉 Anthropic 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain[anthropic]"
uv를 쓴다면:
uv add "langchain[anthropic]"
import os
from langchain.chat_models import init_chat_model
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
model = init_chat_model("claude-sonnet-4-6")
또는 모델 클래스를 직접:
import os
from langchain_anthropic import ChatAnthropic
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
model = ChatAnthropic(model="claude-sonnet-4-6")
Azure
👉 Azure 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain[openai]"
uv를 쓴다면:
uv add "langchain[openai]"
import os
from langchain.chat_models import init_chat_model
os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"
model = init_chat_model(
"azure_openai:gpt-5.5",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
또는 모델 클래스를 직접:
import os
from langchain_openai import AzureChatOpenAI
os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"
model = AzureChatOpenAI(
model="gpt-5.5",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
)
Google Gemini
👉 Google GenAI 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain[google-genai]"
uv를 쓴다면:
uv add "langchain[google-genai]"
import os
from langchain.chat_models import init_chat_model
os.environ["GOOGLE_API_KEY"] = "..."
model = init_chat_model("google_genai:gemini-3.7-flash")
또는 모델 클래스를 직접:
import os
from langchain_google_genai import ChatGoogleGenerativeAI
os.environ["GOOGLE_API_KEY"] = "..."
model = ChatGoogleGenerativeAI(model="gemini-3.7-flash")
AWS Bedrock
👉 AWS Bedrock 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain[aws]"
uv를 쓴다면:
uv add "langchain[aws]"
from langchain.chat_models import init_chat_model
# Follow the steps here to configure your credentials:
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html
model = init_chat_model(
"us.anthropic.claude-sonnet-4-6",
model_provider="bedrock_converse",
)
또는 모델 클래스를 직접:
from langchain_aws import ChatBedrock
model = ChatBedrock(model="us.anthropic.claude-sonnet-4-6")
HuggingFace
👉 HuggingFace 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain[huggingface]"
uv를 쓴다면:
uv add "langchain[huggingface]"
import os
from langchain.chat_models import init_chat_model
os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."
model = init_chat_model(
"microsoft/Phi-3-mini-4k-instruct",
model_provider="huggingface",
temperature=0.7,
max_tokens=1024,
)
또는 모델 클래스를 직접:
import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint
os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."
llm = HuggingFaceEndpoint(
repo_id="microsoft/Phi-3-mini-4k-instruct",
temperature=0.7,
max_length=1024,
)
model = ChatHuggingFace(llm=llm)
OpenRouter
👉 OpenRouter 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain-openrouter"
uv를 쓴다면:
uv add "langchain-openrouter"
import os
from langchain.chat_models import init_chat_model
os.environ["OPENROUTER_API_KEY"] = "sk-..."
model = init_chat_model(
"auto",
model_provider="openrouter",
)
또는 모델 클래스를 직접:
import os
from langchain_openrouter import ChatOpenRouter
os.environ["OPENROUTER_API_KEY"] = "sk-..."
model = ChatOpenRouter(model="auto")
2. 데이터베이스 설정 (Configure the database)
이 튜토리얼에서는 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. 데이터베이스 상호작용 도구 추가 (Add tools for database interactions)
아래 데이터베이스 도구는 데모 목적의 최소 래퍼일 뿐이에요. 보안이 보장되거나 프로덕션에 쓰이도록 만들어진 게 아니에요. 좁게 범위가 지정된 데이터베이스 권한을 사용하고, 모델이 생성한 SQL을 실행하기 전에 애플리케이션별 검증을 추가하세요.
langchain.tools의 @tool 데코레이터를 사용해 데이터베이스 도구를 얇은 래퍼로 구현할 수 있어요.
import sqlite3
from langchain.tools import tool
# Below are minimal tools for demonstration purposes.
@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()
tools = [sql_db_list_tables, sql_db_schema, sql_db_query]
# Use a distinct loop variable so it does not shadow the `tool` decorator,
# which is reused later to wrap the query tool for human review.
for t in tools:
print(f"{t.name}: {t.description}\n")
sql_db_list_tables: Input is an empty string, output is a comma-separated list of tables in the database.
sql_db_schema: 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
sql_db_query: 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.
4. 애플리케이션 단계 정의 (Define application steps)
다음 단계를 위한 전용 노드를 구성해요.
- DB 테이블 나열
- "get schema" 도구 호출
- 쿼리 생성
- 쿼리 검사
이 단계들을 전용 노드에 두면 (1) 필요할 때 도구 호출을 강제하고, (2) 각 단계에 연결된 프롬프트를 커스터마이즈할 수 있어요.
from typing import Literal
from langchain.messages import AIMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
get_schema_tool = next(tool for tool in tools if tool.name == "sql_db_schema")
get_schema_node = ToolNode([get_schema_tool], name="get_schema")
run_query_tool = next(tool for tool in tools if tool.name == "sql_db_query")
run_query_node = ToolNode([run_query_tool], name="run_query")
# Example: create a predetermined tool call
def list_tables(state: MessagesState):
tool_call = {
"name": "sql_db_list_tables",
"args": {},
"id": "abc123",
"type": "tool_call",
}
tool_call_message = AIMessage(content="", tool_calls=[tool_call])
list_tables_tool = next(tool for tool in tools if tool.name == "sql_db_list_tables")
tool_message = list_tables_tool.invoke(tool_call)
response = AIMessage(f"Available tables: {tool_message.content}")
return {"messages": [tool_call_message, tool_message, response]}
# Example: force a model to create a tool call
def call_get_schema(state: MessagesState):
# Note that LangChain enforces that all models accept `tool_choice="any"`
# as well as `tool_choice=<string name of tool>`.
llm_with_tools = model.bind_tools([get_schema_tool], tool_choice="any")
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
generate_query_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.
DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database.
""".format(
dialect="sqlite",
top_k=5,
)
def generate_query(state: MessagesState):
system_message = {
"role": "system",
"content": generate_query_system_prompt,
}
# We do not force a tool call here, to allow the model to
# respond naturally when it obtains the solution.
llm_with_tools = model.bind_tools([run_query_tool])
response = llm_with_tools.invoke([system_message] + state["messages"])
return {"messages": [response]}
check_query_system_prompt = """
You are a SQL expert with a strong attention to detail.
Double check the {dialect} query 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.
You will call the appropriate tool to execute the query after running this check.
""".format(dialect="sqlite")
def check_query(state: MessagesState):
system_message = {
"role": "system",
"content": check_query_system_prompt,
}
# Generate an artificial user message to check
tool_call = state["messages"][-1].tool_calls[0]
user_message = {"role": "user", "content": tool_call["args"]["query"]}
llm_with_tools = model.bind_tools([run_query_tool], tool_choice="any")
response = llm_with_tools.invoke([system_message, user_message])
response.id = state["messages"][-1].id
return {"messages": [response]}
5. 에이전트 구현 (Implement the agent)
이제 이 단계들을 Graph API로 워크플로에 조합할 수 있어요. 쿼리 생성 단계에 조건부 엣지를 정의해요. 쿼리가 생성되면 쿼리 검사기로 라우팅하고, LLM이 응답을 전달해 도구 호출이 없으면 종료해요.
def should_continue(state: MessagesState) -> Literal[END, "check_query"]:
messages = state["messages"]
last_message = messages[-1]
if not last_message.tool_calls:
return END
else:
return "check_query"
builder = StateGraph(MessagesState)
builder.add_node(list_tables)
builder.add_node(call_get_schema)
builder.add_node(get_schema_node, "get_schema")
builder.add_node(generate_query)
builder.add_node(check_query)
builder.add_node(run_query_node, "run_query")
builder.add_edge(START, "list_tables")
builder.add_edge("list_tables", "call_get_schema")
builder.add_edge("call_get_schema", "get_schema")
builder.add_edge("get_schema", "generate_query")
builder.add_conditional_edges(
"generate_query",
should_continue,
)
builder.add_edge("check_query", "run_query")
builder.add_edge("run_query", "generate_query")
agent = builder.compile()
애플리케이션을 시각화해요.
import pathlib
pathlib.Path("graph.png").write_bytes(agent.get_graph().draw_mermaid_png())
이제 그래프를 호출할 수 있어요.
question = "Which genre on average has the longest tracks?"
stream = agent.stream_events(
{"messages": [{"role": "user", "content": question}]},
version="v3",
)
for message in stream.messages:
for token in message.text:
print(token, end="", flush=True)
final_state = stream.output
================================ Human Message =================================
Which genre on average has the longest tracks?
================================== Ai Message ==================================
Available tables: Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track
================================== Ai Message ==================================
Tool Calls:
sql_db_schema (call_yzje0tj7JK3TEzDx4QnRR3lL)
Call ID: call_yzje0tj7JK3TEzDx4QnRR3lL
Args:
table_names: Genre, Track
================================= Tool Message =================================
Name: sql_db_schema
CREATE TABLE "Genre" (
"GenreId" INTEGER NOT NULL,
"Name" NVARCHAR(120),
PRIMARY KEY ("GenreId")
)
/*
3 rows from Genre table:
GenreId Name
1 Rock
2 Jazz
3 Metal
*/
CREATE TABLE "Track" (
"TrackId" INTEGER NOT NULL,
"Name" NVARCHAR(200) NOT NULL,
"AlbumId" INTEGER,
"MediaTypeId" INTEGER NOT NULL,
"GenreId" INTEGER,
"Composer" NVARCHAR(220),
"Milliseconds" INTEGER NOT NULL,
"Bytes" INTEGER,
"UnitPrice" NUMERIC(10, 2) NOT NULL,
PRIMARY KEY ("TrackId"),
FOREIGN KEY("MediaTypeId") REFERENCES "MediaType" ("MediaTypeId"),
FOREIGN KEY("GenreId") REFERENCES "Genre" ("GenreId"),
FOREIGN KEY("AlbumId") REFERENCES "Album" ("AlbumId")
)
/*
3 rows from Track table:
TrackId Name AlbumId MediaTypeId GenreId Composer Milliseconds Bytes UnitPrice
1 For Those About To Rock (We Salute You) 1 1 1 Angus Young, Malcolm Young, Brian Johnson 343719 11170334 0.99
2 Balls to the Wall 2 2 1 U. Dirkschneider, W. Hoffmann, H. Frank, P. Baltes, S. Kaufmann, G. Hoffmann 342562 5510424 0.99
3 Fast As a Shark 3 2 1 F. Baltes, S. Kaufman, U. Dirkscneider & W. Hoffman 230619 3990994 0.99
*/
================================== Ai Message ==================================
Tool Calls:
sql_db_query (call_cb9ApLfZLSq7CWg6jd0im90b)
Call ID: call_cb9ApLfZLSq7CWg6jd0im90b
Args:
query: SELECT Genre.Name, AVG(Track.Milliseconds) AS AvgMilliseconds FROM Track JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.GenreId ORDER BY AvgMilliseconds DESC LIMIT 5;
================================== Ai Message ==================================
Tool Calls:
sql_db_query (call_DMVALfnQ4kJsuF3Yl6jxbeAU)
Call ID: call_DMVALfnQ4kJsuF3Yl6jxbeAU
Args:
query: SELECT Genre.Name, AVG(Track.Milliseconds) AS AvgMilliseconds FROM Track JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.GenreId ORDER BY AvgMilliseconds DESC LIMIT 5;
================================= Tool Message =================================
Name: sql_db_query
[('Sci Fi & Fantasy', 2911783.0384615385), ('Science Fiction', 2625549.076923077), ('Drama', 2575283.78125), ('TV Shows', 2145041.0215053763), ('Comedy', 1585263.705882353)]
================================== Ai Message ==================================
The genre with the longest tracks on average is "Sci Fi & Fantasy," with an average track length of approximately 2,911,783 milliseconds. Other genres with relatively long tracks include "Science Fiction," "Drama," "TV Shows," and "Comedy."
위 실행에 대한 LangSmith 트레이스를 확인해요.
6. Human-in-the-loop 검토 구현 (Implement human-in-the-loop review)
에이전트의 SQL 쿼리가 실행되기 전에 의도치 않은 동작이나 비효율이 없는지 검사하는 것이 신중할 수 있어요.
여기서는 LangGraph의 human-in-the-loop 기능을 활용해 SQL 쿼리를 실행하기 전에 실행을 일시정지하고 인간 검토를 기다려요. LangGraph의 영속성 레이어를 사용하면 실행을 무기한(또는 적어도 영속성 레이어가 살아있는 동안) 일시정지할 수 있어요.
sql_db_query 도구를 인간 입력을 받는 노드로 감싸 볼게요. interrupt 함수로 구현할 수 있어요. 아래에서는 도구 호출을 승인하거나, 그 인자를 편집하거나, 사용자 피드백을 제공하는 입력을 허용해요.
from langchain.tools import tool
from langgraph.types import interrupt
from langchain_core.runnables import RunnableConfig
@tool(
run_query_tool.name,
description=run_query_tool.description,
args_schema=run_query_tool.args_schema,
)
def run_query_tool_with_interrupt(config: RunnableConfig, **tool_input):
request = {
"action": run_query_tool.name,
"args": tool_input,
"description": "Please review the tool call",
}
response = interrupt([request]) # [!code highlight]
# approve the tool call
if response["type"] == "accept":
tool_response = run_query_tool.invoke(tool_input, config)
# update tool call args
elif response["type"] == "edit":
tool_input = response["args"]["args"]
tool_response = run_query_tool.invoke(tool_input, config)
# respond to the LLM with user feedback
elif response["type"] == "response":
user_feedback = response["args"]
tool_response = user_feedback
else:
raise ValueError(f"Unsupported interrupt response type: {response['type']}")
return tool_response
# Redefine the tool node to use the interrupt version
run_query_node = ToolNode([run_query_tool_with_interrupt], name="run_query") # [!code highlight]
위 구현은 더 넓은 human-in-the-loop 가이드의 tool 인터럽트 예시를 따랐어요. 자세한 내용과 대안은 그 가이드를 참고해요.
이제 그래프를 다시 조립할 거예요. 프로그래밍 방식 검사를 인간 검토로 바꿀 거예요. 이제 checkpointer를 포함한다는 점에 주목하세요. 이것은 실행을 일시정지·재개하는 데 필요해요.
from langgraph.checkpoint.memory import InMemorySaver
def should_continue(state: MessagesState) -> Literal[END, "run_query"]:
messages = state["messages"]
last_message = messages[-1]
if not last_message.tool_calls:
return END
else:
return "run_query"
builder = StateGraph(MessagesState)
builder.add_node(list_tables)
builder.add_node(call_get_schema)
builder.add_node(get_schema_node, "get_schema")
builder.add_node(generate_query)
builder.add_node(run_query_node, "run_query")
builder.add_edge(START, "list_tables")
builder.add_edge("list_tables", "call_get_schema")
builder.add_edge("call_get_schema", "get_schema")
builder.add_edge("get_schema", "generate_query")
builder.add_conditional_edges(
"generate_query",
should_continue,
)
builder.add_edge("run_query", "generate_query")
checkpointer = InMemorySaver() # [!code highlight]
agent = builder.compile(checkpointer=checkpointer) # [!code highlight]
전처럼 그래프를 호출할 수 있어요. 이번에는 실행이 인터럽트돼요.
question = "Which genre on average has the longest tracks?"
stream = agent.stream_events(
{"messages": [{"role": "user", "content": question}]},
config,
version="v3",
)
for message in stream.messages:
for token in message.text:
print(token, end="", flush=True)
if stream.interrupted:
action = stream.interrupts[0]
print("INTERRUPTED:")
for request in action.value:
print(json.dumps(request, indent=2))
...
INTERRUPTED:
{
"action": "sql_db_query",
"args": {
"query": "SELECT Genre.Name, AVG(Track.Milliseconds) AS AvgLength FROM Track JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.Name ORDER BY AvgLength DESC LIMIT 5;"
},
"description": "Please review the tool call"
}
Command로 도구 호출을 수락하거나 편집할 수 있어요.
from langgraph.types import Command
stream = agent.stream_events(
Command(resume={"type": "accept"}),
# Command(resume={"type": "edit", "args": {"query": "..."}}),
config,
version="v3",
)
for message in stream.messages:
for token in message.text:
print(token, end="", flush=True)
if stream.interrupted:
action = stream.interrupts[0]
print("INTERRUPTED:")
for request in action.value:
print(json.dumps(request, indent=2))
================================== Ai Message ==================================
Tool Calls:
sql_db_query (call_t4yXkD6shwdTPuelXEmY3sAY)
Call ID: call_t4yXkD6shwdTPuelXEmY3sAY
Args:
query: SELECT Genre.Name, AVG(Track.Milliseconds) AS AvgLength FROM Track JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.Name ORDER BY AvgLength DESC LIMIT 5;
================================= Tool Message =================================
Name: sql_db_query
[('Sci Fi & Fantasy', 2911783.0384615385), ('Science Fiction', 2625549.076923077), ('Drama', 2575283.78125), ('TV Shows', 2145041.0215053763), ('Comedy', 1585263.705882353)]
================================== Ai Message ==================================
The genre with the longest average track length is "Sci Fi & Fantasy" with an average length of about 2,911,783 milliseconds. Other genres with long average track lengths include "Science Fiction," "Drama," "TV Shows," and "Comedy."
자세한 내용은 human-in-the-loop 가이드를 참고해요.
다음 단계 (Next steps)
LangSmith로 이와 같은 SQL 에이전트를 포함한 LangGraph 애플리케이션을 평가하는 방법은 그래프 평가하기 가이드를 확인해요.
더 알아보기 (Learn more)
- SQL 에이전트 (prebuilt): 더 높은 수준의 LangChain 추상화로 만든 SQL 에이전트를 다뤄요.
- Human-in-the-loop 가이드: human-in-the-loop 상세와 대안을 다뤄요.
- Graph API: 노드·엣지·조건부 엣지를 다뤄요.