LangChain Skills 模式实战：构建按需加载知识的 SQL 助手
2/13/2026
8 min read
В попередній статті ми досліджували, як імітувати режим використання Skills за допомогою Deep Agents CLI. Сьогодні 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. З розвитком бізнесу, коли кількість таблиць розширюється до сотень, цей метод створює значні проблеми:
- **Величезне споживання токенів: кожна розмова містить велику кількість нерелевантних структур таблиць, що призводить до марної витрати ресурсів.**
- **Збільшення ризику галюцинацій: надмірна кількість нерелевантної інформації знижує точність міркувань моделі.**
- **Складність підтримки: знання всіх бізнес-ліній тісно пов'язані, що ускладнює незалежну ітерацію.**
### **Режим Skills: рішення на основі поступового розкриття**
Режим Skills базується на принципі **поступового розкриття (Progressive Disclosure)**, який обробляє процес отримання знань пошарово:
- **Початковий стан Agent: знає лише, які «навички» (Skills) існують та їх короткий опис (Description), залишаючись легким.**
- **Завантаження під час виконання: коли стикається з конкретною проблемою (наприклад, «запит на інвентар»), Agent активно викликає інструмент (**`load_skill`**) для завантаження детального контексту цієї навички (Schema + Prompt).**
- **Виконання завдання: на основі завантаженого точного контексту виконує конкретне завдання (наприклад, написання та виконання SQL).**
Цей режим ефективно підтримує **необмежене розширення** та **роз'єднання команд**, дозволяючи Agent адаптуватися до дедалі складніших бізнес-сценаріїв.
## **2. Проектування системної архітектури**
Цей практичний проект створить SQL Assistant, що містить дві основні Skills, щоб продемонструвати практичне застосування цього режиму:
- **Sales Analytics (аналіз продажів): відповідає за таблицю**`sales_data`**, обробляє статистику доходів, аналіз тенденцій замовлень тощо.**
- **Inventory Management (управління запасами): відповідає за таблицю**`inventory_items`**, обробляє моніторинг рівня запасів, запити на місцезнаходження тощо.**
## **3. Створення середовища розробки**
У цьому проекті використовується Python`uv` для ефективного управління залежностями.
### **Встановлення основних залежностей**
`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, щоб суворо дотримуватися стандартної операційної процедури (SOP) `Identify -> Load -> Query`.
`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)**
`import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
import os
from dotenv import load_dotenv
load_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")
def create_database():
try:
# Підключення до стандартної бази даних '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()
if not exists:
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()
except Exception as e:
print(f"Error creating database: {e}")
def create_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.")
except Exception as e:
print(f"Error setting up tables: {e}")
if __name__ == "__main__":
create_database()
create_tables_and_data()
`
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
Published in Technology
