watsonx.ai를 활용해 LangGraph와 Mistral Medium 3으로 SQL 에이전트 구축

작성자

AI Engineer, Developer Advocate

IBM

이 튜토리얼에서는 사용자 지정 SQLite 데이터베이스에 대해 Python 및 SQL 쿼리를 실행하고 생성할 수 있는 AI 에이전트를 구축합니다. 에이전트는 LangGraph와 Mistral Medium 3 대규모 언어 모델(LLM)을 사용하여 처음부터 구축됩니다. 이 LLM은 코딩 및 멀티모달 이해와 같은 전문적인 사용 사례에 최적화되어 있어, 이 작업에 매우 적합한 선택입니다.¹ 또한 IBM® Granite Guardian 모델을 사용하여 유해한 메시지를 탐지하고 차단합니다.

LangGraph란 무엇인가요?

LangChain에서 개발한 LangGraph는 복잡한 생성형 AI 에이전트 워크플로를 구축, 배포 및 관리하도록 설계된 오픈 소스 AI 에이전트 프레임워크입니다. 이는 사용자가 LLM을 확장 가능하고 효율적인 방식으로 생성, 실행 및 최적화할 수 있도록 하는 툴과 라이브러리 세트를 제공합니다. 핵심적으로 LangGraph는 그래프 기반 아키텍처의 강점을 활용하여 다중 에이전트 시스템에 포함된 구성 요소를 포함해 AI 에이전트 워크플로의 다양한 구성 요소 간의 복잡한 관계를 모델링하고 관리합니다.

IBM watsonx.ai

LangGraph와 watsonx.ai를 활용한 데이터 인사이트

AI 에이전트가 인간의 자연어 쿼리를 이해하고 처리하여 의미 있는 결과물을 도출할 수 있을까요? IBM은 여러 오픈 소스 기술과 watsonx.ai의 강력한 기능을 사용해 그 가능성을 테스트합니다.

watsonx.ai 살펴보기

전제조건

watsonx.ai 프로젝트를 생성하려면 IBM® Cloud 계정이 필요합니다.
이 튜토리얼에서는 여러 Python 버전을 사용할 수 있습니다. 게시 시점을 기준으로 최신 버전인 Python 3.13을 다운로드할 것을 권장합니다.

단계

1단계. 환경 설정

여러 툴 중에서 선택할 수 있지만, 이 튜토리얼에서는 Jupyter Notebook을 사용하기 위해 IBM 계정을 설정하는 방법을 안내합니다.

IBM Cloud 계정을 사용하여 watsonx.ai에 로그인하세요.
watsonx.ai 프로젝트를 생성합니다.

프로젝트 내에서 프로젝트 ID를 가져올 수 있습니다. Manage (관리) 탭을 클릭합니다. 그런 다음 일반 페이지의 세부 정보 섹션에서 프로젝트 ID를 복사합니다. 이 튜토리얼에는 이 ID가 필요합니다.
Jupyter Notebook을 만듭니다.

이 단계에서는 이 튜토리얼의 코드를 복사할 수 있는 Jupyter Notebook 환경이 열립니다. 또는 이 노트북을 로컬 시스템에 다운로드하여 watsonx.ai 프로젝트에 에셋으로 업로드할 수 있습니다. 이 튜토리얼은 GitHub에서도 확인할 수 있습니다.

2단계. watsonx.ai Runtime 인스턴스 및 API 키 설정

watsonx.ai Runtime 서비스 인스턴스를 생성하세요(적절한 리전을 선택하고 무료 인스턴스인 Lite 요금제를 선택하세요).
애플리케이션 프로그래밍 인터페이스 (API) 키를 생성하세요.
생성한 watsonx.ai Runtime 서비스 인스턴스를 watsonx.ai에서 만든 프로젝트와 연결하세요.

3단계. 관련 라이브러리를 설치하고 가져온 다음 자격 증명을 설정합니다.

이 튜토리얼에는 몇 가지 라이브러리와 모듈이 필요합니다. 다음 항목을 가져와야 합니다. 설치되지 않은 경우, 빠른 PIP 설치로 이 문제를 해결할 수 있습니다.

#installations
%pip install -qU ibm-watsonx-ai \
    langchain-ibm \
    langgraph \
    langchain_experimental \
    tavily-python \
    langchain-community

터미널을 재시작한 후 다음 패키지를 가져옵니다.

# imports
import sqlite3
import getpass

from ibm_watsonx_ai import APIClient, Credentials
from ibm_watsonx_ai.foundation_models.moderations import Guardian
from IPython.display import Image, display
from langchain import hub  
from langchain_ibm import ChatWatsonx
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_community.utilities.sql_database import SQLDatabase
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage, AIMessage
from langchain_core.tools import tool
from langchain_experimental.tools.python.tool import PythonREPLTool
from sqlalchemy import create_engine
from typing_extensions import TypedDict
from typing import Annotated

자격 증명을 설정하려면 1단계에서 생성한 WATSONX_APIKEY와 WATSONX_PROJECT_ID가 필요합니다. 또한 URL을 API 엔드포인트로 설정합니다.

WATSONX_APIKEY = getpass.getpass("watson.ai 런타임 API 키를 입력하세요(Enter 키 누르기): ")

WATSONX_PROJECT_ID = getpass.getpass("프로젝트 ID를 입력하세요(Enter 키 누르기): ")

URL = "https://us-south.ml.cloud.ibm.com"

LLM을 초기화하기 전에 Credentials 클래스를 사용하여 전달된 API 자격 증명을 캡슐화할 수 있습니다.

credentials = Credentials(url=URL, api_key=WATSONX_APIKEY)

