In een eerder artikel hebben we onderzocht hoe je met de Deep Agents CLI het gebruik van Skills door Deep Agents kunt simuleren. Nu ondersteunt LangChain deze functie native, wat het ontwikkelingsproces aanzienlijk vereenvoudigt. In dit artikel nemen we je mee om deze functie diepgaand te ervaren en een intelligentere SQL-assistent te bouwen.

Bij het bouwen van complexe AI-agents bevinden ontwikkelaars zich vaak in een dilemma: alle context (database tabelstructuur, API-documentatie, bedrijfsregels) in één keer in de System Prompt injecteren, wat leidt tot overloop van het contextvenster (Context Window) en afleiding van de modelaandacht? Of kiezen voor dure frequente fijnafstemming (Fine-tuning)?

Het Skills-patroon (Skills Pattern) biedt een elegante tussenoplossing. Het maakt efficiënt gebruik van de context mogelijk door de benodigde kennis dynamisch te laden. De native ondersteuning van LangChain voor dit patroon betekent dat we gemakkelijker agents kunnen bouwen met de mogelijkheid om "op aanvraag te leren".

Dit artikel combineert de officiële documentatie Build a SQL assistant with on-demand skills en begeleidt de lezer vanaf nul bij het bouwen van een SQL-assistent die "kennis op aanvraag laden" ondersteunt.

1. Kernconcepten: Waarom kiezen voor het Skills-patroon?

Beperkingen van traditionele SQL-agents

In de traditionele SQL-agentarchitectuur moeten we meestal de volledige databaseschema in de System Prompt opnemen. Naarmate het bedrijf groeit en het aantal tabellen uitbreidt tot honderden, brengt deze aanpak aanzienlijke problemen met zich mee:

Enorm tokenverbruik: elke dialoog bevat een grote hoeveelheid irrelevante tabelstructuren, wat leidt tot verspilling van middelen.
Verhoogd risico op hallucinaties: te veel irrelevante storende informatie vermindert de redeneernauwkeurigheid van het model.
Moeilijk te onderhouden: de kennis van alle bedrijfsonderdelen is nauw met elkaar verbonden, waardoor onafhankelijke iteratie moeilijk is.

Skills-patroon: een oplossing op basis van progressieve openbaarmaking

Het Skills-patroon is gebaseerd op het principe van progressieve openbaarmaking (Progressive Disclosure) en verwerkt het proces van kennisverwerving in lagen:

Initiële status van de agent: alleen beheersing van welke "vaardigheden" (Skills) er zijn en hun korte beschrijving (Description), waardoor het lichtgewicht blijft.
Runtime laden: wanneer de agent wordt geconfronteerd met een specifiek probleem (zoals "voorraad opvragen"), roept hij actief een tool aan (load_skill) om de gedetailleerde context van die vaardigheid te laden (Schema + Prompt).
Taak uitvoeren: voer specifieke taken uit (zoals het schrijven en uitvoeren van SQL) op basis van de geladen nauwkeurige context.

Dit patroon ondersteunt effectief onbeperkte uitbreiding en teamontkoppeling, waardoor de agent zich kan aanpassen aan steeds complexere bedrijfsscenario's.

2. Systeemarchitectuurontwerp

Dit praktische project bouwt een SQL-assistent met twee kernvaardigheden om de praktische toepassing van dit patroon te demonstreren:

Sales Analytics (verkoopanalyse): verantwoordelijk voor de tabel sales_data, het verwerken van inkomstenstatistieken, analyse van ordertrends, enz.
Inventory Management (voorraadbeheer): verantwoordelijk voor de tabel inventory_items, het verwerken van monitoring van voorraadniveaus, locatiequery's, enz.

3. Ontwikkelomgeving opzetten

Dit project gebruikt Pythonuv voor efficiënt afhankelijkheidsbeheer.

Kernafhankelijkheden installeren

uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community

PostgreSQL-omgeving configureren

Start lokaal een Postgres-instantie en maak de database agent_platform. We bieden het script setup_db.py om de tabelstructuur en testgegevens automatisch te initialiseren (zie de broncode aan het einde van het artikel).

4. Gedetailleerde uitleg van de belangrijkste implementatiestappen### Stap 1: Definieer Domeinvaardigheden (The Knowledge)

We definiëren vaardigheden als een dictionary-structuur, waarbij we het laden vanuit een bestandssysteem of database simuleren. Let op het onderscheid tussen description (voor Agent besluitvorming) en content (de daadwerkelijke gedetailleerde context).

SKILLS = {"sales_analytics": {"description":"Nuttig voor het analyseren van verkoopopbrengsten, trends...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"Nuttig voor het controleren van voorraadniveaus...","content":"""... Table Schema: inventory_items ..."" }}

Stap 2: Implementeer Kernhulpmiddelen (The Capabilities)

De Agent is afhankelijk van twee cruciale hulpmiddelen om taken te voltooien:

