پچھلے مضمون میں، ہم نے دریافت کیا کہ کس طرح ڈیپ ایجنٹس CLI کے ذریعے ڈیپ ایجنٹ کے Skills استعمال کرنے کے پیٹرن کی تقلید کی جائے۔ آج کل، LangChain مقامی طور پر اس خصوصیت کو سپورٹ کرتا ہے، جو ترقی کے عمل کو بہت آسان بناتا ہے۔ یہ مضمون آپ کو اس فنکشن کا گہرائی سے تجربہ کرنے اور ایک زیادہ ذہین SQL معاون بنانے کے لیے رہنمائی کرے گا۔

پیچیدہ AI ایجنٹ بناتے وقت، ڈویلپرز اکثر ایک مشکل صورتحال میں پھنس جاتے ہیں: کیا تمام سیاق و سباق (ڈیٹا بیس ٹیبل ڈھانچہ، API دستاویزات، کاروباری قواعد) کو ایک ہی بار میں سسٹم پرامپٹ میں داخل کیا جائے، جس سے سیاق و سباق کی ونڈو (Context Window) اوور فلو ہو جائے اور ماڈل کی توجہ منتشر ہو جائے؟ یا مہنگی بار بار باریک ٹیوننگ (Fine-tuning) کا انتخاب کریں؟

Skills پیٹرن ایک خوبصورت درمیانی راستہ فراہم کرتا ہے۔ یہ مطلوبہ علم کو متحرک طور پر لوڈ کرکے سیاق و سباق کے موثر استعمال کو حاصل کرتا ہے۔ LangChain کی طرف سے اس پیٹرن کے لیے مقامی سپورٹ کا مطلب ہے کہ ہم "مطالبے پر سیکھنے" کی صلاحیتوں کے ساتھ ایجنٹ کو زیادہ آسانی سے بنا سکتے ہیں۔

یہ مضمون سرکاری دستاویز Build a SQL assistant with on-demand skills کے ساتھ مل کر، قاری کو شروع سے ہی ایک SQL اسسٹنٹ بنانے کی رہنمائی کرے گا جو "مطالبے پر علم کو لوڈ کرنے" کی حمایت کرتا ہے۔

1. بنیادی تصور: Skills پیٹرن کا انتخاب کیوں؟

روایتی SQL ایجنٹ کی حدود

روایتی SQL ایجنٹ فن تعمیر میں، ہمیں عام طور پر سسٹم پرامپٹ میں مکمل ڈیٹا بیس اسکیما فراہم کرنے کی ضرورت ہوتی ہے۔ کاروبار کی ترقی کے ساتھ، جب ٹیبل کی تعداد سینکڑوں تک پھیل جاتی ہے، تو یہ طریقہ نمایاں مسائل لاتا ہے:

ٹوکن کی بہت زیادہ کھپت: ہر گفتگو بڑی تعداد میں غیر متعلقہ ٹیبل ڈھانچے کو لے کر جاتی ہے، جس سے وسائل کا ضیاع ہوتا ہے۔
فریب کے خطرے میں اضافہ: بہت زیادہ غیر متعلقہ مداخلت کرنے والی معلومات ماڈل کی استدلال کی درستگی کو کم کر دے گی۔
بحالی میں دشواری: تمام کاروباری لائنوں کا علم مضبوطی سے جڑا ہوا ہے، جس سے آزادانہ طور پر تکرار کرنا مشکل ہو جاتا ہے۔

Skills پیٹرن: بتدریج انکشاف پر مبنی حل

Skills پیٹرن بتدریج انکشاف (Progressive Disclosure) کے اصول پر مبنی ہے، جو علم کے حصول کے عمل کو پرتوں میں تقسیم کرتا ہے:

