LangChain Skills 모드 실전: 필요에 따라 지식을 로드하는 SQL 도우미 구축
이전 글에서 Deep Agents CLI를 통해 Deep Agent가 Skills를 사용하는 모드를 시뮬레이션하는 방법을 살펴보았습니다. 이제 LangChain은 이 기능을 기본적으로 지원하여 개발 프로세스를 크게 간소화했습니다. 이 글에서는 이 기능을 자세히 살펴보고 더 스마트한 SQL 도우미를 구축합니다.
복잡한 AI Agent를 구축할 때 개발자는 종종 딜레마에 빠집니다. 모든 컨텍스트(데이터베이스 테이블 구조, API 문서, 비즈니스 규칙)를 한 번에 System Prompt에 주입하여 컨텍스트 창(Context Window)이 넘치고 모델의 주의력을 분산시킬까요? 아니면 비용이 많이 드는 빈번한 미세 조정(Fine-tuning)을 선택할까요?
**Skills 모드(Skills Pattern)**는 우아한 중간 경로를 제공합니다. 필요한 지식을 동적으로 로드하여 컨텍스트를 효율적으로 활용합니다. LangChain의 이 모드에 대한 기본 지원은 "필요에 따라 학습"할 수 있는 Agent를 더 쉽게 구축할 수 있음을 의미합니다.
이 글에서는 공식 문서 Build a SQL assistant with on-demand skills를 통해 독자가 처음부터 "필요에 따라 지식 로드"를 지원하는 SQL Assistant를 구축하도록 안내합니다.
1. 핵심 개념: Skills 모드를 선택하는 이유는 무엇입니까?
기존 SQL Agent의 한계
기존 SQL Agent 아키텍처에서는 일반적으로 System Prompt에 전체 Database Schema를 제공해야 합니다. 비즈니스가 발전함에 따라 테이블 수가 수백 개로 확장되면 이 방법은 다음과 같은 중요한 문제를 야기합니다.
-
Token 소모가 큼: 대량의 관련 없는 테이블 구조를 매번 대화에 포함하여 리소스 낭비를 초래합니다.
-
환각 위험 증가: 너무 많은 관련 없는 간섭 정보는 모델의 추론 정확도를 떨어뜨립니다.
-
유지 관리 어려움: 모든 비즈니스 라인의 지식이 긴밀하게 결합되어 독립적으로 반복하기 어렵습니다.
Skills 모드: 점진적 공개 기반 솔루션
Skills 모드는 점진적 공개(Progressive Disclosure) 원칙을 기반으로 지식 습득 프로세스를 계층화합니다.
-
Agent 초기 상태: 어떤 "기술(Skills)"이 있는지와 간략한 설명(Description)만 파악하여 가볍게 유지합니다.
-
런타임 로드: 구체적인 문제(예: "재고 조회")에 직면하면 Agent는 도구(
load_skill)를 적극적으로 호출하여 해당 기술의 자세한 컨텍스트(Schema + Prompt)를 로드합니다. -
작업 실행: 로드된 정확한 컨텍스트를 기반으로 구체적인 작업(예: SQL 작성 및 실행)을 실행합니다.
이 모드는 무한 확장과 팀 분리를 효과적으로 지원하여 Agent가 점점 더 복잡해지는 비즈니스 시나리오에 적응할 수 있도록 합니다.
2. 시스템 아키텍처 설계
이 실전 프로젝트에서는 두 개의 핵심 Skills를 포함하는 SQL Assistant를 구축하여 이 모드의 실제 적용을 보여줍니다.
-
Sales Analytics(판매 분석):
sales_data** 테이블을 담당하며, 수익 통계, 주문 추세 분석 등을 처리합니다.** -
Inventory Management(재고 관리):
inventory_items** 테이블을 담당하며, 재고 수준 모니터링, 위치 조회 등을 처리합니다.**
3. 개발 환경 구축
이 프로젝트에서는 Pythonuv를 사용하여 효율적인 종속성 관리를 수행합니다.
핵심 종속성 설치
uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community
PostgreSQL 환경 구성
로컬에서 Postgres 인스턴스를 시작하고agent_platform 데이터베이스를 만듭니다. 테이블 구조와 테스트 데이터를 자동으로 초기화하는setup_db.py 스크립트를 제공합니다(자세한 내용은 글 말미의 소스 코드 참조).
4. 핵심 구현 단계 상세 설명### 1단계: 도메인 기술 정의 (The Knowledge)
기술을 딕셔너리 구조로 정의하여 파일 시스템 또는 데이터베이스에서 로드하는 과정을 시뮬레이션합니다. description (Agent 의사 결정에 사용)과 content (실제로 로드되는 상세 컨텍스트)를 구분하십시오.
SKILLS = {"sales_analytics": {"description":"매출 수익, 추세 분석에 유용...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"재고 수준 확인에 유용...","content":"""... Table Schema: inventory_items ..."" }}
2단계: 핵심 도구 구현 (The Capabilities)
Agent는 작업을 완료하기 위해 두 가지 핵심 도구에 의존합니다.
-
load_skill(skill_name)**: 런타임에 지정된 기술의 세부 정보를 동적으로 로드합니다. ** -
run_sql_query(query)**: 구체적인 SQL 문을 실행합니다. **
3단계: Agent 로직 편곡 (The Brain)
LangGraph를 사용하여 ReAct Agent를 구축합니다. System Prompt는 여기서 중요한 역할을 하며, Agent가 Identify -> Load -> Query 표준 작업 절차(SOP)를 엄격하게 따르도록 안내합니다.
system_prompt ="""1. Identify the relevant skill.2. Use 'load_skill' to get schema.3. Write and execute SQL using 'run_sql_query'....Do not guess table names. Always load the skill first.""
5. 실행 효과 검증
test_agent.py를 실행하여 Sales 및 Inventory의 두 가지 다른 영역에 대한 쿼리를 테스트했습니다. 다음은 콘솔의 실제 출력 로그로, Agent가 문제에 따라 기술을 동적으로 로드하는 방법을 보여줍니다.
Testing Sales Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'sales_analytics'}, 'id': 'call_f270d76b7ce4404cb5f61bf2', 'type': 'tool_call'}]Tool output:You are a Sales Analytics Expert.You have access to the 'sales_data' table.Table Schema:- id: integer...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': 'SELECT SUM(amount) as total_revenue FROM sales_data;'}, 'id': 'call_b4f3e686cc7f4f22b3bb9ea7', 'type': 'tool_call'}]Tool output: [(Decimal('730.50'),)]...Agent response: The total revenue is $730.50.Testing Inventory Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'inventory_management'}, 'id': 'call_18c823b2d5064e95a0cfe2e3', 'type': 'tool_call'}]Tool output:You are an Inventory Management Expert.You have access to the 'inventory_items' table.Table Schema...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': "SELECT warehouse_location FROM inventory_items WHERE product_name = 'Laptop';"}, 'id': 'call_647ee3a444804bd98a045f00', 'type': 'tool_call'}]Tool output: [('Warehouse A',)]...Agent response: The Laptop is located in **Warehouse A**.## 6. 전체 소스 코드 참조
다음은 프로젝트의 전체 소스 코드이며, 데이터베이스 초기화 스크립트와 Agent 주 프로그램을 포함합니다.
1. 데이터베이스 초기화 (setup_db.py)
importpsycopg2frompsycopg2.extensionsimportISOLATION_LEVEL_AUTOCOMMITimportosfromdotenvimportload_dotenvload_dotenv()# .env 파일에 데이터베이스 연결 정보가 구성되어 있는지 확인하십시오.DB_HOST = os.getenv("DB_HOST","localhost")DB_PORT = os.getenv("DB_PORT","5432")DB_USER = os.getenv("DB_USER","postgres")DB_PASSWORD = os.getenv("DB_PASSWORD","your_password")# 실제 비밀번호로 바꾸십시오.DB_NAME = os.getenv("DB_NAME","agent_platform")defcreate_database():try:# 새 DB를 생성하기 위해 기본 'postgres' 데이터베이스에 연결 conn = psycopg2.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, dbname="postgres" ) conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = conn.cursor()# 데이터베이스가 존재하는지 확인 cur.execute(f"SELECT 1 FROM pg_catalog.pg_database WHERE datname = '{DB_NAME}'") exists = cur.fetchone()ifnotexists: print(f"Creating database{DB_NAME}...") cur.execute(f"CREATE DATABASE{DB_NAME}")else: print(f"Database{DB_NAME}already exists.") cur.close() conn.close()exceptExceptionase: print(f"Error creating database:{e}")defcreate_tables_and_data():try: conn = psycopg2.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, dbname=DB_NAME ) cur = conn.cursor()# Sales 테이블 생성 print("Creating sales_data table...") cur.execute(""" CREATE TABLE IF NOT EXISTS sales_data ( id SERIAL PRIMARY KEY, transaction_date DATE, product_id VARCHAR(50), amount DECIMAL(10, 2), region VARCHAR(50) ) """)# Inventory 테이블 생성 print("Creating inventory_items table...") cur.execute(""" CREATE TABLE IF NOT EXISTS inventory_items ( id SERIAL PRIMARY KEY, product_id VARCHAR(50), product_name VARCHAR(100), stock_count INTEGER, warehouse_location VARCHAR(50) ) """)# 모의 데이터 삽입 print("Inserting mock data...") cur.execute("TRUNCATE sales_data, inventory_items") sales_data = [ ('2023-01-01','P001',100.00,'North'), ('2023-01-02','P002',150.50,'South'), ('2023-01-03','P001',120.00,'East'), ('2023-01-04','P003',200.00,'West'), ('2023-01-05','P002',160.00,'North') ] cur.executemany("INSERT INTO sales_data (transaction_date, product_id, amount, region) VALUES (%s, %s, %s, %s)", sales_data ) inventory_data = [ ('P001','Laptop',50,'Warehouse A'), ('P002','Mouse',200,'Warehouse B'), ('P003','Keyboard',150,'Warehouse A'), ('P004','Monitor',30,'Warehouse C') ] cur.executemany("INSERT INTO inventory_items (product_id, product_name, stock_count, warehouse_location) VALUES (%s, %s, %s, %s)", inventory_data ) conn.commit() cur.close() conn.close() print("Database setup complete.")exceptExceptionase: print(f"Error setting up tables:{e}")if__name__ =="__main__": create_database() create_tables_and_data()### 2. Agent 메인 프로그램 (main.py)
`import os from typing import Annotated, Literal, TypedDict, Union, Dict from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import SQLDatabaseToolkit from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.prebuilt import ToolNode, tools_condition
load_dotenv()
--- Configuration ---
BASE_URL = os.getenv("BASIC_MODEL_BASE_URL") API_KEY = os.getenv("BASIC_MODEL_API_KEY") MODEL_NAME = os.getenv("BASIC_MODEL_MODEL") DB_URI = f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
--- Database Setup ---
db = SQLDatabase.from_uri(DB_URI)
--- Skills Definition ---
SKILLS: Dict[str, Dict[str, str]] = { "sales_analytics": { "description": "Useful for analyzing sales revenue, trends, and regional performance.", "content": """ You are a Sales Analytics Expert. You have access to the 'sales_data' table. Table Schema:
- id: integer (primary key)
- transaction_date: date
- product_id: varchar(50)
- amount: decimal(10, 2)
- region: varchar(50)
Common queries:
- Total revenue: SUM(amount)
- Revenue by region: GROUP BY region
- Sales trend: GROUP BY transaction_date""" }, "inventory_management": { "description": "Useful for checking stock levels, product locations, and warehouse management.", "content": """ You are an Inventory Management Expert. You have access to the 'inventory_item""" s' table.Table Schema:- id: integer (primary key)- product_id: varchar(50)- product_name: varchar(100)- stock_count: integer- warehouse_location: varchar(50)Common queries:- Check stock: WHERE product_name = '...'- Low stock: WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:""" 특정 스킬에 대한 자세한 프롬프트와 스키마를 로드합니다. 사용 가능한 스킬: - sales_analytics: 판매, 수익 및 거래 분석용. - inventory_management: 재고, 제품 및 창고 쿼리용. """ skill = SKILLS.get(skill_name)ifnotskill:returnf"Error: Skill '{skill_name}'을(를) 찾을 수 없습니다. 사용 가능한 스킬: {list(SKILLS.keys())}"returnskill["content"]@tooldefrun_sql_query(query: str)-> str:""" 데이터베이스에 대해 SQL 쿼리를 실행합니다. 스키마를 이해하기 위해 적절한 스킬을 로드한 후에만 이 도구를 사용하십시오. """try:returndb.run(query)exceptExceptionase:returnf"Error executing SQL:{e}"@tooldeflist_tables()-> str:"""데이터베이스에서 사용 가능한 모든 테이블을 나열합니다."""returnstr(db.get_usable_table_names())tools = [load_skill, run_sql_query, list_tables]# --- Agent Setup ---llm = ChatOpenAI( base_url=BASE_URL, api_key=API_KEY, model=MODEL_NAME, temperature=0)llm_with_tools = llm.bind_tools(tools)# --- Graph Definition ---classAgentState(MessagesState):# 필요한 경우 사용자 정의 상태를 추가할 수 있지만 간단한 채팅에는 MessagesState로 충분합니다.passdefagent_node(state: AgentState): messages = state["messages"] response = llm_with_tools.invoke(messages)return{"messages": [response]}workflow = StateGraph(AgentState)workflow.add_node("agent", agent_node)workflow.add_node("tools", ToolNode(tools))workflow.add_edge(START,"agent")workflow.add_conditional_edges("agent", tools_condition)workflow.add_edge("tools","agent")app = workflow.compile()# --- Main Execution ---if__name__ =="main": system_prompt ="""You are a helpful SQL Assistant.You have access to specialized skills that contain database schemas and domain knowledge.To answer a user's question:1. Identify the relevant skill (sales_analytics or inventory_management).2. Use the 'load_skill' tool to get the schema and instructions.3. Based on the loaded skill, write and execute a SQL query using 'run_sql_query'.4. Answer the user's question based on the query results.Do not guess table names. Always load the skill first.""" print("SQL Assistant initialized. Type 'quit' to exit.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Pre-warm connection checktry: print(f"Connected to database:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Database connection warning:{e}")whileTrue:try: user_input = input("User: ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Stream the execution print("Agent: ", end="", flush=True) final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# In 'values' mode, we get the full state. We just want to see the last message if it's new. last_message = event["messages"][-1]# Update our message history with the latest statepass# After stream finishes, the last state has the final answer final_state = app.invoke({"messages": messages}) last_msg = final_state["messages"][-1] ifisinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]# Update history print("-"*50) exceptExceptionase: print(f"\nError:{e}") break`
