Në artikullin e mëparshëm, ne eksploruam se si të simulojmë modelin e përdorimit të Skills nga Deep Agent përmes Deep Agents CLI. Sot, LangChain e mbështet këtë veçori në mënyrë native, duke thjeshtuar jashtëzakonisht procesin e zhvillimit. Ky artikull do t'ju udhëheqë të përjetoni thellësisht këtë funksionalitet, duke ndërtuar një asistent SQL më inteligjent.

Gjatë ndërtimit të Agjentëve të ndërlikuar AI, zhvilluesit shpesh gjenden në një dilemë: a duhet të injektojnë të gjithë kontekstin (struktura e tabelave të bazës së të dhënave, dokumentacioni API, rregullat e biznesit) njëherësh në System Prompt, duke shkaktuar tejmbushje të dritares së kontekstit (Context Window) dhe shpërqendrim të vëmendjes së modelit? Apo të zgjedhin rregullim të shpeshtë (Fine-tuning) me kosto të lartë?

Modeli Skills (Skills Pattern) ofron një rrugë të ndërmjetme elegante. Ai realizon përdorimin efikas të kontekstit duke ngarkuar dinamikisht njohuritë e nevojshme. Mbështetja native e LangChain për këtë model do të thotë se ne mund të ndërtojmë më lehtë Agjentë që kanë aftësinë e "të mësuarit sipas kërkesës".

Ky artikull do t'i udhëzojë lexuesit nga e para, në kombinim me dokumentacionin zyrtar Build a SQL assistant with on-demand skills, për të ndërtuar një Asistent SQL që mbështet "ngarkimin e njohurive sipas kërkesës".

1. Koncepti thelbësor: Pse të zgjidhni modelin Skills?

Kufizimet e Agjentit tradicional SQL

Në arkitekturën tradicionale të Agjentit SQL, ne zakonisht duhet të ofrojmë Schema të plotë të Bazës së të Dhënave në System Prompt. Me zhvillimin e biznesit, kur numri i tabelave zgjerohet në qindra, kjo metodë do të sjellë probleme të rëndësishme:

• Konsum i madh i Token: Çdo dialog mbart një numër të madh strukturash tabelash të parëndësishme, duke shkaktuar humbje të burimeve.

• Rritje e rrezikut të halucinacioneve: Shumë informacione të parëndësishme do të ulin saktësinë e arsyetimit të modelit.

• Mirëmbajtje e vështirë: Të gjitha njohuritë e linjave të biznesit janë të lidhura ngushtë, duke e bërë të vështirë përsëritjen e pavarur.

Modeli Skills: Zgjidhje e bazuar në zbulimin progresiv

Modeli Skills bazohet në parimin e zbulimit progresiv (Progressive Disclosure), duke trajtuar procesin e marrjes së njohurive në shtresa:

• Gjendja fillestare e Agjentit: Zotëron vetëm se cilat "aftësi" (Skills) ekzistojnë dhe përshkrimin e tyre të shkurtër (Description), duke e mbajtur të lehtë.

• Ngarkimi gjatë ekzekutimit: Kur përballet me një problem specifik (si p.sh. "kërkimi i inventarit"), Agjenti thërret në mënyrë aktive mjetin (load_skill) për të ngarkuar kontekstin e detajuar të kësaj aftësie (Schema + Prompt).

• Ekzekutimi i detyrës: Bazuar në kontekstin e saktë të ngarkuar, ekzekuton detyra specifike (si p.sh. shkrimi dhe ekzekutimi i SQL).

Ky model mbështet në mënyrë efektive zgjerimin e pakufizuar dhe shkëputjen e ekipit, duke i mundësuar Agjentit të përshtatet me skenarë biznesi gjithnjë e më kompleksë.

2. Dizajni i Arkitekturës së Sistemit

Ky projekt praktik do të ndërtojë një Asistent SQL që përmban dy Skills thelbësore, për të demonstruar aplikimin praktik të këtij modeli:

• Analiza e Shitjeve (Sales Analytics): Përgjegjës për tabelënsales_data, duke trajtuar statistikat e të ardhurave, analizën e tendencave të porosive, etj.

• Menaxhimi i Inventarit (Inventory Management): Përgjegjës për tabelëninventory_items, duke trajtuar monitorimin e nivelit të inventarit, kërkimin e vendndodhjes, etj.