ایجنٹ کی ابتدائی حالت: صرف یہ جاننا کہ کون سی "مہارتیں" (Skills) ہیں اور ان کی مختصر تفصیل (Description)، ہلکا پھلکا رہنا۔
رن ٹائم لوڈنگ: جب کسی خاص مسئلے (جیسے "انوینٹری کوئری") کا سامنا کرنا پڑتا ہے، تو ایجنٹ فعال طور پر ٹول (load_skill) کو کال کرتا ہے تاکہ اس مہارت کے تفصیلی سیاق و سباق (اسکیما + پرامپٹ) کو لوڈ کیا جا سکے۔
کام انجام دینا: لوڈ کیے گئے درست سیاق و سباق کی بنیاد پر، مخصوص کام انجام دیں (جیسے SQL لکھنا اور چلانا)۔

یہ پیٹرن مؤثر طریقے سے لامحدود توسیع اور ٹیم ڈی کپلنگ کی حمایت کرتا ہے، جس سے ایجنٹ کو تیزی سے پیچیدہ کاروباری منظرناموں کے مطابق ڈھالنے کی اجازت ملتی ہے۔

2. سسٹم آرکیٹیکچر ڈیزائن

یہ عملی پروجیکٹ ایک SQL اسسٹنٹ بنائے گا جس میں دو بنیادی Skills شامل ہیں، تاکہ اس پیٹرن کے عملی استعمال کو ظاہر کیا جا سکے:

Sales Analytics (سیلز تجزیات):sales_data** ٹیبل کے لیے ذمہ دار، آمدنی کے اعدادوشمار، آرڈر کے رجحان کے تجزیے وغیرہ کو سنبھالتا ہے۔**
Inventory Management (انوینٹری مینجمنٹ):inventory_items** ٹیبل کے لیے ذمہ دار، انوینٹری کی سطح کی نگرانی، مقام کی کوئری وغیرہ کو سنبھالتا ہے۔**

3. ترقیاتی ماحول کی تنصیب

یہ پروجیکٹ Pythonuv کو موثر انحصار کے انتظام کے لیے استعمال کرتا ہے۔

بنیادی انحصار کی تنصیب

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":"فروخت کی آمدنی، رجحانات کا تجزیہ کرنے کے لیے مفید...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"اسٹاک کی سطح کو چیک کرنے کے لیے مفید...","content":"""... Table Schema: inventory_items ..."" }}

قدم دوم: بنیادی ٹولز کا نفاذ (صلاحیتیں)

ایجنٹ کو کام مکمل کرنے کے لیے دو اہم ٹولز پر انحصار کرنا پڑتا ہے:

load_skill(skill_name): رن ٹائم پر مخصوص مہارت کی تفصیلات کو متحرک طور پر لوڈ کریں۔
run_sql_query(query): مخصوص SQL اسٹیٹمنٹ پر عمل کریں۔

قدم سوم: ایجنٹ منطق کو ترتیب دینا (دماغ)

LangGraph کا استعمال کرتے ہوئے ReAct ایجنٹ بنائیں۔ سسٹم پرامپٹ یہاں ایک اہم کردار ادا کرتا ہے، یہ ایجنٹ کو 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."""

5. چلانے کے اثر کی توثیق

test_agent.py چلا کر، ہم نے سیلز اور انوینٹری کے دو مختلف ڈومینز میں سوالات کی جانچ کی۔ ذیل میں کنسول کے اصل آؤٹ پٹ لاگز ہیں، جو یہ ظاہر کرتے ہیں کہ ایجنٹ کس طرح سوال کے مطابق مہارتوں کو متحرک طور پر لوڈ کرتا ہے:

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. Agent \u0645\u06cc\u0646 \u067e\u0631\u0648\u06af\u0631\u0627\u0645 (main.py)**\n\nimportosfromtypingimportAnnotated, 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(\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"] @tool def run_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) except Exception as e: returnf"Error executing SQL:{e}" @tool def list_tables()-> str: """List all available tables in the 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 can add custom state if needed, but MessagesState is sufficient for simple 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 ="""آپ ایک مددگار 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 =Noneforeventinapp.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]

if isinstance(last_msg, AIMessage):

print(last_msg.content)

messages = final_state["messages"]