Во претходната статија, истражувавме како да симулираме Deep Agent користејќи Skills модел преку Deep Agents CLI. Денес, LangChain веќе природно ја поддржува оваа функција, што значително го поедноставува процесот на развој. Оваа статија ќе ве води низ оваа функција и ќе ви помогне да изградите поинтелигентен SQL асистент.

При градење сложени AI агенти, програмерите често се наоѓаат во дилема: дали да ги инјектираат сите контексти (структура на базата на податоци, API документација, деловни правила) во System Prompt одеднаш, што доведува до прелевање на контекстниот прозорец (Context Window) и го одвлекува вниманието на моделот? Или да изберат скапо и често фино подесување (Fine-tuning)?

Skills модел (Skills Pattern) обезбедува елегантно средно решение. Со динамичко вчитување на потребните знаења, тој постигнува ефикасно користење на контекстот. Природната поддршка на LangChain за овој модел значи дека можеме полесно да изградиме агент со способност за „учење на барање“.

Оваа статија ќе го комбинира официјалниот документ Build a SQL assistant with on-demand skills за да ги води читателите од нула да изградат SQL Assistant кој поддржува „вчитување знаење на барање“.

1. Клучен концепт: Зошто да изберете Skills модел?

Ограничувања на традиционалниот SQL Agent

Во традиционалната архитектура на SQL Agent, обично треба да ја обезбедиме целосната Database Schema во System Prompt. Како што се развива бизнисот, кога бројот на табели се проширува на стотици, овој метод ќе донесе значителни проблеми:

• Огромна потрошувачка на токени: Секој разговор носи голема количина на нерелевантни структури на табели, што доведува до губење на ресурси.

• Зголемен ризик од халуцинации: Премногу нерелевантни информации за пречки ќе ја намалат точноста на заклучувањето на моделот.

• Тешкотии во одржувањето: Знаењето на сите деловни линии е тесно поврзано, што го отежнува независното повторување.

Skills модел: Решение засновано на прогресивно откривање

Skills моделот се заснова на принципот на прогресивно откривање (Progressive Disclosure), кој го дели процесот на стекнување знаење на слоеви:

• Почетна состојба на Agent: Само знае кои „вештини“ (Skills) постојат и нивните кратки описи (Description), одржувајќи го лесен.

• Вчитување за време на извршување: Кога се соочува со специфичен проблем (како што е „барање залиха“), Agent активно повикува алатка (load_skill) за да го вчита деталниот контекст на таа вештина (Schema + Prompt).

• Извршување на задачата: Врз основа на вчитаниот прецизен контекст, извршете специфични задачи (како што е пишување и извршување на SQL).

Овој модел ефикасно поддржува неограничено проширување и тимско раздвојување, овозможувајќи му на Agent да се прилагоди на сè посложените деловни сценарија.

2. Дизајн на системска архитектура

Овој практичен проект ќе изгради SQL Assistant кој содржи две основни 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. Детално објаснување на основните чекори за имплементација### Чекор еден: Дефинирање на вештини од областа (The Knowledge)

Вештините ќе ги дефинираме како структура од речник, симулирајќи го процесот на вчитување од датотечен систем или база на податоци. Ве молиме забележете ја разликата помеѓу description (се користи за избор на Agent при донесување одлуки) и 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 ..."" }}

Чекор два: Имплементација на основни алатки (The Capabilities)

Agent-от зависи од две клучни алатки за да ги заврши задачите:

• load_skill(skill_name)**: Динамички ги вчитува деталите за одредена вештина за време на извршувањето. **

• run_sql_query(query)**: Извршува конкретни SQL изјави. **

Чекор три: Организирање на логиката на Agent-от (The Brain)

Користете LangGraph за да изградите ReAct Agent. System Prompt игра клучна улога овде, насочувајќи го Agent-от строго да го следи стандардниот оперативен процес (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. Верификација на ефектот на извршување

Со извршување на test_agent.py, ги тестиравме прашањата од две различни области, Sales и Inventory. Подолу се дадени вистинските излезни логови од конзолата, кои покажуваат како Agent-от динамички вчитува вештини според прашањето:

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. Целосен изворен код - референца

Следново е целосниот изворен код на проектот, кој ги вклучува скриптата за иницијализација на базата на податоци и главната програма на Agent.

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: # Поврзете се со стандардната база на податоци '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()

    # Креирајте табела Sales
    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
    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 主程序 (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_itemTable 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:\n""" 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:\n""" 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:\n"""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. Идентификувајте ја релевантната вештина (анализа на продажба или управување со залихи).2. Користете ја алатката 'load_skill' за да ги добиете шемата и инструкциите.3. Врз основа на вчитаната вештина, напишете и извршете SQL query користејќи 'run_sql_query'.4. Одговорете на прашањето на корисникот врз основа на резултатите од query-то.Не погодувајте имиња на табели. Секогаш прво вчитајте ја вештината.""" 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' mode, го добиваме целосниот статус. Ние само сакаме да ја видиме последната порака ако е нова. 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`