load_skill(skill_name): Laadt dynamisch de details van de gespecificeerde vaardigheid tijdens runtime.
run_sql_query(query): Voert specifieke SQL-statements uit.

Stap 3: Arrangeer Agent Logica (The Brain)

Gebruik LangGraph om een ReAct Agent te bouwen. De System Prompt speelt hier een cruciale rol; het begeleidt de Agent om strikt de Identify -> Load -> Query Standard Operating Procedure (SOP) te volgen.

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. Verificatie van de Uitvoeringseffecten

Door test_agent.py uit te voeren, hebben we query's getest in twee verschillende domeinen: Sales en Inventory. Hieronder staan de daadwerkelijke outputlogboeken van de console, die laten zien hoe de Agent dynamisch vaardigheden laadt op basis van de vraag:

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. Volledige broncode referentie

Hieronder staat de volledige broncode van het project, inclusief het database initialisatiescript en het Agent hoofdprogramma.

1. Database initialisatie (setup_db.py)

`import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT import os from dotenv import load_dotenv

load_dotenv()

Zorg ervoor dat de database verbindingsinformatie is geconfigureerd in .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") # Vervang door het daadwerkelijke wachtwoord DB_NAME = os.getenv("DB_NAME", "agent_platform")

def create_database(): try: # Verbind met de standaard 'postgres' database om een nieuwe database te maken 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()

    # Controleer of de database bestaat
    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()

    # Maak Sales Table
    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)
        )
        """
    )

    # Maak Inventory Table
    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)
        )
        """
    )

    # Voeg Mock Data in
    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() `### 2. Agent Hoofdprogramma (main.py)

`importosfromtypingimportAnnotated, Literal, TypedDict, Union, Dictfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.toolsimporttoolfromlangchain_core.messagesimportSystemMessage, HumanMessage, AIMessage, ToolMessagefromlangchain_community.utilitiesimportSQLDatabasefromlangchain_community.agent_toolkitsimportSQLDatabaseToolkitfromlanggraph.graphimportStateGraph, START, END, MessagesStatefromlanggraph.prebuiltimportToolNode, tools_conditionload_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":"Handig voor het analyseren van verkoopopbrengsten, trends en regionale prestaties.","content":"""Je bent een Sales Analytics Expert.Je hebt toegang tot de 'sales_data' tabel.Tabel Schema:- id: integer (primary key)- transaction_date: date- product_id: varchar(50)- amount: decimal(10, 2)- region: varchar(50)Veelvoorkomende queries:- Totale omzet: SUM(amount)- Omzet per regio: GROUP BY region- Verkooptrend: GROUP BY transaction_date""" },"inventory_management": {"description":"Handig voor het controleren van voorraadniveaus, productlocaties en magazijnbeheer.","content":"""Je bent een Inventory Management Expert.Je hebt toegang tot de 'inventory_item"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:""" Laad de gedetailleerde prompt en het schema voor een specifieke vaardigheid. Beschikbare vaardigheden:
- sales_analytics: Voor verkoop-, omzet- en transactieanalyse.
- inventory_management: Voor voorraad-, product- en magazijnquery's. """ skill = SKILLS.get(skill_name) if not skill: return f"Error: Vaardigheid '{skill_name}' niet gevonden. Beschikbare vaardigheden: {list(SKILLS.keys())}" return skill["content"]

@tool def run_sql_query(query: str) -> str: """ Voer een SQL-query uit op de database. Gebruik deze tool ALLEEN nadat u de juiste vaardigheid hebt geladen om het schema te begrijpen. """ try: return db.run(query) except Exception as e: return f"Fout bij het uitvoeren van SQL: {e}"

@tool def list_tables() -> str: """Maak een lijst van alle beschikbare tabellen in de database.""" return str(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 ---

class AgentState(MessagesState):

We kunnen indien nodig aangepaste status toevoegen, maar MessagesState is voldoende voor eenvoudige chat

pass

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()# --- Main Execution ---if__name__ =="main": system_prompt ="""Je bent een behulpzame SQL Assistent.Je hebt toegang tot gespecialiseerde vaardigheden die databaseschema's en domeinkennis bevatten.Om de vraag van een gebruiker te beantwoorden:1. Identificeer de relevante vaardigheid (sales_analytics of inventory_management).2. Gebruik de 'load_skill' tool om het schema en de instructies te verkrijgen.3. Gebaseerd op de geladen vaardigheid, schrijf en voer een SQL query uit met 'run_sql_query'.4. Beantwoord de vraag van de gebruiker op basis van de query resultaten.Raad geen tabelnamen. Laad altijd eerst de vaardigheid.""" print("SQL Assistent geïnitialiseerd. Typ 'quit' om af te sluiten.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Pre-warm connection checktry: print(f"Verbonden met database:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Database verbindingswaarschuwing:{e}")whileTrue:try: user_input = input("Gebruiker: ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Stream de uitvoering 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`

LangChain Skills-patroon in de praktijk: een SQL-assistent bouwen die kennis op aanvraag laadt