커스텀 SQL 에이전트 만들기
커스텀 SQL 에이전트 만들기
이 튜토리얼에서는 LangGraph를 사용해 SQL 데이터베이스에 대한 질문에 답할 수 있는 커스텀 에이전트를 만들 거예요.
LangChain은 LangGraph 프리미티브로 구현된 내장 agent 구현을 제공해요. 더 깊은 커스터마이징이 필요하면 에이전트를 LangGraph에서 직접 구현할 수 있어요. 이 가이드는 SQL 에이전트의 예시 구현을 보여줍니다. 실용적인 소개는 더 높은 수준의 LangChain 추상화로 SQL 에이전트 만들기를 참고하세요.
출처: 공식문서
⚠️ 경고 — SQL 데이터베이스에 대한 Q&A 시스템을 만드는 데는 모델이 생성한 SQL 쿼리를 실행하는 일이 포함돼요. 이 과정에는 본질적인 위험이 있습니다. 데이터베이스 연결 권한이 에이전트의 필요에 맞게 항상 최대한 좁게 범위 지정되어 있는지 확인하세요. 이렇게 하면 모델 주도 시스템 구축의 위험을 줄일 수는 있지만 없애지는 않습니다.
사전 빌드된 에이전트는 빠르게 시작할 수 있게 해주지만, 우리는 그 동작을 제약하기 위해 시스템 프롬프트에 의존했어요—예를 들어 에이전트가 항상 "list tables" 도구로 시작하고, 쿼리를 실행하기 전에 항상 쿼리-검사기 도구를 실행하도록 지시했죠.
LangGraph에서 에이전트를 커스터마이징하면 더 높은 수준의 제어를 강제할 수 있어요. 여기서는 특정 도구 호출을 위한 전용 노드를 가진 간단한 ReAct-에이전트 구성을 구현합니다. 사전 빌드된 에이전트와 같은 [state]를 사용할 거예요.
개념
다음 개념을 다룹니다:
- SQL 데이터베이스에서 읽기 위한 Tools
- 상태·노드·엣지·조건부 엣지를 포함한 LangGraph Graph API
- Human-in-the-loop 프로세스
준비
설치
pip
pip install langchain langgraph
LangSmith
LangSmith를 설정해 체인이나 에이전트 안에서 무슨 일이 일어나는지 검사하세요. 그런 다음 다음 환경 변수를 설정합니다.
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
1. LLM 선택
tool-calling을 지원하는 모델을 선택하세요.
OpenAI — 👉 OpenAI 채팅 모델 통합 문서를 읽어보세요.
pip install -U "langchain[openai]"
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 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 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 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 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 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 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")
아래 예시에 나오는 출력은 OpenAI를 사용한 결과입니다.
2. 데이터베이스 구성
이 튜토리얼에서는 SQLite 데이터베이스를 만들 거예요. SQLite는 설정과 사용이 쉬운 경량 데이터베이스예요. 디지털 미디어 스토어를 나타내는 샘플 데이터베이스인 chinook 데이터베이스를 로드할 거예요.
편의를 위해 공개 GCS 버킷에 데이터베이스(Chinook.db)를 호스팅했습니다.
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. 데이터베이스 상호작용 도구 추가
⚠️ 경고 — 다음 데이터베이스 도구는 데모 목적의 최소 래퍼일 뿐이에요. 보안 강화나 프로덕션 사용을 위한 것이 아닙니다. 좁은 범위의 데이터베이스 권한을 사용하고, 모델 생성 SQL을 실행하기 전에 애플리케이션 특화 검증을 추가하세요.
langchain.tools의 @tool 데코레이터를 사용해 데이터베이스 tools를 얇은 래퍼로 구현할 수 있어요:
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. 애플리케이션 단계 정의
다음 단계를 위한 전용 노드를 구성합니다:
- 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. 에이전트 구현
이제 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 검토 구현
에이전트의 SQL 쿼리를 실행하기 전에 의도하지 않은 동작이나 비효율이 없는지 검사하는 것이 현명할 수 있어요.
여기서는 LangGraph의 human-in-the-loop 기능을 활용해 SQL 쿼리를 실행하기 전에 런을 멈추고 사람 검토를 기다리게 합니다. LangGraph의 persistence 계층을 사용하면 런을 무기한(또는 적어도 persistence 계층이 살아있는 동안) 멈출 수 있어요.
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 interrupt 예시를 따릅니다. 자세한 내용과 대안은 그 가이드를 참고하세요.
이제 그래프를 다시 조립합시다. 프로그램적 검사를 사람 검토로 대체할 거예요. 이제 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 가이드를 참고하세요.
다음 단계
LangSmith를 사용해 SQL 에이전트 같은 LangGraph 애플리케이션을 평가하는 방법은 그래프 평가 가이드를 확인하세요.