3. Ndërtimi i Mjedisit të Zhvillimit

Ky projekt përdor Pythonuv për menaxhim efikas të varësive.

Instalimi i varësive thelbësore

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

Konfigurimi i mjedisit PostgreSQL

Nisni një instancë Postgres lokale dhe krijoni bazën e të dhënaveagent_platform. Ne ofrojmë skriptinsetup_db.py për të inicializuar automatikisht strukturën e tabelës dhe të dhënat e testimit (shih kodin burimor në fund të artikullit).

4. Shpjegimi i detajuar i hapave thelbësorë të implementimit### Hapi i Parë: Përcaktimi i Aftësive të Fushës (Njohuria)

Ne do t'i përcaktojmë aftësitë si një strukturë fjalori, duke simuluar procesin e ngarkimit nga një sistem skedari ose bazë të dhënash. Ju lutemi vini re dallimin midis description (përdoret nga Agjenti për të marrë vendime për zgjedhjen) dhe content (konteksti i detajuar aktual i ngarkuar).

SKILLS = {"sales_analytics": {"description":"E dobishme për analizimin e të ardhurave nga shitjet, tendencat...","content":"""... Skema e Tabelës: sales_data ..."" },"inventory_management": {"description":"E dobishme për kontrollimin e niveleve të aksioneve...","content":"""... Skema e Tabelës: inventory_items ..."" }}

Hapi i Dytë: Zbatimi i Mjeteve Themelore (Aftësitë)

Agjenti varet nga dy mjete kryesore për të përfunduar detyrat:

• load_skill(skill_name)**: Ngarkon dinamikisht detajet e aftësisë së specifikuar gjatë ekzekutimit. **

• run_sql_query(query)**: Ekzekuton deklarata specifike SQL. **

Hapi i Tretë: Orkestrimi i Logjikës së Agjentit (Truri)

Përdorni LangGraph për të ndërtuar Agjentin ReAct. System Prompt luan një rol kyç këtu, duke udhëzuar Agjentin të ndjekë në mënyrë rigoroze procedurën standarde të funksionimit (SOP) Identify -> Load -> Query.

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. Verifikimi i Efektit të Ekzekutimit

Duke ekzekutuar test_agent.py, ne kemi testuar pyetjet në dy fusha të ndryshme, Shitjet dhe Inventari. Më poshtë janë regjistrat aktualë të daljes së konsolës, duke treguar se si Agjenti ngarkon dinamikisht aftësitë bazuar në pyetje:

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. Referenca e plotë e kodit burimor

Më poshtë është kodi i plotë burimor i projektit, duke përfshirë skriptin e inicializimit të bazës së të dhënave dhe programin kryesor të Agjentit.

1. Inicializimi i bazës së të dhënave (setup_db.py)

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

load_dotenv()

Ju lutemi sigurohuni që të konfiguroni informacionin e lidhjes së bazës së të dhënave në .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") # Ju lutemi zëvendësoni me fjalëkalimin aktual DB_NAME = os.getenv("DB_NAME", "agent_platform")

def create_database(): try: # Lidhuni me bazën e të dhënave të parazgjedhur 'postgres' për të krijuar bazën e re të të dhënave 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()

    # Kontrollo nëse baza e të dhënave ekziston
    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()

    # Krijo tabelën e shitjeve
    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)
        )
    """)

    # Krijo tabelën e inventarit
    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)
        )
    """)

    # Fut të dhëna të simuluara
    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)

importosfromtypingimportAnnotated, 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("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')}"# --- Database Setup ---db = SQLDatabase.from_uri(DB_URI)# --- Skills Definition ---SKILLS: Dict[str, Dict[str, str]] = {"sales_analytics": {"description":"Useful for analyzing sales revenue, trends, and regional performance.","content":"""You are a Sales Analytics Expert.You have access to the 'sales_data' table.Table Schema:- id: integer (primary key)- transaction_date: date- product_id: varchar(50)- amount: decimal(10, 2)- region: varchar(50)Common queries:- Total revenue: SUM(amount)- Revenue by region: GROUP BY region- Sales trend: GROUP BY transaction_date""" },"inventory_management": {"description":"Useful for checking stock levels, product locations, and warehouse management.","content":"""You are an Inventory Management Expert.You have access to the '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:e)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]ifisinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]# Update history print("-"*50)exceptExceptionase: print(f"\nError:{e}")break