ယခင်ဆောင်းပါးတွင်၊ Deep Agents CLI ကို အသုံးပြု၍ Skills မော်ဒယ်လ်ကို အသုံးပြုသော Deep Agent ကို မည်သို့ပုံစံတူဖန်တီးနိုင်ကြောင်းကို ကျွန်ုပ်တို့လေ့လာခဲ့သည်။ ယခုအခါ LangChain သည် ဤအင်္ဂါရပ်ကို မူလကပင် ပံ့ပိုးပေးထားပြီး၊ ဖွံ့ဖြိုးတိုးတက်မှုလုပ်ငန်းစဉ်ကို များစွာရိုးရှင်းစေသည်။ ဤဆောင်းပါးတွင်၊ ဤလုပ်ဆောင်ချက်ကို အသေးစိတ်လေ့လာပြီး ပိုမိုထက်မြက်သော SQL လက်ထောက်ကို တည်ဆောက်ရန် သင့်အား လမ်းညွှန်ပေးပါမည်။

ရှုပ်ထွေးသော AI Agent ကို တည်ဆောက်သောအခါ၊ developer များသည် အခက်အခဲနှစ်ခုကြားတွင် ရုန်းကန်နေရသည်- အကြောင်းအရာအားလုံး (ဒေတာဘေ့စ်ဇယားဖွဲ့စည်းပုံ၊ API စာရွက်စာတမ်းများ၊ လုပ်ငန်းစည်းမျဉ်းများ) ကို System Prompt ထဲသို့ တစ်ကြိမ်တည်းထည့်သွင်းခြင်းသည် Context Window ပြည့်လျှံသွားပြီး မော်ဒယ်လ်၏အာရုံစူးစိုက်မှုကို လျော့နည်းစေသလား။ သို့မဟုတ် ကုန်ကျစရိတ်ကြီးမြင့်သော မကြာခဏ Fine-tuning ကို ရွေးချယ်မလား။

Skills မော်ဒယ်လ် (Skills Pattern) သည် လှပသော အလယ်အလတ်လမ်းကြောင်းကို ပေးပါသည်။ လိုအပ်သော ဗဟုသုတကို တက်ကြွစွာထည့်သွင်းခြင်းဖြင့် အကြောင်းအရာကို ထိရောက်စွာအသုံးပြုနိုင်စေပါသည်။ LangChain ၏ ဤမော်ဒယ်လ်အတွက် မူလပံ့ပိုးမှုသည် "လိုအပ်သလို သင်ယူနိုင်စွမ်း" ရှိသော Agent ကို ပိုမိုလွယ်ကူစွာ တည်ဆောက်နိုင်သည်ဟု ဆိုလိုပါသည်။

ဤဆောင်းပါးသည် တရားဝင်စာရွက်စာတမ်း Build a SQL assistant with on-demand skills နှင့် ပေါင်းစပ်၍ စာဖတ်သူများအား "လိုအပ်သလို ဗဟုသုတထည့်သွင်းခြင်း" ကို ပံ့ပိုးပေးသော SQL Assistant ကို အစမှစ၍ တည်ဆောက်ရန် လမ်းညွှန်ပေးပါမည်။

၁။ အဓိကအယူအဆ- Skills မော်ဒယ်လ်ကို အဘယ်ကြောင့်ရွေးချယ်သင့်သနည်း။

ရိုးရာ SQL Agent ၏ အားနည်းချက်များ

ရိုးရာ SQL Agent ဗိသုကာတွင်၊ ကျွန်ုပ်တို့သည် System Prompt တွင် Database Schema အပြည့်အစုံကို ပေးရလေ့ရှိသည်။ လုပ်ငန်းတိုးတက်လာသည်နှင့်အမျှ ဇယားအရေအတွက်သည် ရာနှင့်ချီ၍ တိုးလာသောအခါ၊ ဤနည်းလမ်းသည် သိသာထင်ရှားသော ပြဿနာများကို ဖြစ်စေသည်-

