मागील लेखात, आम्ही Deep Agents CLI वापरून Deep Agent च्या कौशल्ये वापरण्याच्या पद्धतीचे अनुकरण कसे करावे याबद्दल चर्चा केली. आता, LangChain ने हे वैशिष्ट्य मूळ रूपात समर्थित केले आहे, ज्यामुळे विकास प्रक्रिया मोठ्या प्रमाणात सुलभ झाली आहे. हा लेख तुम्हाला या कार्याचा सखोल अनुभव घेण्यास आणि अधिक बुद्धिमान SQL सहाय्यक तयार करण्यास मार्गदर्शन करेल.

गुंतागुंतीचे AI Agent तयार करताना, विकासक अनेकदा एका अडचणीत सापडतात: सर्व संदर्भ (डेटाबेस टेबल स्ट्रक्चर, API डॉक्युमेंटेशन, व्यवसाय नियम) एकाच वेळी System Prompt मध्ये टाकायचे, ज्यामुळे संदर्भ विंडो (Context Window) ओव्हरफ्लो होईल आणि मॉडेलचे लक्ष विचलित होईल? की वारंवार फाइन-ट्यूनिंगचा (Fine-tuning) महाग पर्याय निवडायचा?

कौशल्ये मॉडेल (Skills Pattern) एक चांगला मध्यम मार्ग प्रदान करते. हे आवश्यक ज्ञान गतिशीलपणे लोड करून संदर्भाचा प्रभावी वापर सुनिश्चित करते. LangChain च्या या मॉडेलसाठी असलेल्या मूळ समर्थनाचा अर्थ असा आहे की, आपण "मागणीनुसार शिकण्याची" क्षमता असलेले Agent अधिक सहजपणे तयार करू शकतो.

हा लेख अधिकृत डॉक्युमेंटेशन Build a SQL assistant with on-demand skills च्या आधारे वाचकांना "मागणीनुसार ज्ञान लोड" करणारा SQL सहाय्यक तयार करण्यासाठी मार्गदर्शन करेल.

1. मुख्य संकल्पना: कौशल्ये मॉडेल का निवडायचे?

पारंपारिक SQL Agent च्या मर्यादा

पारंपारिक SQL Agent आर्किटेक्चरमध्ये, आम्हाला सामान्यतः System Prompt मध्ये संपूर्ण डेटाबेस स्कीमा (Database Schema) प्रदान करणे आवश्यक असते. व्यवसायाच्या वाढीसह, जेव्हा टेबलची संख्या शेकडोंपर्यंत वाढते, तेव्हा या पद्धतीमुळे लक्षणीय समस्या येतात:

• जास्त टोकन वापर: प्रत्येक संवादात मोठ्या प्रमाणात अनावश्यक टेबल स्ट्रक्चर सोबत ठेवल्याने संसाधनांचा अपव्यय होतो.

• भ्रम होण्याचा धोका वाढतो: जास्त प्रमाणात अनावश्यक माहिती मॉडेलच्या अनुमान अचूकतेवर परिणाम करते.

• देखभाल करणे कठीण: सर्व व्यवसाय लाइनचे ज्ञान एकमेकांशी जोडलेले असल्याने स्वतंत्रपणे सुधारणे कठीण होते.

कौशल्ये मॉडेल: प्रगतीशील प्रकटीकरणावर आधारित उपाय

कौशल्ये मॉडेल प्रगतीशील प्रकटीकरण (Progressive Disclosure) या तत्त्वावर आधारित आहे, जे ज्ञान संपादनाची प्रक्रिया स्तरांमध्ये विभाजित करते:

• Agent ची प्रारंभिक स्थिती: फक्त कोणती "कौशल्ये" (Skills) आहेत आणि त्यांचे संक्षिप्त वर्णन (Description) काय आहे हे माहीत असते, त्यामुळे ते हलके राहते.

• रनटाइम लोडिंग: जेव्हा एखाद्या विशिष्ट समस्येचा सामना करावा लागतो (जसे की "इन्व्हेंटरी क्वेरी करा"), तेव्हा Agent सक्रियपणे साधन (load_skill) वापरून त्या कौशल्याचा तपशीलवार संदर्भ (स्कीमा + प्रॉम्प्ट) लोड करते.

• कार्य अंमलबजावणी: लोड केलेल्या अचूक संदर्भावर आधारित, विशिष्ट कार्ये (जसे की SQL लिहिणे आणि कार्यान्वित करणे) पार पाडली जातात.

ही पद्धत अमर्याद विस्तारास आणि टीम डिकूपलिंगला प्रभावीपणे समर्थन देते, ज्यामुळे Agent ला अधिकाधिक जटिल व्यवसाय परिस्थितीशी जुळवून घेता येते.

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