4단계. 채팅 모델 인스턴스화

watsonx.ai Runtime에서 사용 가능한 모든 리소스와 상호작용하려면 APIClient를 설정해야 합니다. 여기에서는 자격 증명과 WATSONX_PROJECT_ID를 전달합니다.

client = APIClient(credentials=credentials, project_id=WATSONX_PROJECT_ID)

이 튜토리얼에서는 ChatWatsonx 래퍼를 사용하여 채팅 모델을 설정합니다. 이 래퍼는 툴 호출과 체이닝의 통합을 간소화합니다. 자세한 내용은 ChatWatsonx 공식 문서의 API 참조를 확인하는 것을 권장합니다. Mistral Medium 3에 대한 model_id와 클라이언트를 매개변수로 전달할 수 있습니다.

다른 API 제공자를 사용하는 경우 이에 맞게 래퍼를 변경해야 합니다. 예를 들어 GPT-4와 같은 모델에 액세스하기 위해 OpenAI API를 사용하려면 openai_api_key와 ChatOpenAI 래퍼가 필요합니다.

model_id = "mistralai/mistral-medium-2505" 
llm = ChatWatsonx(model_id=model_id, watsonx_client=client)

5단계. SQLite 데이터베이스 설정

이 튜토리얼에서 에이전트는 사용 가능한 툴을 활용하여 SQLite 데이터베이스와 상호작용합니다. 이미 PostgreSQL이나 SQLite와 같은 관계형 데이터베이스 관리 시스템에 데이터세트가 저장되어 있다면 이 단계를 건너뛰어도 됩니다. 그렇지 않은 경우 다음 셀을 실행하여 sql_script 변수를 가상의 자동차 딜러십 판매 데이터가 포함된 데이터베이스를 생성하는 스크립트로 설정합니다. 이 데이터베이스에는 개별 딜러십, 판매 및 차량 데이터를 저장하기 위한 여러 테이블이 포함됩니다.

sql_script = """
DROP TABLE IF EXISTS [Dealerships];

DROP TABLE IF EXISTS [Cars];

DROP TABLE IF EXISTS [Sales];

CREATE TABLE [Dealerships] (
    [id] INTEGER PRIMARY KEY,
    [name] TEXT NOT NULL,
    [location] TEXT NOT NULL
);

CREATE TABLE [Cars] (
    [id] INTEGER PRIMARY KEY,
    [make] TEXT NOT NULL,
    [model] TEXT NOT NULL,
    [year] INTEGER NOT NULL,
    [color] TEXT NOT NULL
);

CREATE TABLE [Sales] (
    [id] INTEGER PRIMARY KEY,
    [car_id] INTEGER NOT NULL,
    [dealership_id] INTEGER NOT NULL,
    [sale_date] DATE NOT NULL,
    [sale_price] REAL NOT NULL,
    FOREIGN KEY ([car_id]) REFERENCES [Cars] ([id]),
    FOREIGN KEY ([dealership_id]) REFERENCES [Dealerships] ([id])
);


INSERT INTO [Dealerships] ([id], [name], [location]) VALUES
    (1, 'Dealership A', 'New York'),
    (2, 'Dealership B', 'Los Angeles'),
    (3, 'Dealership C', 'Chicago'),
    (4, 'Dealership D', 'Houston'),
    (5, 'Dealership E', 'Phoenix'),
    (6, 'Dealership F', 'Philadelphia'),
    (7, 'Dealership G', 'San Antonio'),
    (8, 'Dealership H', 'San Diego'),
    (9, 'Dealership I', 'Dallas'),
    (10, 'Dealership J', 'San Jose');

INSERT INTO [Cars] ([id], [make], [model], [year], [color]) VALUES
    (1, 'Toyota', 'Camry', 2020, 'Blue'),
    (2, 'Honda', 'Civic', 2019, 'Red'),
    (3, 'Ford', 'Mustang', 2021, 'Black'),
    (4, 'Chevrolet', 'Silverado', 2018, 'White'),
    (5, 'Nissan', 'Altima', 2020, 'Gray'),
    (6, 'Kia', 'Optima', 2020, 'Silver'),
    (7, 'Hyundai', 'Elantra', 2019, 'Black'),
    (8, 'Volkswagen', 'Golf', 2021, 'Red'),
    (9, 'BMW', '3 Series', 2018, 'White'),
    (10, 'Mercedes-Benz', 'C-Class', 2020, 'Gray'),
    (11, 'Audi', 'A4', 2019, 'Blue'),
    (12, 'Lexus', 'ES', 2021, 'Black'),
    (13, 'Toyota', 'Corolla', 2018, 'White'),
    (14, 'Honda', 'Accord', 2020, 'Gray'),
    (15, 'Ford', 'Fusion', 2019, 'Red');

INSERT INTO [Sales] ([id], [car_id], [dealership_id], [sale_date], [sale_price]) VALUES
    (1, 1, 1, '2022-01-01', 25000.0),
    (2, 2, 2, '2022-02-01', 20000.0),
    (3, 3, 3, '2022-03-01', 30000.0),
    (4, 4, 1, '2022-04-01', 40000.0),
    (5, 5, 2, '2022-05-01', 28000.0),
    (6, 6, 4, '2022-06-01', 22000.0),
    (7, 7, 5, '2022-07-01', 20000.0),
    (8, 8, 6, '2022-08-01', 28000.0),
    (9, 9, 7, '2022-09-01', 35000.0),
    (10, 10, 8, '2022-10-01', 32000.0),
    (11, 11, 9, '2022-11-01', 30000.0),
    (12, 12, 10, '2022-12-01', 38000.0),
    (13, 13, 1, '2023-01-01', 25000.0),
    (14, 14, 2, '2023-02-01', 28000.0),
    (15, 15, 3, '2023-03-01', 22000.0);
    
"""

