পূর্ববর্তী নিবন্ধে, আমরা আলোচনা করেছি কিভাবে Deep Agents CLI ব্যবহার করে Skills ব্যবহারের Deep Agent মডেল অনুকরণ করা যায়। বর্তমানে, LangChain এই বৈশিষ্ট্যটিকে স্থানীয়ভাবে সমর্থন করে, যা উন্নয়নের প্রক্রিয়াকে অনেক সহজ করে। এই নিবন্ধটি আপনাকে এই বৈশিষ্ট্যটি গভীরভাবে অনুভব করতে এবং একটি আরও বুদ্ধিমান SQL সহকারী তৈরি করতে পরিচালিত করবে।

জটিল AI Agent তৈরি করার সময়, বিকাশকারীরা প্রায়শই একটি দ্বিধাদ্বন্দ্বের মধ্যে পড়ে: সমস্ত প্রাসঙ্গিক তথ্য (ডাটাবেস টেবিলের গঠন, API ডকুমেন্টেশন, ব্যবসার নিয়ম) কি একবারেই System Prompt-এ প্রবেশ করানো হবে, যার ফলে Context Window উপচে পড়বে এবং মডেলের মনোযোগ বিক্ষিপ্ত হবে? নাকি ব্যয়বহুল ঘন ঘন ফাইন-টিউনিং বেছে নেওয়া হবে?

Skills মডেল (Skills Pattern) একটি মার্জিত মধ্যবর্তী পথ সরবরাহ করে। এটি প্রয়োজনীয় জ্ঞান গতিশীলভাবে লোড করার মাধ্যমে প্রাসঙ্গিক তথ্যের দক্ষ ব্যবহার নিশ্চিত করে। LangChain-এর এই মডেলের স্থানীয় সমর্থন মানে আমরা আরও সহজে "চাহিদা অনুযায়ী শেখার" ক্ষমতা সম্পন্ন Agent তৈরি করতে পারি।

এই নিবন্ধটি অফিসিয়াল ডকুমেন্ট Build a SQL assistant with on-demand skills-এর সাথে মিলিতভাবে, পাঠকদের একটি SQL Assistant তৈরি করতে গাইড করবে যা "চাহিদা অনুযায়ী জ্ঞান লোড" সমর্থন করে।

1. মূল ধারণা: কেন Skills মডেল নির্বাচন করবেন?

ঐতিহ্যবাহী SQL Agent-এর সীমাবদ্ধতা

ঐতিহ্যবাহী SQL Agent কাঠামোতে, আমাদের সাধারণত System Prompt-এ সম্পূর্ণ Database Schema সরবরাহ করতে হয়। ব্যবসার বিকাশের সাথে সাথে, যখন টেবিলের সংখ্যা কয়েকশতে প্রসারিত হয়, তখন এই পদ্ধতিটি উল্লেখযোগ্য সমস্যা নিয়ে আসে:

Token-এর বিশাল খরচ: প্রতিটি কথোপকথনে প্রচুর অপ্রাসঙ্গিক টেবিলের গঠন বহন করা, যা সম্পদের অপচয় ঘটায়।
অলীক কল্পনার ঝুঁকি বৃদ্ধি: অতিরিক্ত অপ্রাসঙ্গিক তথ্য মডেলের যুক্তির নির্ভুলতা হ্রাস করে।
রক্ষণাবেক্ষণ কঠিন: সমস্ত ব্যবসার জ্ঞানের লাইন ঘনিষ্ঠভাবে মিলিত, যা স্বাধীনভাবে পুনরাবৃত্তি করা কঠিন করে তোলে।

Skills মডেল: প্রগতিশীল প্রকাশের উপর ভিত্তি করে সমাধান

Skills মডেল প্রগতিশীল প্রকাশ (Progressive Disclosure) নীতির উপর ভিত্তি করে, জ্ঞান অর্জনের প্রক্রিয়াটিকে স্তরযুক্ত করে:

Agent-এর প্রাথমিক অবস্থা: শুধুমাত্র কী কী "দক্ষতা" (Skills) আছে এবং তাদের সংক্ষিপ্ত বিবরণ (Description) জানে, যা হালকা থাকে।
রানটাইমে লোড: যখন কোনো নির্দিষ্ট সমস্যার সম্মুখীন হয় (যেমন "ইনভেন্টরি অনুসন্ধান") তখন Agent সক্রিয়ভাবে সরঞ্জাম (load_skill) ব্যবহার করে সেই দক্ষতার বিস্তারিত প্রাসঙ্গিক তথ্য (Schema + Prompt) লোড করে।
কাজ সম্পাদন: লোড করা সঠিক প্রাসঙ্গিক তথ্যের উপর ভিত্তি করে, নির্দিষ্ট কাজগুলি সম্পাদন করে (যেমন SQL লেখা এবং চালানো)।

এই মডেলটি কার্যকরভাবে অসীম প্রসারণ এবং দলের বিচ্ছিন্নতাকে সমর্থন করে, যা Agent-কে ক্রমবর্ধমান জটিল ব্যবসার পরিস্থিতিতে মানিয়ে নিতে সক্ষম করে।

2. সিস্টেম আর্কিটেকচার ডিজাইন

এই বাস্তবায়ন প্রকল্পটি দুটি মূল Skills সহ একটি SQL Assistant তৈরি করবে, যা এই মডেলের বাস্তব প্রয়োগ প্রদর্শন করবে:

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":"বিক্রয় রাজস্ব, প্রবণতা বিশ্লেষণের জন্য উপযোগী...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"স্টক লেভেল পরীক্ষা করার জন্য উপযোগী...","content":"""... Table Schema: inventory_items ..."" }}

ধাপ ২: মূল সরঞ্জামগুলি বাস্তবায়ন করা (ক্ষমতা)

এজেন্ট দুটি গুরুত্বপূর্ণ সরঞ্জামের উপর নির্ভর করে কাজ সম্পন্ন করার জন্য:

load_skill(skill_name): রানটাইমে নির্দিষ্ট দক্ষতার বিবরণ গতিশীলভাবে লোড করে।
run_sql_query(query): নির্দিষ্ট SQL স্টেটমেন্ট চালায়।

ধাপ ৩: এজেন্ট লজিক তৈরি করা (মস্তিষ্ক)

LangGraph ব্যবহার করে ReAct এজেন্ট তৈরি করুন। 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 চালানোর মাধ্যমে, আমরা বিক্রয় এবং ইনভেন্টরি এই দুটি ভিন্ন ডোমেইনের প্রশ্নের পরীক্ষা করেছি। নিচে কনসোলের প্রকৃত আউটপুট লগ দেওয়া হল, যা দেখায় কিভাবে এজেন্ট প্রশ্নের ভিত্তিতে গতিশীলভাবে দক্ষতা লোড করে:

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() `### ২. এজেন্ট মূল প্রোগ্রাম (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 (primary key) - 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(\কে({"messages": messages}) last_msg = final_state["messages"][-1]ifisinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]# ইতিহাস আপডেট করুন print("-"*50)exceptExceptionase: print(f"\nError:{e}")break`