पिछली लेख में, हमने पता लगाया कि डीप एजेंट्स सीएलआई के माध्यम से डीप एजेंट द्वारा स्किल्स का उपयोग करने के पैटर्न का अनुकरण कैसे करें। अब, लैंगचेन ने मूल रूप से इस सुविधा का समर्थन किया है, जिससे विकास प्रक्रिया बहुत सरल हो गई है। यह लेख आपको इस सुविधा का गहराई से अनुभव करने और एक अधिक बुद्धिमान SQL सहायक बनाने के लिए मार्गदर्शन करेगा।

जटिल AI एजेंटों का निर्माण करते समय, डेवलपर्स अक्सर एक दुविधा में फंस जाते हैं: क्या सभी संदर्भ (डेटाबेस तालिका संरचना, API दस्तावेज़, व्यावसायिक नियम) को एक बार में सिस्टम प्रॉम्प्ट में इंजेक्ट किया जाए, जिससे संदर्भ विंडो (कॉन्टेक्स्ट विंडो) अतिप्रवाहित हो जाए और मॉडल का ध्यान भटक जाए? या उच्च लागत वाली लगातार फाइन-ट्यूनिंग (फाइन-ट्यूनिंग) का विकल्प चुनें?

स्किल्स पैटर्न (स्किल्स पैटर्न) एक सुरुचिपूर्ण मध्य मार्ग प्रदान करता है। यह आवश्यक ज्ञान को गतिशील रूप से लोड करके संदर्भ के कुशल उपयोग को प्राप्त करता है। लैंगचेन द्वारा इस पैटर्न के लिए मूल समर्थन का मतलब है कि हम "मांग पर सीखने" क्षमताओं वाले एजेंटों का निर्माण अधिक आसानी से कर सकते हैं।

यह लेख आधिकारिक दस्तावेज़ Build a SQL assistant with on-demand skills के साथ मिलकर पाठकों को शुरू से ही "मांग पर ज्ञान लोड करने" का समर्थन करने वाले SQL सहायक का निर्माण करने के लिए मार्गदर्शन करेगा।

1. मुख्य अवधारणा: स्किल्स पैटर्न क्यों चुनें?

पारंपरिक SQL एजेंट की सीमाएं

पारंपरिक SQL एजेंट आर्किटेक्चर में, हमें आमतौर पर सिस्टम प्रॉम्प्ट में पूरा डेटाबेस स्कीमा प्रदान करने की आवश्यकता होती है। जैसे-जैसे व्यवसाय बढ़ता है, जब तालिकाओं की संख्या सैकड़ों तक बढ़ जाती है, तो यह विधि महत्वपूर्ण समस्याएं लाएगी:

• टोकन की भारी खपत: प्रत्येक बातचीत बड़ी मात्रा में अप्रासंगिक तालिका संरचनाओं को ले जाती है, जिससे संसाधनों की बर्बादी होती है।

• भ्रम का बढ़ता जोखिम: बहुत अधिक अप्रासंगिक हस्तक्षेप जानकारी मॉडल की तर्क सटीकता को कम कर देगी।

• रखरखाव में कठिनाई: सभी व्यावसायिक लाइनों का ज्ञान कसकर युग्मित है, जिसे स्वतंत्र रूप से दोहराना मुश्किल है।

स्किल्स पैटर्न: क्रमिक प्रकटीकरण पर आधारित समाधान

स्किल्स पैटर्न क्रमिक प्रकटीकरण (प्रोग्रेसिव डिस्क्लोजर) सिद्धांत पर आधारित है, जो ज्ञान अधिग्रहण प्रक्रिया को स्तरित रूप से संसाधित करता है:

• एजेंट की प्रारंभिक स्थिति: केवल यह जानना कि कौन से "कौशल" (स्किल्स) हैं और उनका संक्षिप्त विवरण (डिस्क्रिप्शन), हल्का रहना।

• रनटाइम लोडिंग: जब किसी विशिष्ट समस्या (जैसे "इन्वेंट्री क्वेरी") का सामना करना पड़ता है, तो एजेंट सक्रिय रूप से टूल (load_skill) को उस कौशल के विस्तृत संदर्भ (स्कीमा + प्रॉम्प्ट) को लोड करने के लिए कॉल करता है।

• कार्य निष्पादित करें: लोड किए गए सटीक संदर्भ के आधार पर, विशिष्ट कार्यों को निष्पादित करें (जैसे SQL लिखना और निष्पादित करना)।

यह पैटर्न प्रभावी रूप से अनंत विस्तार और टीम डीकपलिंग का समर्थन करता है, जिससे एजेंट तेजी से जटिल व्यावसायिक परिदृश्यों के अनुकूल हो सकता है।

2. सिस्टम आर्किटेक्चर डिजाइन

यह व्यावहारिक परियोजना दो मुख्य स्किल्स वाले SQL सहायक का निर्माण करेगी, ताकि इस पैटर्न के वास्तविक अनुप्रयोग का प्रदर्शन किया जा सके:

• Sales Analytics (बिक्री विश्लेषण):sales_data** तालिका के लिए जिम्मेदार, राजस्व सांख्यिकी, ऑर्डर प्रवृत्ति विश्लेषण आदि को संभालना।**

• Inventory Management (इन्वेंट्री प्रबंधन):inventory_items** तालिका के लिए जिम्मेदार, इन्वेंट्री स्तर की निगरानी, स्थान क्वेरी आदि को संभालना।**

3. विकास पर्यावरण सेटअप

यह परियोजना कुशल निर्भरता प्रबंधन के लिए पायथनuv का उपयोग करती है।