이 스크립트를 실행하고 로컬 SQLite 데이터베이스를 생성하려면 데이터베이스 Connection 객체를 설정해야 합니다. ":memory:" 경로 이름을 전달하면 일시적인 인메모리 데이터베이스에 연결할 수 있습니다.

connection = sqlite3.connect(":memory:")

다음으로 executescript 메서드를 사용하여 데이터베이스 커서를 생성하고 SQL 스크립트를 실행할 수 있습니다. 이 Cursor 객체를 통해 이러한 스크립트를 실행할 수 있습니다.

connection.executescript(sql_script)

아웃풋:

<sqlite3.Cursor at 0x30c511240>

마지막으로 다음 명령을 실행하여 작업 디렉터리에 있는 데이터베이스 파일에 연결된 SQLAlechemy Engine 인스턴스를 생성합니다. 해당 파일은 Jupyter Notebook과 동일한 이름에 데이터베이스 파일 확장자를 가져야 합니다. URL을 비워 두면 SQLite 로컬 데이터베이스로 연결됩니다. creator 매개변수에 Connection 객체를 제공할 수 있습니다. 이 매개변수는 데이터베이스 연결을 생성하는 역할을 합니다.

engine = create_engine(
    "sqlite://",
    creator=lambda: connection
)

이 튜토리얼에서 로컬에 저장되지 않은 대체 데이터베이스를 사용하려는 경우, 빈 경로를 "sqlite:///Chinook.db"로 바꾸기만 하면 샘플 디지털 미디어 스토어 데이터베이스에 액세스할 수 있습니다.

6단계. 툴 설정

이 튜토리얼에서는 LangChain에서 미리 빌드된 SQLDatabaseToolkit을 사용합니다. 이 툴킷은 SQLAlchemy Engine을 매개변수로 받는 SQLDatabase 객체와 선택한 LLM을 필요로 합니다. 툴킷을 인스턴스화한 후 해당 툴들을 가져올 수 있습니다. 툴을 출력하여 자세한 내용을 확인해 보겠습니다.

db = SQLDatabase(engine)
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()
툴

아웃풋:

[QuerySQLDatabaseTool(description="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.", db=<langchain_community.utilities.sql_database.SQLDatabase object at 0x30c3ecd10>),
 InfoSQLDatabaseTool(description='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', db=<langchain_community.utilities.sql_database.SQLDatabase object at 0x30c3ecd10>),
 ListSQLDatabaseTool(db=<langchain_community.utilities.sql_database.SQLDatabase object at 0x30c3ecd10>),
 QuerySQLCheckerTool(description='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!', db=<langchain_community.utilities.sql_database.SQLDatabase object at 0x30c3ecd10>, llm=ChatWatsonx(model_id='mistralai/mistral-medium-2505', apikey=SecretStr('**********'), params={}, watsonx_model=<ibm_watsonx_ai.foundation_models.inference.model_inference.ModelInference object at 0x309c34690>, watsonx_client=<ibm_watsonx_ai.client.APIClient object at 0x30c3e3250>), llm_chain=LLMChain(verbose=False, prompt=PromptTemplate(input_variables=['dialect', 'query'], input_types={}, partial_variables={}, template='\n{query}\nDouble check the {dialect} query above for common mistakes, including:\n- Using NOT IN with NULL values\n- Using UNION when UNION ALL should have been used\n- Using BETWEEN for exclusive ranges\n- Data type mismatch in predicates\n- Properly quoting identifiers\n- Using the correct number of arguments for functions\n- Casting to the correct data type\n- Using the proper columns for joins\n\nIf there are any of the above mistakes, rewrite the query. If there are no mistakes, just reproduce the original query.\n\nOutput the final SQL query only.\n\nSQL Query: '), llm=ChatWatsonx(model_id='mistralai/mistral-medium-2505', apikey=SecretStr('**********'), params={}, watsonx_model=<ibm_watsonx_ai.foundation_models.inference.model_inference.ModelInference object at 0x309c34690>, watsonx_client=<ibm_watsonx_ai.client.APIClient object at 0x30c3e3250>), output_parser=StrOutputParser(), llm_kwargs={}))]

툴킷의 일부로 총 4개의 툴이 제공되는 것을 확인할 수 있습니다. 각 툴은 설명에서 볼 수 있듯이 고유한 목적을 가지고 있습니다. 이 툴들은 데이터베이스 목록을 조회하고, 쿼리를 실행하며, 테이블 스키마를 반환하고, SQL 쿼리를 실행 전에 검증하도록 설계되었습니다.

에이전트가 Python 코드를 생성하고 실행할 수 있도록 LangChain의 PythonREPLTool 클래스를 활용할 수 있습니다. 이 코드는 Python REPL(Read-Eval-Print Loop) 툴을 설정하고, 해당 기능을 정의한 뒤 나중에 사용하기 위해 툴 목록에 추가합니다.

python_repl = PythonREPLTool()
tools.append(python_repl)

에이전트가 SQL 툴과 데이터베이스에 맞게 설계된 시스템 프롬프트를 제공받도록 하기 위해, LangChain Hub에서 제공하는 프롬프트를 가져올 수 있습니다. 프롬프트를 자세히 보기 위해 system_message를 출력해 보겠습니다.

chatprompttemplate = hub.pull("langchain-ai/sql-agent-system-prompt")  
system_message = chatprompttemplate.format(dialect="SQLite", top_k=5)  
print(system_message)

아웃풋:

System: You are an agent designed to interact with a SQL database.
Given an input question, create a syntactically correct SQLite 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 5 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 have access to tools for interacting with the database.
Only use the below tools. Only use the information returned by the below tools to construct your final answer.
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.

7단계. 에이전트 상태 정의

LangGraph에는 ReAct(추론 및 행동) 아키텍처를 따르는 에이전트 그래프를 생성하는 미리 빌드된 create_react_agent 함수가 제공됩니다. 이 아키텍처를 통해 에이전트는 중단 조건이 충족될 때까지 반복적으로 툴을 호출할 수 있습니다.

ReAct 에이전트 흐름

보다 실습 중심의 접근을 위해 이 튜토리얼에서는 ReAct 에이전트를 처음부터 직접 구축합니다. 첫 단계로 사용자, 툴, 그리고 에이전트 자체에서 발생하는 메시지의 컨텍스트를 저장하기 위한 AgentState 클래스를 생성할 수 있습니다. 여기서는 메시지가 적절한 딕셔너리 형식을 따르도록 보장하기 위해 Python의 TypedDict 클래스를 사용합니다. 또한 LangGraph의 add_messages 리듀서 함수를 사용하여 새 메시지를 기존 메시지 목록에 추가할 수 있습니다.

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

8단계. ReActAgent 클래스 정의

다음으로 ReActAgent 클래스를 생성할 수 있습니다. 이 클래스는 주변 환경의 변화에 반응할 수 있는 에이전트를 생성하기 위한 기본 구조를 제공합니다. ReActAgent 클래스의 함수들은 그래프 상태에 따라 반복적으로 툴을 호출할 수 있게 합니다.

init 함수는 대규모 언어 모델, 툴, 시스템 메시지를 매개변수로 받아 클래스의 속성을 초기화합니다. 이 생성자는 가디언 모델, LLM, 그리고 툴에 대한 노드를 포함하는 상태 그래프를 생성합니다. 그래프는 guardian 노드에서 시작되며, 이 노드는 LLM과 데이터베이스에 도달하기 전에 유해한 콘텐츠를 탐지하기 위해 guardian_moderation 메서드를 호출합니다. guardian 노드와 llm 노드 사이의 조건부 엣지는 그래프의 상태를 llm 노드 또는 종료 지점으로 라우팅합니다. 이는 guardian_moderation 함수의 출력에 따라 결정됩니다. 안전한 메시지는 llm 노드로 전달되며, 이 노드는 call_llm 메서드를 실행합니다. 또한 메시지를 적절히 라우팅하기 위해 llm 노드와 tools 노드 사이에 조건부 엣지를 추가합니다. LLM이 툴 호출을 반환하면 should_call_tools 메서드는 True를 반환합니다. 그렇지 않으면 False가 반환되고 그래프는 종료로 이동합니다. 이 단계는 ReAct 에이전트 아키텍처의 일부로, 에이전트가 툴의 출력을 받은 후 상태 변화를 반영하여 다음 행동을 결정하도록 하기 위함입니다.

다음으로 그래프를 컴파일하면 이후 단계에서 에이전트를 호출할 수 있습니다. 메시지를 유지하기 위해 MemorySaver 체크포인터를 사용할 수 있습니다. 이 메서드의 마지막 두 줄은 클래스 인스턴스의 tools 및 llm 속성을 초기화합니다. tools 속성은 툴 이름을 툴 객체에 매핑하는 딕셔너리입니다. llm 속성은 bind_tools 메서드를 사용해 툴과 바인딩된 LLM입니다.

class ReActAgent:
    
    def __init__(self, llm, tools, system_message=""):
        memory = MemorySaver()
        graph = StateGraph(AgentState)
        graph.add_node("guardian", self.guardian_moderation)
        graph.add_node("llm", self.call_llm)
        graph.add_node("tools", self.call_tools)
        graph.add_node("block_message", self.block_message)
        graph.add_conditional_edges(
            "guardian",
            lambda state: state["moderation_verdict"],  
            {
                "inappropriate": "block_message",  
                "safe": "llm"           
            }
        )
        graph.add_edge("block_message", END)
        graph.add_conditional_edges(
            "llm",
            self.should_call_tools,
            ["tools", END]
        )
        graph.add_edge("tools", "llm")
        graph.add_edge(START, "guardian")
        self.system_message = system_message
        self.graph = graph.compile(checkpointer=memory)
        self.tools = {t.name: t for t in tools}
        self.llm = llm.bind_tools(tools)

ReActAgent 클래스의 다음 함수는 call_llm입니다. 이 함수는 상태에서 메시지를 가져와 LLM을 호출합니다. 시스템 메시지가 있는 경우, 이 메서드는 이를 메시지 목록의 맨 앞에 추가합니다. 그런 다음 메시지와 함께 LLM이 호출되며, LLM 응답이 포함된 새로운 상태가 반환됩니다.

def call_llm(self, state: AgentState):
    messages = state['messages']
    if self.system_message:
        messages = [SystemMessage(content=self.system_message)] + messages
    message = self.llm.invoke(messages)
    return {'messages': [message]}

ReActAgent 클래스의 다음 함수는 call_tools입니다. 이 메서드는 상태의 마지막 메시지에서 툴 호출을 가져와 반복 처리하며, 주어진 인수로 각 툴을 호출합니다. 다음으로 각 툴 호출의 결과는 results라는 리스트에 저장됩니다. 마지막으로 messages 키가 results 리스트에 매핑된 딕셔너리 형태로 새로운 상태가 반환됩니다.

