LangChain Skills 模式实战：构建按需加载知识的 SQL 助手
Dans l'article précédent, nous avons exploré comment simuler le modèle d'utilisation des Skills par Deep Agent via Deep Agents CLI. Aujourd'hui, LangChain prend en charge nativement cette fonctionnalité, simplifiant considérablement le processus de développement. Cet article vous guidera à travers une expérience approfondie de cette fonctionnalité, en construisant un assistant SQL plus intelligent.
Lors de la construction d'un agent d'IA complexe, les développeurs sont souvent confrontés à un dilemme : injecter tout le contexte (structure de la table de base de données, documentation de l'API, règles métier) dans l'invite système en une seule fois, ce qui entraîne un débordement de la fenêtre de contexte (Context Window) et une dispersion de l'attention du modèle ? Ou choisir un réglage fin (Fine-tuning) fréquent et coûteux ?
Le modèle Skills (Skills Pattern) offre une voie intermédiaire élégante. Il réalise une utilisation efficace du contexte en chargeant dynamiquement les connaissances nécessaires. La prise en charge native de ce modèle par LangChain signifie que nous pouvons plus facilement construire un agent doté de la capacité d'« apprentissage à la demande ».
Cet article combinera la documentation officielle Build a SQL assistant with on-demand skills pour guider les lecteurs à partir de zéro, en construisant un assistant SQL qui prend en charge le « chargement de connaissances à la demande ».
1. Concept clé : Pourquoi choisir le modèle Skills ?
Limites de l'agent SQL traditionnel
Dans l'architecture traditionnelle de l'agent SQL, nous devons généralement fournir le schéma complet de la base de données dans l'invite système. Au fur et à mesure que l'entreprise se développe, lorsque le nombre de tables s'étend à des centaines, cette méthode pose des problèmes importants :
-
Consommation énorme de jetons : chaque conversation transporte un grand nombre de structures de table non pertinentes, ce qui entraîne un gaspillage de ressources.
-
Risque accru d'hallucinations : trop d'informations d'interférence non pertinentes réduiront la précision du raisonnement du modèle.
-
Maintenance difficile : toutes les connaissances des secteurs d'activité sont étroitement couplées, ce qui rend difficile l'itération indépendante.
Modèle Skills : une solution basée sur la divulgation progressive
Le modèle Skills est basé sur le principe de divulgation progressive (Progressive Disclosure), qui traite le processus d'acquisition de connaissances en couches :
-
État initial de l'agent : ne maîtrise que les « compétences » (Skills) disponibles et leurs brèves descriptions (Description), en restant léger.
-
Chargement au moment de l'exécution : face à un problème spécifique (tel que « interroger l'inventaire »), l'agent appelle activement l'outil (
load_skill) pour charger le contexte détaillé de cette compétence (Schéma + Invite). -
Exécution de la tâche : sur la base du contexte précis chargé, exécutez la tâche spécifique (telle que la rédaction et l'exécution de SQL).
Ce modèle prend en charge efficacement l'extension illimitée et le découplage d'équipe, permettant à l'agent de s'adapter à des scénarios commerciaux de plus en plus complexes.
2. Conception de l'architecture du système
Ce projet pratique construira un assistant SQL contenant deux Skills principaux pour démontrer l'application pratique de ce modèle :
-
Sales Analytics (Analyse des ventes) : responsable de la table
sales_data, traitant les statistiques de revenus, l'analyse des tendances des commandes, etc. -
Inventory Management (Gestion des stocks) : responsable de la table
inventory_items, traitant la surveillance du niveau des stocks, les requêtes de localisation, etc.
3. Configuration de l'environnement de développement
Ce projet utilise Python uv pour une gestion efficace des dépendances.
Installation des dépendances principales
uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community
Configuration de l'environnement PostgreSQL
Démarrez une instance Postgres locale et créez la base de données agent_platform. Nous fournissons un script setup_db.py pour initialiser automatiquement la structure de la table et les données de test (voir le code source à la fin de l'article).
4. Explication détaillée des étapes de mise en œuvre principales### Étape 1 : Définir les compétences du domaine (The Knowledge)
Nous définissons les compétences comme une structure de dictionnaire, simulant le processus de chargement à partir d'un système de fichiers ou d'une base de données. Veuillez noter la distinction entre description (utilisée par l'Agent pour la sélection de décision) et content (contexte détaillé réellement chargé).
SKILLS = {"sales_analytics": {"description":"Utile pour analyser les revenus des ventes, les tendances...","content":"""... Schéma de la table : sales_data ..."" },"inventory_management": {"description":"Utile pour vérifier les niveaux de stock...","content":"""... Schéma de la table : inventory_items ..."" }}
Étape 2 : Implémenter les outils principaux (The Capabilities)
L'Agent dépend de deux outils essentiels pour accomplir les tâches :
-
load_skill(skill_name): Charge dynamiquement les détails de la compétence spécifiée au moment de l'exécution. -
run_sql_query(query): Exécute des instructions SQL spécifiques.
Étape 3 : Orchestrer la logique de l'Agent (The Brain)
Utiliser LangGraph pour construire un Agent ReAct. Le System Prompt joue un rôle essentiel ici, guidant l'Agent à suivre strictement la procédure opérationnelle standard (SOP) Identifier -> Charger -> Interroger.
system_prompt ="""1. Identifier la compétence pertinente.2. Utiliser 'load_skill' pour obtenir le schéma.3. Écrire et exécuter SQL en utilisant 'run_sql_query'....Ne pas deviner les noms de table. Toujours charger la compétence en premier."""
5. Vérification de l'efficacité de l'exécution
En exécutant test_agent.py, nous avons testé les requêtes dans deux domaines différents : Ventes et Inventaire. Voici les journaux de sortie réels de la console, montrant comment l'Agent charge dynamiquement les compétences en fonction de la question :
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. Référence complète du code source
Voici le code source complet du projet, comprenant le script d'initialisation de la base de données et le programme principal de l'Agent.
1. Initialisation de la base de données (setup_db.py)
`import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT import os from dotenv import load_dotenv
load_dotenv()
Veuillez vous assurer que les informations de connexion à la base de données sont configurées dans .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") # Veuillez remplacer par le mot de passe réel DB_NAME = os.getenv("DB_NAME", "agent_platform")
def create_database(): try: # Connexion à la base de données par défaut 'postgres' pour créer une nouvelle base de données 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()
# Vérifier si la base de données existe
cur.execute(f"SELECT 1 FROM pg_catalog.pg_database WHERE datname = '{DB_NAME}'")
exists = cur.fetchone()
if not exists:
print(f"Création de la base de données {DB_NAME}...")
cur.execute(f"CREATE DATABASE {DB_NAME}")
else:
print(f"La base de données {DB_NAME} existe déjà.")
cur.close()
conn.close()
except Exception as e:
print(f"Erreur lors de la création de la base de données : {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()
# Créer la table Sales
print("Création de la table sales_data...")
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)
)
"""
)
# Créer la table Inventory
print("Création de la table inventory_items...")
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)
)
"""
)
# Insérer des données factices
print("Insertion de données factices...")
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("Configuration de la base de données terminée.")
except Exception as e:
print(f"Erreur lors de la configuration des 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": """Vous êtes un expert en analyse des ventes. Vous avez accès à la table 'sales_data'. Schéma de la table:
- id: integer (clé primaire)
- transaction_date: date
- product_id: varchar(50)
- amount: decimal(10, 2)
- region: varchar(50) Requêtes courantes:
- Revenu total: SUM(amount)
- Revenu par région: GROUP BY region
- Tendance des ventes: GROUP BY transaction_date""" }, "inventory_management": { "description": "Useful for checking stock levels, product locations, and warehouse management.", "content": """Vous êtes un expert en gestion des stocks. Vous avez accès à la table 'inventory_item""" }s' table.Schéma de la table :
- id : entier (clé primaire)
- product_id : varchar(50)
- product_name : varchar(100)
- stock_count : entier
- warehouse_location : varchar(50) Requêtes courantes :
- Vérifier le stock : WHERE product_name = '...'
- Stock faible : WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str :"""
Chargez l’invite détaillée et le schéma pour une compétence spécifique.
Compétences disponibles :
- sales_analytics : Pour l’analyse des ventes, des revenus et des transactions.
- inventory_management : Pour les requêtes sur les stocks, les produits et l’entrepôt. """ skill = SKILLS.get(skill_name) if not skill: return f"Erreur : La compétence « {skill_name} » est introuvable. Compétences disponibles : {list(SKILLS.keys())}" return skill["content"]
@tool def run_sql_query(query : str) -> str : """ Exécute une requête SQL sur la base de données. N’utilisez cet outil QU’APRÈS avoir chargé la compétence appropriée pour comprendre le schéma. """ try: return db.run(query) except Exception as e: return f"Erreur lors de l’exécution de SQL : {e}"
@tool def list_tables() -> str : """Répertorie toutes les tables disponibles dans la base de données.""" return str(db.get_usable_table_names())
tools = [load_skill, run_sql_query, list_tables]
--- Configuration de l’agent ---
llm = ChatOpenAI( base_url=BASE_URL, api_key=API_KEY, model=MODEL_NAME, temperature=0 )
llm_with_tools = llm.bind_tools(tools)
--- Définition du graphique ---
class AgentState(MessagesState):
Nous pouvons ajouter un état personnalisé si nécessaire, mais MessagesState est suffisant pour une conversation simple
pass
@tool def agent_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()# --- Exécution principale ---if__name__ =="main": system_prompt ="""Vous êtes un assistant SQL serviable.Vous avez accès à des compétences spécialisées qui contiennent des schémas de base de données et des connaissances du domaine.Pour répondre à la question d'un utilisateur :1. Identifiez la compétence pertinente (sales_analytics ou inventory_management).2. Utilisez l'outil 'load_skill' pour obtenir le schéma et les instructions.3. Sur la base de la compétence chargée, écrivez et exécutez une requête SQL en utilisant 'run_sql_query'.4. Répondez à la question de l'utilisateur en fonction des résultats de la requête.Ne devinez pas les noms des tables. Chargez toujours la compétence en premier.""" print("Assistant SQL initialisé. Tapez 'quit' pour quitter.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Vérification de la connexion pré-chaufféetry: print(f"Connecté à la base de données :{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Avertissement de connexion à la base de données :{e}")whileTrue:try: user_input = input("Utilisateur : ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Diffusez l'exécution print("Agent : ", end="", flush=True) final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# En mode 'values', nous obtenons l'état complet. Nous voulons juste voir le dernier message s'il est nouveau. last_message = event["messages"][-1]# Mettez à jour notre historique de messages avec le dernier étatpass# Une fois la diffusion terminée, le dernier état contient la réponse finale final_state = app.invoke({"messages": messages}) last_msg = final_state["messages"][-1] if isinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]# Mettre à jour l'historique print("-"*50) except Exception as e: print(f"\nErreur:{e}") break`
