LangChain Skills загварын туршлага: Шаардлагатай үед мэдлэг ачаалах SQL туслах бүтээх

Өмнөх нийтлэлдээ бид Deep Agents CLI-ээр дамжуулан Deep Agent-ийн Skills ашиглах загварыг хэрхэн дуурайх талаар судалсан. Өнөөдөр LangChain энэ функцийг эх хэлээрээ дэмжиж, хөгжүүлэлтийн процессыг эрс хялбаршуулж байна. Энэхүү нийтлэл нь та бүхнийг энэ функцийг гүнзгийрүүлэн судлах, илүү ухаалаг SQL туслах бүтээхэд чиглүүлэх болно.

Нарийн төвөгтэй AI Agent бүтээхдээ хөгжүүлэгчид ихэвчлэн хоёр сонголтын аль нэгийг хийхэд хүрч байна: бүх контекст (өгөгдлийн сангийн хүснэгтийн бүтэц, API баримт бичиг, бизнесийн дүрэм) -ийг System Prompt-д нэг дор оруулж, контекстийн цонх (Context Window) хальж, загварын анхаарлыг сарниулах уу? Эсвэл өндөр өртөгтэй давтамжтай нарийн тохируулга (Fine-tuning) хийх үү?

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 instance ажиллуулж, agent_platform өгөгдлийн санг үүсгэ. Бид хүснэгтийн бүтэц болон туршилтын өгөгдлийг автоматаар эхлүүлэх setup_db.py скриптийг санал болгож байна (дэлгэрэнгүйг эх кодоос үзнэ үү).

4. Үндсэн хэрэгжилтийн алхмуудын дэлгэрэнгүй тайлбар

### **Алхам 1: Салбарын ур чадварыг тодорхойлох (Мэдлэг)** Бид ур чадварыг толь бичгийн бүтэц хэлбэрээр тодорхойлж, файлын систем эсвэл мэдээллийн сангаас ачаалах процессыг дуурайлган үзүүлнэ. `description` (Агентын шийдвэр гаргах сонголтод ашиглах) болон `content` (бодит ачаалсан дэлгэрэнгүй контекст)-ийг ялгаж салгахыг анхаарна уу. `SKILLS = {"sales_analytics": {"description":"Борлуулалтын орлого, чиг хандлагыг шинжлэхэд хэрэгтэй...","content":"""... Хүснэгтийн бүдүүвч: sales_data ..."" },"inventory_management": {"description":"Бараа материалын түвшинг шалгахад хэрэгтэй...","content":"""... Хүснэгтийн бүдүүвч: inventory_items ..."" }}` ### **Алхам 2: Гол хэрэгслүүдийг хэрэгжүүлэх (Чадавхи)** Агент нь даалгаврыг гүйцэтгэхийн тулд хоёр чухал хэрэгсэлд тулгуурладаг: - `load_skill(skill_name)`**: Заасан ур чадварын дэлгэрэнгүй мэдээллийг ажиллах үед динамикаар ачаалах.** - `run_sql_query(query)`**: Тодорхой SQL өгүүлбэрийг гүйцэтгэх.** ### **Алхам 3: Агентын логикийг зохион байгуулах (Тархи)** LangGraph ашиглан ReAct Агентыг бүтээх. System Prompt нь энд чухал үүрэг гүйцэтгэдэг бөгөөд Агентыг `Identify -> Load -> Query` стандарт ажлын журмыг (SOP) чандлан дагаж мөрдөхөд чиглүүлдэг. `system_prompt ="""1. Холбогдох ур чадварыг тодорхойлох.2. Бүдүүвчийг авахын тулд 'load_skill'-ийг ашиглах.3. 'run_sql_query'-г ашиглан SQL бичиж, гүйцэтгэх....Хүснэгтийн нэрийг таах хэрэггүй. Ур чадварыг эхлээд үргэлж ачаалах хэрэгтэй."""` ## **5. Ажиллах үр дүнг баталгаажуулах** `test_agent.py`-г ажиллуулснаар бид Sales болон Inventory гэсэн хоёр өөр салбарын асуулгыг туршиж үзсэн. Доорх нь Агент асуудлын дагуу ур чадварыг хэрхэн динамикаар ачаалж байгааг харуулсан консолын бодит гаралтын бүртгэл юм: `Testing Sales Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'sales_analytics'}, 'id': 'call_f270d76b7ce4404cb5f61bf1', 'type': 'tool_call'}]Tool output:Та борлуулалтын шинжилгээний мэргэжилтэн.Танд 'sales_data' хүснэгтэд хандах эрх бий.Хүснэгтийн бүдүүвч:- 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: Нийт орлого $730.50 байна.Testing Inventory Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'inventory_management'}, 'id': 'call_18c823b2d5064e95a0cfe2e3', 'type': 'tool_call'}]Tool output:Та бараа материалын удирдлагын мэргэжилтэн.Танд 'inventory_items' хүснэгтэд хандах эрх бий.Хүснэгтийн бүдүүвч...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: Зөөврийн компьютер **Агуулах 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) ) """ ) # Хуурамч өгөгдөл оруулах 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()# --- Тохиргоо ---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: бүхэл тоо (үндсэн түлхүүр)- transaction_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' хүснэгт. Хүснэгтийн бүдүүвч: - id: бүхэл тоо (үндсэн түлхүүр) - product_id: varchar(50) - product_name: varchar(100) - stock_count: бүхэл тоо - warehouse_location: varchar(50) Нийтлэг асуултууд: - Нөөцийг шалгах: WHERE product_name = '...' - Бага нөөц: WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:""" Тодорхой ур чадварын дэлгэрэнгүй prompt болон бүдүүвчийг ачаалах. Боломжтой ур чадварууд: - sales_analytics: Борлуулалт, орлого, гүйлгээний шинжилгээнд зориулагдсан. - inventory_management: Нөөц, бүтээгдэхүүн, агуулахын асуултанд зориулагдсан. """ skill = SKILLS.get(skill_name) if not skill: return f"Алдаа: '{skill_name}' ур чадвар олдсонгүй. Боломжтой ур чадварууд: {list(SKILLS.keys())}" return skill["content"] @tool def run_sql_query(query: str) -> str: """ Мэдээллийн сангаас SQL асуулгыг гүйцэтгэх. Бүдүүвчийг ойлгохын тулд зохих ур чадварыг ачаалсны дараа л энэ хэрэгслийг ашиглана уу. """ try: return db.run(query) except Exception as e: return f"SQL гүйцэтгэхэд алдаа гарлаа: {e}" @tool def list_tables() -> str: """Мэдээллийн санд байгаа бүх хүснэгтийг жагсаах.""" 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): # Шаардлагатай бол бид өөрчлөн тохируулах төлөв нэмж болно, гэхдээ MessagesState нь энгийн чатлахад хангалттай 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()# --- Үндсэн гүйцэтгэл ---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]ifisinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]# Update history print("-"*50)exceptExceptionase: print(f"\nError:{e}")break`