• Token အမြောက်အမြားသုံးစွဲခြင်း- ဆွေးနွေးမှုတိုင်းသည် သက်ဆိုင်မှုမရှိသော ဇယားဖွဲ့စည်းပုံအမြောက်အမြားကို သယ်ဆောင်လာပြီး အရင်းအမြစ်များကို ဖြုန်းတီးစေသည်။

• မှားယွင်းမှုအန္တရာယ်တိုးလာခြင်း- သက်ဆိုင်မှုမရှိသော အနှောင့်အယှက်အချက်အလက်များလွန်းခြင်းသည် မော်ဒယ်လ်၏ အကြောင်းပြချက်တိကျမှုကို လျော့နည်းစေသည်။

• ထိန်းသိမ်းရန်ခက်ခဲခြင်း- လုပ်ငန်းလိုင်းအားလုံး၏ ဗဟုသုတသည် အချင်းချင်းအလွန်အမင်းချိတ်ဆက်နေပြီး သီးခြားစီ ပြန်လည်ပြင်ဆင်ရန်ခက်ခဲသည်။

Skills မော်ဒယ်လ်- တဖြည်းဖြည်းထုတ်ဖော်ပြသမှုအပေါ်အခြေခံသော ဖြေရှင်းချက်

Skills မော်ဒယ်လ်သည် တဖြည်းဖြည်းထုတ်ဖော်ပြသမှု (Progressive Disclosure) အခြေခံမူအပေါ် အခြေခံပြီး ဗဟုသုတရယူခြင်းလုပ်ငန်းစဉ်ကို အလွှာလိုက်စီမံသည်-

• Agent ၏ မူလအခြေအနေ- ၎င်းတွင် မည်သည့် "စွမ်းရည်များ" (Skills) ရှိသည်နှင့် ၎င်းတို့၏ အကျဉ်းချုပ်ဖော်ပြချက် (Description) ကိုသာ သိရှိထားပြီး ပေါ့ပါးမှုကို ထိန်းသိမ်းထားသည်။

• လည်ပတ်ချိန်တွင် ထည့်သွင်းခြင်း- သီးခြားပြဿနာတစ်ခု (ဥပမာ "စာရင်းစစ်ဆေးခြင်း") နှင့် ရင်ဆိုင်ရသောအခါ၊ Agent သည် ကိရိယာ (load_skill) ကို တက်ကြွစွာခေါ်ယူပြီး ထိုစွမ်းရည်၏ အသေးစိတ်အကြောင်းအရာ (Schema + Prompt) ကို ထည့်သွင်းသည်။

• တာဝန်ကိုဆောင်ရွက်ခြင်း- ထည့်သွင်းထားသော တိကျသောအကြောင်းအရာအပေါ်အခြေခံ၍ သီးခြားတာဝန်ကို ဆောင်ရွက်သည် (ဥပမာ SQL ကိုရေးသားပြီး လုပ်ဆောင်ခြင်း)။

ဤမော်ဒယ်လ်သည် အကန့်အသတ်မရှိ တိုးချဲ့နိုင်မှု နှင့် အဖွဲ့ခွဲထွက်နိုင်မှု ကို ထိရောက်စွာပံ့ပိုးပေးပြီး Agent သည် တိုးများလာသော ရှုပ်ထွေးသည့် လုပ်ငန်းအခြေအနေများနှင့် လိုက်လျောညီထွေဖြစ်စေသည်။

၂။ စနစ်ဗိသုကာဒီဇိုင်း

ဤလက်တွေ့စီမံကိန်းသည် ဤမော်ဒယ်လ်၏ လက်တွေ့အသုံးချမှုကို သရုပ်ပြရန်အတွက် အဓိက Skills နှစ်ခုပါဝင်သော SQL Assistant ကို တည်ဆောက်မည်ဖြစ်သည်။