मुख्य निर्भरता स्थापना

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

PostgreSQL पर्यावरण कॉन्फ़िगरेशन

स्थानीय रूप से एक Postgres उदाहरण शुरू करें औरagent_platform डेटाबेस बनाएं। हमने तालिका संरचना और परीक्षण डेटा को स्वचालित रूप से आरंभ करने के लिएsetup_db.py स्क्रिप्ट प्रदान की है (विवरण के लिए लेख के अंत में स्रोत कोड देखें)।

4. मुख्य कार्यान्वयन चरणों का विस्तृत विवरण### चरण 1: डोमेन कौशल को परिभाषित करना (ज्ञान)

हम कौशल को एक शब्दकोश संरचना के रूप में परिभाषित करेंगे, जो फ़ाइल सिस्टम या डेटाबेस से लोडिंग की प्रक्रिया का अनुकरण करता है। कृपया description (एजेंट निर्णय चयन के लिए) और content (वास्तविक लोडिंग विस्तृत संदर्भ) के बीच अंतर करें।

SKILLS = {"sales_analytics": {"description":"बिक्री राजस्व, रुझानों का विश्लेषण करने के लिए उपयोगी...","content":"""... टेबल स्कीमा: sales_data ..."" },"inventory_management": {"description":"स्टॉक स्तर की जाँच के लिए उपयोगी...","content":"""... टेबल स्कीमा: inventory_items ..."" }}

चरण 2: मुख्य उपकरण लागू करना (क्षमताएं)

एजेंट को कार्यों को पूरा करने के लिए दो महत्वपूर्ण उपकरणों पर निर्भर रहना पड़ता है:

• load_skill(skill_name): रनटाइम पर निर्दिष्ट कौशल का विवरण गतिशील रूप से लोड करें।

• run_sql_query(query): विशिष्ट SQL स्टेटमेंट निष्पादित करें।

चरण 3: एजेंट लॉजिक को व्यवस्थित करना (मस्तिष्क)

LangGraph का उपयोग करके ReAct एजेंट का निर्माण करें। सिस्टम प्रॉम्प्ट यहां महत्वपूर्ण भूमिका निभाता है, यह एजेंट को Identify -> Load -> Query के मानक संचालन प्रक्रिया (SOP) का सख्ती से पालन करने का निर्देश देता है।

system_prompt ="""1. प्रासंगिक कौशल की पहचान करें।2. स्कीमा प्राप्त करने के लिए 'load_skill' का उपयोग करें।3. 'run_sql_query' का उपयोग करके SQL लिखें और निष्पादित करें....टेबल नामों का अनुमान न लगाएं। हमेशा पहले कौशल लोड करें।"""

5. रनिंग इफेक्ट वेरिफिकेशन

test_agent.py चलाकर, हमने Sales और Inventory के दो अलग-अलग क्षेत्रों के प्रश्नों का परीक्षण किया। नीचे कंसोल का वास्तविक आउटपुट लॉग है, जो दिखाता है कि एजेंट प्रश्नों के आधार पर गतिशील रूप से कौशल कैसे लोड करता है:

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. पूर्ण स्रोत कोड संदर्भ

यहां परियोजना का पूर्ण स्रोत कोड है, जिसमें डेटाबेस इनिशियलाइज़ेशन स्क्रिप्ट और एजेंट मुख्य प्रोग्राम शामिल हैं।

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: # नया 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()

    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()

    # सेल्स टेबल बनाएं
    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)
        )
        """
    )

    # इन्वेंटरी टेबल बनाएं
    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() `### 2. एजेंट मुख्य प्रोग्राम (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()

--- कॉन्फ़िगरेशन ---

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')}"

--- डेटाबेस सेटअप ---

db = SQLDatabase.from_uri(DB_URI)

--- कौशल परिभाषा ---

SKILLS: Dict[str, Dict[str, str]] = { "sales_analytics": { "description": "बिक्री राजस्व, रुझानों और क्षेत्रीय प्रदर्शन का विश्लेषण करने के लिए उपयोगी।", "content": """ आप एक बिक्री विश्लेषण विशेषज्ञ हैं। आपके पास 'sales_data' टेबल तक पहुंच है। टेबल स्कीमा:

• id: पूर्णांक (प्राथमिक कुंजी)
• transaction_date: तिथि
• product_id: varchar(50)
• amount: दशमलव(10, 2)
• region: varchar(50) सामान्य प्रश्न:
• कुल राजस्व: SUM(amount)
• क्षेत्र द्वारा राजस्व: GROUP BY region
• बिक्री प्रवृत्ति: GROUP BY transaction_date""" }, "inventory_management": { "description": "स्टॉक स्तर, उत्पाद स्थान और गोदाम प्रबंधन की जांच के लिए उपयोगी।", "content": """ आप एक इन्वेंटरी प्रबंधन विशेषज्ञ हैं। आपके पास '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:""" Load the detailed prompt and schema for a specific skill. Available skills: - sales_analytics: For sales, revenue, and transaction analysis. - inventory_management: For stock, products, and warehouse queries. """ skill = SKILLS.get(skill_name)ifnotskill:returnf"Error: Skill '{skill_name}' not found. Available skills:{list(SKILLS.keys())}"returnskill["content"]@tooldefrun_sql_query(query: str)-> str:""" Execute a SQL query against the database. Only use this tool AFTER loading the appropriate skill to understand the schema. """try:returndb.run(query)exceptExceptionase:returnf"Error executing SQL:{e}"@tooldeflist_tables()-> str:"""List all available tables in the database."""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):# We can add custom state if needed, but MessagesState is sufficient for simple chatpassdefagent_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(\के({"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`