या प्रात्यक्षिक प्रकल्पात, आम्ही दोन मुख्य कौशल्ये असलेला SQL सहाय्यक तयार करू, जो या मॉडेलचा प्रत्यक्ष वापर दर्शवेल:

• 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. मुख्य अंमलबजावणी चरणांचे तपशीलवार स्पष्टीकरण### पायरी १: डोमेन कौशल्ये परिभाषित करणे (ज्ञान)

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

SKILLS = {"sales_analytics": {"description":"Useful for analyzing sales revenue, trends...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"Useful for checking stock levels...","content":"""... Table Schema: inventory_items ..."" }}

पायरी २: मुख्य साधने लागू करणे (क्षमता)

एजेंटला कार्य पूर्ण करण्यासाठी दोन महत्त्वाची साधने लागतात:

• load_skill(skill_name): रनटाइममध्ये निर्दिष्ट कौशल्ये तपशीलवार लोड करते.

• run_sql_query(query): विशिष्ट SQL वाक्ये कार्यान्वित करते.

पायरी ३: एजेंट लॉजिकची मांडणी (मेंदू)

LangGraph वापरून ReAct Agent तयार करा. येथे System Prompt महत्त्वाची भूमिका बजावते, जे एजेंटला 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."""

५. रनिंग इफेक्ट पडताळणी

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

खाली प्रकल्पाचा संपूर्ण स्त्रोत कोड आहे, ज्यात डेटाबेस इनिशियलायझेशन स्क्रिप्ट आणि एजंट मुख्य प्रोग्राम समाविष्ट आहे.

१. डेटाबेस इनिशियलायझेशन (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()

    # सेल्स टेबल तयार करा
    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. 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()

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

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: integer (प्रायमरी की)
• transaction_date: date
• product_id: varchar(50)
• amount: decimal(10, 2)
• region: varchar(50) सामान्य क्वेरी:-
• एकूण महसूल: SUM(amount)
• प्रदेशानुसार महसूल: GROUP BY region
• विक्री ट्रेंड: GROUP BY transaction_date""" }, "inventory_management": { "description": "स्टॉक लेव्हल, उत्पादन स्थाने आणि वेअरहाउस व्यवस्थापन तपासण्यासाठी उपयुक्त.", "content": """ तुम्ही इन्व्हेंटरी मॅनेजमेंट एक्सपर्ट आहात. तुमच्याकडे 'inventory_items' 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("tools", ToolNode(tools))workflow.add_edge(START,"agent")workflow.add_conditional_edges("agent", tools_condition)workflow.add_edge("tools","agent")app = workflow.compile()# --- मुख्य अंमलबजावणी ---if__name__ =="main": system_prompt ="""तुम्ही एक उपयुक्त SQL सहाय्यक आहात. तुमच्याकडे डेटाबेस स्कीमा आणि डोमेन ज्ञान असलेली विशेष कौशल्ये आहेत. वापरकर्त्याच्या प्रश्नाचे उत्तर देण्यासाठी: 1. संबंधित कौशल्य ओळखा (sales_analytics किंवा inventory_management). 2. स्कीमा आणि सूचना मिळवण्यासाठी 'load_skill' टूल वापरा. 3. लोड केलेल्या कौशल्यावर आधारित, 'run_sql_query' वापरून SQL क्वेरी लिहा आणि कार्यान्वित करा. 4. क्वेरी निकालांवर आधारित वापरकर्त्याच्या प्रश्नाचे उत्तर द्या. टेबल नावांचा अंदाज लावू नका. नेहमी प्रथम कौशल्य लोड करा.""" print("SQL सहाय्यक सुरू झाला. बाहेर पडण्यासाठी 'quit' टाइप करा.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# कनेक्शन तपासणी पूर्व-गरम करण्याचा प्रयत्न करा: try: print(f"डेटाबेसशी कनेक्ट केलेले:{DB_URI.split('@')[-1]}") exceptExceptionase: print(f"डेटाबेस कनेक्शन इशारा:{e}") whileTrue: try: user_input = input("वापरकर्ता: ") ifuser_input.lower()in["quit","exit"]: break messages.append(HumanMessage(content=user_input))# अंमलबजावणी प्रवाहित करा print("एजंट: ", end="", flush=True) final_response =None foreventinapp.stream({"messages": messages}, stream_mode="values"):# 'values' मोडमध्ये, आम्हाला पूर्ण स्थिती मिळते. आम्हाला फक्त शेवटचा संदेश नवीन असल्यास पहायचा आहे. last_message = event["messages"][-1]# नवीनतम स्थितीसह आमचा संदेश इतिहास अद्यतनित करा pass# प्रवाह पूर्ण झाल्यावर, अंतिम स्थितीत अंतिम उत्तर असते 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`