def call_tools(self, state: AgentState):
    tool_calls = state['messages'][-1].tool_calls
    results = []
    for t in tool_calls:
        result = self.tools[t['name']].invoke(t['args'])
        results.append(ToolMessage(tool_call_id=t['id'], 
                                    name=t['name'], 
                                    content=str(result)))
    return {'messages': results}

ReActAgent 클래스의 다음 함수는 should_call_tools입니다. 이 함수는 상태에서 이전 LLM 응답을 가져와 툴 호출이 포함되어 있는지 확인함으로써 툴을 호출할지 여부를 결정합니다.

def should_call_tools(self, state: AgentState):
    result = state['messages'][-1]
    return "tools" if len(result.tool_calls) > 0 else END

guardrain 노드에서 실행되는 guardian_moderation 함수는 원치 않거나 민감한 콘텐츠를 탐지하고 차단하기 위해 가디언 시스템을 사용하여 메시지를 검토하도록 설계되었습니다. 먼저 마지막 메시지를 가져옵니다. 다음으로 탐지기 구성과 임계값을 포함하는 detectors라는 딕셔너리를 정의합니다. 이 탐지기들은 개인 식별 정보(PII), 증오 발언, 학대성 언어, 욕설(HAP)과 같은 특정 유형의 콘텐츠를 메시지에서 식별합니다. 다음으로 client라는 이름의 api_client 객체와 detectors 딕셔너리를 전달하여 Guardian 클래스의 인스턴스를 생성합니다. Guardian 인스턴스의 detect 메서드를 호출하여 마지막 메시지의 콘텐츠와 detectors 딕셔너리를 전달합니다. 이 메서드는 Granite Guardian 모델의 출력에 따라 moderation_verdict 키에 "safe" 또는 "inappropriate" 값을 저장한 딕셔너리를 반환합니다.

def guardian_moderation(self, state: AgentState):
    message = state['messages'][-1]  
    detectors = {
        "granite_guardian": {"threshold": 0.4},
        "hap": {"threshold": 0.4},
        "pii": {},
    }
    guardian = Guardian(
        api_client=client,  
        detectors=detectors 
    )
    response = guardian.detect(
        text=message.content,
        detectors=detectors
    )
    if len(response['detections']) != 0 and response['detections'][0]['detection'] == "Yes":
        return {"moderation_verdict": "inappropriate"}
    else:
        return {"moderation_verdict": "safe"}

block_message 함수는 사용자의 입력 쿼리에 부적절한 콘텐츠가 포함되어 있어 차단되었음을 알리는 알림 메커니즘 역할을 합니다.

def block_message(self, state: AgentState):
    return {"messages": [AIMessage(content="This message has been blocked due to inappropriate content.")]}

이제 이 모든 코드를 하나로 모아 다음 셀을 실행할 수 있습니다.

class ReActAgent:
    
    def __init__(self, llm, tools, system_message=""):
        memory = MemorySaver()
        graph = StateGraph(AgentState)
        graph.add_node("guardian", self.guardian_moderation)
        graph.add_node("llm", self.call_llm)
        graph.add_node("tools", self.call_tools)
        graph.add_node("block_message", self.block_message)
        graph.add_conditional_edges(
            "guardian",
            lambda state: state["moderation_verdict"],  
            {
                "inappropriate": "block_message",  
                "safe": "llm"           
            }
        )
        graph.add_edge("block_message", END)
        graph.add_conditional_edges(
            "llm",
            self.should_call_tools,
            ["tools", END]
        )
        graph.add_edge("tools", "llm")
        graph.add_edge(START, "guardian")
        self.system_message = system_message
        self.graph = graph.compile(checkpointer=memory)
        self.tools = {t.name: t for t in tools}
        self.llm = llm.bind_tools(tools)
    
    def call_llm(self, state: AgentState):
        messages = state['messages']
        if self.system_message:
            messages = [SystemMessage(content=self.system_message)] + messages
        message = self.llm.invoke(messages)
        return {'messages': [message]}
    
    def call_tools(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []
        for t in tool_calls:
            result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], 
                                       name=t['name'], 
                                       content=str(result)))
        return {'messages': results}
    
    def should_call_tools(self, state: AgentState):
        result = state['messages'][-1]
        return "tools" if len(result.tool_calls) > 0 else END
    
    def guardian_moderation(self, state: AgentState):
        message = state['messages'][-1]  
        detectors = {
            "granite_guardian": {"threshold": 0.4},
            "hap": {"threshold": 0.4},
            "pii": {},
        }
        guardian = Guardian(
            api_client=client,  
            detectors=detectors 
        )
        response = guardian.detect(
            text=message.content,
            detectors=detectors
        )
        if len(response['detections']) != 0 and response['detections'][0]['detection'] == "Yes":
            return {"moderation_verdict": "inappropriate"}
        else:
            return {"moderation_verdict": "safe"}
        
    def block_message(self, state: AgentState):
        return {"messages": [AIMessage(content="This message has been blocked due to inappropriate content.")]}

9단계. ReActAgent 객체 생성 및 호출

다음 코드 블록의 첫 번째 줄에서는 LLM, SQL 툴, 시스템 메시지를 매개변수로 전달하여 ReActAgent 클래스의 인스턴스를 생성합니다. 다음으로 그래프 상태를 메모리에 저장할 스레드를 지정합니다. 각 thread_id는 새로운 채팅 창을 나타낸다고 생각하면 됩니다. 또한 사용자 입력을 임의의 문자열로 정의할 수 있습니다. 다음으로 HumanMessage 타입의 사용자 입력으로 구성된 리스트를 전달하여 에이전트를 호출할 수 있습니다.

먼저 Granite Guardian 모델에 의해 차단되어야 하는 프롬프트를 사용해 보겠습니다.