• Sales Analytics (အရောင်းခွဲခြမ်းစိတ်ဖြာခြင်း)-sales_dataဇယားကို တာဝန်ယူပြီး ဝင်ငွေစာရင်းအင်း၊ အမှာစာလမ်းကြောင်းခွဲခြမ်းစိတ်ဖြာခြင်းစသည်တို့ကို စီမံဆောင်ရွက်သည်။

• Inventory Management (စာရင်းစီမံခန့်ခွဲမှု)-inventory_itemsဇယားကို တာဝန်ယူပြီး စာရင်းအဆင့်စောင့်ကြည့်ခြင်း၊ တည်နေရာရှာဖွေခြင်းစသည်တို့ကို စီမံဆောင်ရွက်သည်။

၃။ ဖွံ့ဖြိုးတိုးတက်မှုပတ်ဝန်းကျင်ကို တည်ဆောက်ခြင်း

ဤစီမံကိန်းသည် Pythonuv ကို အသုံးပြု၍ ထိရောက်သော မှီခိုမှုစီမံခန့်ခွဲမှုကို လုပ်ဆောင်သည်။

အဓိကမှီခိုမှု တပ်ဆင်ခြင်း

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

PostgreSQL ပတ်ဝန်းကျင်ကို စီစဉ်သတ်မှတ်ခြင်း

ဒေသတွင်း Postgres instance ကို စတင်ပြီးagent_platformဒေတာဘေ့စ်ကို ဖန်တီးပါ။ ဇယားဖွဲ့စည်းပုံနှင့် စမ်းသပ်ဒေတာကို အလိုအလျောက် စတင်ရန်အတွက်setup_db.py script ကို ပံ့ပိုးပေးထားသည် (အသေးစိတ်အတွက် အဆုံးတွင်ရှိသော source code ကိုကြည့်ပါ)။

၄။ အဓိကအကောင်အထည်ဖော်မှုအဆင့်များ အသေးစိတ်ရှင်းပြချက်### အဆင့် ၁: နယ်ပယ်ကျွမ်းကျင်မှု (The Knowledge) ကိုသတ်မှတ်ပါ

ကျွမ်းကျင်မှုကို ဖိုင်စနစ် သို့မဟုတ် ဒေတာဘေ့စ်မှ တင်သွင်းခြင်းကို အတုယူကာ အဘိဓာန်ဖွဲ့စည်းပုံအဖြစ် သတ်မှတ်ပါမည်။ description (Agent ၏ ဆုံးဖြတ်ချက်ရွေးချယ်မှုအတွက် အသုံးပြုသည်) နှင့် content (အမှန်တကယ်တင်သွင်းထားသော အသေးစိတ်အကြောင်းအရာ) ကို ခွဲခြားသတ်မှတ်ရန် သတိပြုပါ။

SKILLS = {"sales_analytics": {"description":"အရောင်းရငွေ၊ လမ်းကြောင်းများကို ခွဲခြမ်းစိတ်ဖြာရန်အတွက် အသုံးဝင်သည်...","content":"""... Table Schema: sales_data ..."" },"inventory_management": {"description":"စတော့အဆင့်များကို စစ်ဆေးရန်အတွက် အသုံးဝင်သည်...","content":"""... Table Schema: inventory_items ..."" }}

အဆင့် ၂: အဓိကကိရိယာများ (The Capabilities) ကိုအကောင်အထည်ဖော်ပါ

Agent သည် လုပ်ငန်းတာဝန်များကို ပြီးမြောက်ရန်အတွက် အဓိကကိရိယာနှစ်ခုကို မှီခိုအားထားရသည်:

• load_skill(skill_name): သတ်မှတ်ထားသော ကျွမ်းကျင်မှု၏ အသေးစိတ်အချက်အလက်များကို အချိန်နှင့်တပြေးညီ တင်သွင်းပါ။

• run_sql_query(query): သီးခြား SQL ထုတ်ပြန်ချက်ကို လုပ်ဆောင်ပါ။

အဆင့် ၃: Agent ယုတ္တိဗေဒ (The Brain) ကို စီစဉ်ပါ