agent = ReActAgent(llm, tools, system_message=system_message)

config = {"configurable": {"thread_id": "1"}}

user_input = "What is the home address of the customer who purchased the most expensive car last month?"

result = agent.graph.invoke({'messages': [HumanMessage(content=user_input)]}, config)

for message in result["messages"]:
    message.pretty_print()

아웃풋:

    ================================ [1m Human Message  [0m=================================
    
    What is the home address of the customer who purchased the most expensive car last month?
    ================================== [1m Ai Message  [0m==================================
    
    This message has been blocked due to inappropriate content.

Granite Guardian 모델은 사용자가 민감한 고객 정보를 요청하는 것을 성공적으로 차단할 수 있었습니다. 대화를 종료하기 전에 그래프가 노드에 도달하지 않은 것을 볼 수 있습니다. 다음으로 다른 스레드에서 적절한 질문을 드리겠습니다. 예: "2022년 실적 상위 5개 대리점의 총 판매 수익은 얼마인가요?" 사용자 입력으로 사용할 수 있습니다.

user_input = "What is the total sales revenue for the top 5 performing dealerships in the year 2022?"

config2 = {"configurable": {"thread_id": "2"}}

result = agent.graph.invoke({'messages': [HumanMessage(content=user_input)]}, config2)

for message in result["messages"]:
    message.pretty_print()

아웃풋:

 ================================ [1m Human Message  [0m=================================

What is the total sales revenue for the top 5 performing dealerships in the year 2022?
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_list_tables (H2irrlsPM)
 Call ID: H2irrlsPM
  Args:
    tool_input:
================================= [1m Tool Message  [0m=================================
Name: sql_db_list_tables

Cars, Dealerships, Sales
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_schema (ILZWiZvGX)
 Call ID: ILZWiZvGX
  Args:
    table_names: Cars, Dealerships, Sales
================================= [1m Tool Message  [0m=================================
Name: sql_db_schema


CREATE TABLE "Cars" (
	id INTEGER, 
	make TEXT NOT NULL, 
	model TEXT NOT NULL, 
	year INTEGER NOT NULL, 
	color TEXT NOT NULL, 
	PRIMARY KEY (id)
)

/*
3 rows from Cars table:
id	make	model	year	color
1	Toyota	Camry	2020	Blue
2	Honda	Civic	2019	Red
3	Ford	Mustang	2021	Black
*/


CREATE TABLE "Dealerships" (
	id INTEGER, 
	name TEXT NOT NULL, 
	location TEXT NOT NULL, 
	PRIMARY KEY (id)
)

/*
3 rows from Dealerships table:
id	name	location
1	Dealership A	New York
2	Dealership B	Los Angeles
3	Dealership C	Chicago
*/


CREATE TABLE "Sales" (
	id INTEGER, 
	car_id INTEGER NOT NULL, 
	dealership_id INTEGER NOT NULL, 
	sale_date DATE NOT NULL, 
	sale_price REAL NOT NULL, 
	PRIMARY KEY (id), 
	FOREIGN KEY(dealership_id) REFERENCES "Dealerships" (id), 
	FOREIGN KEY(car_id) REFERENCES "Cars" (id)
)

/*
3 rows from Sales table:
id	car_id	dealership_id	sale_date	sale_price
1	1	1	2022-01-01	25000.0
2	2	2	2022-02-01	20000.0
3	3	3	2022-03-01	30000.0
*/
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_query_checker (yIZ0tk4VP)
 Call ID: yIZ0tk4VP
  Args:
    query: SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue FROM Dealerships INNER JOIN Sales ON Dealerships.id = Sales.dealership_id WHERE strftime('%Y', Sales.sale_date) = '2022' GROUP BY Dealerships.id ORDER BY total_sales_revenue DESC LIMIT 5;
================================= [1m Tool Message  [0m=================================
Name: sql_db_query_checker

SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue
FROM Dealerships
INNER JOIN Sales ON Dealerships.id = Sales.dealership_id
WHERE strftime('%Y', Sales.sale_date) = '2022'
GROUP BY Dealerships.id
ORDER BY total_sales_revenue DESC
LIMIT 5;
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_query (cTzJFfvTl)
 Call ID: cTzJFfvTl
  Args:
    query: SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue FROM Dealerships INNER JOIN Sales ON Dealerships.id = Sales.dealership_id WHERE strftime('%Y', Sales.sale_date) = '2022' GROUP BY Dealerships.id ORDER BY total_sales_revenue DESC LIMIT 5;
================================= [1m Tool Message  [0m=================================
Name: sql_db_query

[('Dealership A', 65000.0), ('Dealership B', 48000.0), ('Dealership J', 38000.0), ('Dealership G', 35000.0), ('Dealership H', 32000.0)]
================================== [1m Ai Message  [0m==================================

The total sales revenue for the top 5 performing dealerships in the year 2022 are:

1. Dealership A: $65,000
2. Dealership B: $48,000
3. Dealership J: $38,000
4. Dealership G: $35,000
5. Dealership H: $32,000

================================ [1m Human Message  [0m=================================

What is the total sales revenue for the top 5 performing dealerships in the year 2022?
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_list_tables (H2irrlsPM)
 Call ID: H2irrlsPM
  Args:
    tool_input:
================================= [1m Tool Message  [0m=================================
Name: sql_db_list_tables

Cars, Dealerships, Sales
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_schema (ILZWiZvGX)
 Call ID: ILZWiZvGX
  Args:
    table_names: Cars, Dealerships, Sales
================================= [1m Tool Message  [0m=================================
Name: sql_db_schema


CREATE TABLE "Cars" (
	id INTEGER, 
	make TEXT NOT NULL, 
	model TEXT NOT NULL, 
	year INTEGER NOT NULL, 
	color TEXT NOT NULL, 
	PRIMARY KEY (id)
)

/*
3 rows from Cars table:
id	make	model	year	color
1	Toyota	Camry	2020	Blue
2	Honda	Civic	2019	Red
3	Ford	Mustang	2021	Black
*/


CREATE TABLE "Dealerships" (
	id INTEGER, 
	name TEXT NOT NULL, 
	location TEXT NOT NULL, 
	PRIMARY KEY (id)
)

/*
3 rows from Dealerships table:
id	name	location
1	Dealership A	New York
2	Dealership B	Los Angeles
3	Dealership C	Chicago
*/


CREATE TABLE "Sales" (
	id INTEGER, 
	car_id INTEGER NOT NULL, 
	dealership_id INTEGER NOT NULL, 
	sale_date DATE NOT NULL, 
	sale_price REAL NOT NULL, 
	PRIMARY KEY (id), 
	FOREIGN KEY(dealership_id) REFERENCES "Dealerships" (id), 
	FOREIGN KEY(car_id) REFERENCES "Cars" (id)
)

/*
3 rows from Sales table:
id	car_id	dealership_id	sale_date	sale_price
1	1	1	2022-01-01	25000.0
2	2	2	2022-02-01	20000.0
3	3	3	2022-03-01	30000.0
*/
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_query_checker (yIZ0tk4VP)
 Call ID: yIZ0tk4VP
  Args:
    query: SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue FROM Dealerships INNER JOIN Sales ON Dealerships.id = Sales.dealership_id WHERE strftime('%Y', Sales.sale_date) = '2022' GROUP BY Dealerships.id ORDER BY total_sales_revenue DESC LIMIT 5;
================================= [1m Tool Message  [0m=================================
Name: sql_db_query_checker

SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue
FROM Dealerships
INNER JOIN Sales ON Dealerships.id = Sales.dealership_id
WHERE strftime('%Y', Sales.sale_date) = '2022'
GROUP BY Dealerships.id
ORDER BY total_sales_revenue DESC
LIMIT 5;
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_query (cTzJFfvTl)
 Call ID: cTzJFfvTl
  Args:
    query: SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue FROM Dealerships INNER JOIN Sales ON Dealerships.id = Sales.dealership_id WHERE strftime('%Y', Sales.sale_date) = '2022' GROUP BY Dealerships.id ORDER BY total_sales_revenue DESC LIMIT 5;
================================= [1m Tool Message  [0m=================================
Name: sql_db_query

[('Dealership A', 65000.0), ('Dealership B', 48000.0), ('Dealership J', 38000.0), ('Dealership G', 35000.0), ('Dealership H', 32000.0)]
================================== [1m Ai Message  [0m==================================

The total sales revenue for the top 5 performing dealerships in the year 2022 are:

1. Dealership A: $65,000
2. Dealership B: $48,000
3. Dealership J: $38,000
4. Dealership G: $35,000
5. Dealership H: $32,000

훌륭합니다! 에이전트는 여러 도구를 호출하여 SQL 테이블 스키마를 추출하고, SQL 쿼리를 새로 생성하고, 실행하기 전에 정확성을 확인하는 등 일련의 작업을 성공적으로 실행했습니다. 그 결과, 에이전트는 2022년 데이터베이스 내 실적 상위 5개 자동차 딜러의 정확한 총 판매 수익을 반환합니다. 생성된 각 SQL 쿼리를 인쇄할 때 에이전트의 다단계 추론을 볼 수 있습니다. 또한 응답에서 Granite Guardian 모델이 사용자 쿼리가 적절하다고 판단한 것을 확인할 수 있습니다.

후속 질문을 해보겠습니다. 이번에는 에이전트에게 결과를 막대그래프로 표시하는 Python 코드를 생성하고 실행하도록 요청해 보겠습니다. 같은 thread_id를 사용하고 있으므로 에이전트는 이전 메시지를 가져와야 합니다.

user_input = "Write Python code to plot these results on a bar graph. Then, you must execute the code and display the bar graph."

result = agent.graph.invoke({'messages': [HumanMessage(content=user_input)]}, config2)

for message in result["messages"]:
    message.pretty_print()

아웃풋:

================================ [1m Human Message  [0m=================================

What is the total sales revenue for the top 5 performing dealerships in the year 2022?
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_list_tables (H2irrlsPM)
 Call ID: H2irrlsPM
  Args:
    tool_input:
================================= [1m Tool Message  [0m=================================
Name: sql_db_list_tables

Cars, Dealerships, Sales
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_schema (ILZWiZvGX)
 Call ID: ILZWiZvGX
  Args:
    table_names: Cars, Dealerships, Sales
================================= [1m Tool Message  [0m=================================
Name: sql_db_schema


CREATE TABLE "Cars" (
	id INTEGER, 
	make TEXT NOT NULL, 
	model TEXT NOT NULL, 
	year INTEGER NOT NULL, 
	color TEXT NOT NULL, 
	PRIMARY KEY (id)
)

/*
3 rows from Cars table:
id	make	model	year	color
1	Toyota	Camry	2020	Blue
2	Honda	Civic	2019	Red
3	Ford	Mustang	2021	Black
*/


CREATE TABLE "Dealerships" (
	id INTEGER, 
	name TEXT NOT NULL, 
	location TEXT NOT NULL, 
	PRIMARY KEY (id)
)

/*
3 rows from Dealerships table:
id	name	location
1	Dealership A	New York
2	Dealership B	Los Angeles
3	Dealership C	Chicago
*/


CREATE TABLE "Sales" (
	id INTEGER, 
	car_id INTEGER NOT NULL, 
	dealership_id INTEGER NOT NULL, 
	sale_date DATE NOT NULL, 
	sale_price REAL NOT NULL, 
	PRIMARY KEY (id), 
	FOREIGN KEY(dealership_id) REFERENCES "Dealerships" (id), 
	FOREIGN KEY(car_id) REFERENCES "Cars" (id)
)

/*
3 rows from Sales table:
id	car_id	dealership_id	sale_date	sale_price
1	1	1	2022-01-01	25000.0
2	2	2	2022-02-01	20000.0
3	3	3	2022-03-01	30000.0
*/
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_query_checker (yIZ0tk4VP)
 Call ID: yIZ0tk4VP
  Args:
    query: SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue FROM Dealerships INNER JOIN Sales ON Dealerships.id = Sales.dealership_id WHERE strftime('%Y', Sales.sale_date) = '2022' GROUP BY Dealerships.id ORDER BY total_sales_revenue DESC LIMIT 5;
================================= [1m Tool Message  [0m=================================
Name: sql_db_query_checker

SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue
FROM Dealerships
INNER JOIN Sales ON Dealerships.id = Sales.dealership_id
WHERE strftime('%Y', Sales.sale_date) = '2022'
GROUP BY Dealerships.id
ORDER BY total_sales_revenue DESC
LIMIT 5;
================================== [1m Ai Message  [0m==================================
Tool Calls:
  sql_db_query (cTzJFfvTl)
 Call ID: cTzJFfvTl
  Args:
    query: SELECT Dealerships.name, SUM(Sales.sale_price) AS total_sales_revenue FROM Dealerships INNER JOIN Sales ON Dealerships.id = Sales.dealership_id WHERE strftime('%Y', Sales.sale_date) = '2022' GROUP BY Dealerships.id ORDER BY total_sales_revenue DESC LIMIT 5;
================================= [1m Tool Message  [0m=================================
Name: sql_db_query

[('Dealership A', 65000.0), ('Dealership B', 48000.0), ('Dealership J', 38000.0), ('Dealership G', 35000.0), ('Dealership H', 32000.0)]
================================== [1m Ai Message  [0m==================================

The total sales revenue for the top 5 performing dealerships in the year 2022 are:

1. Dealership A: $65,000
2. Dealership B: $48,000
3. Dealership J: $38,000
4. Dealership G: $35,000
5. Dealership H: $32,000
================================ [1m Human Message  [0m=================================

Write Python code to plot these results on a bar graph. Then, you must execute the code and display the bar graph.
================================== [1m Ai Message  [0m==================================
Tool Calls:
  Python_REPL (5X1fPIODL)
 Call ID: 5X1fPIODL
  Args:
    query: import matplotlib.pyplot as plt

# Data for total sales revenue
dealerships = ['Dealership A', 'Dealership B', 'Dealership J', 'Dealership G', 'Dealership H']
sales_revenue = [65000, 48000, 38000, 35000, 32000]

# Create a bar graph
plt.bar(dealerships, sales_revenue, color='blue')

# Customize the plot
plt.title('Total Sales Revenue for Top 5 Performing Dealerships in 2022')
plt.xlabel('Dealerships')
plt.ylabel('Total Sales Revenue ($)')
plt.xticks(rotation=45)
plt.grid(axis='y')

# Display the plot
plt.tight_layout()
plt.show()
================================= [1m Tool Message  [0m=================================
Name: Python_REPL


================================== [1m Ai Message  [0m==================================

Your requirements have been fulfilled. Code has been executed in repl.
The bar graph has been plotted and shown in the python REPL.
The graph can also be seen below:

예상대로 에이전트는 python_repl 툴을 성공적으로 호출하여 Python 코드를 생성하고 실행했으며, 그 결과를 그래픽으로 표현했습니다. 또한 에이전트가 적절한 x축과 y축 값, 레이블, 제목을 스스로 결정할 수 있었음을 확인할 수 있습니다. 이는 AI 에이전트를 기존 LLM 챗봇과 구분 짓는 에이전트형 추론 능력을 잘 보여줍니다.

에이전트 그래프를 시각적으로 확인하기 위해 그래프 흐름을 표시할 수 있습니다.

display(Image(agent.graph.get_graph().draw_mermaid_png()))

아웃풋:

LangGraph 에이전트 그래프 시각화

요약

이 튜토리얼에서는 LangGraph를 사용하여 텍스트를 SQL로 변환하는 ReAct 에이전트를 구축하는 방법을 살펴보았습니다. 미리 빌드된 에이전트 실행 함수인 create_react_agent를 사용하는 대신, 에이전트를 처음부터 직접 구축했습니다. 자연어로 작성된 사용자의 질문을 해석한 후, 인공지능 에이전트는 SQL 쿼리 생성, 검증, 실행을 위해 툴을 사용하며 SQL과 Python 모두에 대한 역량을 보여줍니다. LangChain의 PythonREPLTool 클래스를 통해 에이전트는 Python 코드를 생성하는 것뿐만 아니라 실행하는 기능도 갖추게 됩니다. 에이전트에게 응답의 그래픽 표현을 요청했을 때 이 툴이 성공적으로 실행되는 것을 확인했습니다. LLM 노드와 툴 노드 사이의 조건부 루프는 ReAct 에이전트 아키텍처를 가능하게 합니다. 이 튜토리얼을 통해 Mistral Medium 3의 코딩 및 멀티모달 추론 역량을 확인했습니다. 다음 단계로 그래프에 추가 노드와 엣지를 실험해 보는 것을 고려해 보세요.