LangGraph ကို အသုံးပြု၍ ReAct Agent ကို တည်ဆောက်ပါ။ System Prompt သည် ဤနေရာတွင် အဓိကအခန်းကဏ္ဍမှ ပါဝင်ပြီး Identify -> Load -> Query ၏ စံလုပ်ငန်းစဉ် (SOP) ကို တင်းကြပ်စွာ လိုက်နာရန် Agent ကို လမ်းညွှန်ပေးသည်။

system_prompt ="""1. သက်ဆိုင်ရာ ကျွမ်းကျင်မှုကို ရှာဖွေပါ။2. schema ရယူရန် 'load_skill' ကို အသုံးပြုပါ။3. 'run_sql_query' ကို အသုံးပြု၍ SQL ကို ရေးသားပြီး လုပ်ဆောင်ပါ....Table အမည်များကို ခန့်မှန်းမနေပါနှင့်။ ကျွမ်းကျင်မှုကို အမြဲတမ်း ဦးစွာ တင်သွင်းပါ။"""

၅. လုပ်ဆောင်မှုအကျိုးသက်ရောက်မှုကို အတည်ပြုခြင်း

test_agent.py ကို လုပ်ဆောင်ခြင်းဖြင့် အရောင်းနှင့် စာရင်းထိန်းချုပ်ခြင်းနယ်ပယ်နှစ်ခုအတွက် စုံစမ်းမေးမြန်းမှုများကို စမ်းသပ်ခဲ့သည်။ အောက်တွင် Agent သည် ပြဿနာပေါ်မူတည်၍ ကျွမ်းကျင်မှုကို မည်သို့ တင်သွင်းသည်ကို ပြသထားသည့် console ၏ အမှန်တကယ်ထုတ်လွှတ်မှုမှတ်တမ်းဖြစ်သည်။

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**.## ၆။ အပြည့်အစုံရှိသော Source Code ကို ကိုးကားခြင်း

အောက်တွင် ပရောဂျက်၏ အပြည့်အစုံရှိသော source code ကို ဖော်ပြထားပါသည်။ ၎င်းတွင် database စတင်တည်ဆောက်ခြင်း script နှင့် Agent ၏ အဓိက program တို့ ပါဝင်ပါသည်။

၁။ Database စတင်တည်ဆောက်ခြင်း (setup_db.py)

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

load_dotenv()

.env ဖိုင်တွင် database ချိတ်ဆက်မှုအချက်အလက်များ ထည့်သွင်းထားကြောင်း သေချာပါစေ။

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") # အမှန်တကယ် password ဖြင့် အစားထိုးပါ။ DB_NAME = os.getenv("DB_NAME", "agent_platform")

def create_database(): try: # database အသစ်တစ်ခုတည်ဆောက်ရန်အတွက် default 'postgres' database သို့ ချိတ်ဆက်ပါ။ 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()

    # database ရှိပြီးသားလား စစ်ဆေးပါ။
    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()

    # 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)
        )
        """
    )

    # 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)
        )
        """
    )

    # Mock Data ထည့်သွင်းပါ။
    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 \u4e3b\u7a0b\u5e8f (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(\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("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 ="""You are a helpful SQL Assistant.You have access to specialized skills that contain database schemas and domain knowledge.To answer a user's question:1. Identify the relevant skill (sales_analytics or inventory_management).2. Use the 'load_skill' tool to get the schema and instructions.3. Based on the loaded skill, write and execute a SQL query using 'run_sql_query'.4. Answer the user's question based on the query results.Do not guess table names. Always load the skill first.""" print("SQL Assistant initialized. Type 'quit' to exit.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Pre-warm connection checktry: print(f"Connected to database:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Database connection warning:{e}")whileTrue:try: user_input = input("User: ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Stream the execution 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]

if isinstance(last_msg, AIMessage):

print(last_msg.content)

messages = final_state["messages"]

Update history

print("-"*50)

except Exception as e:

print(f"\nError:{e}")

break`