LangChain Skills 模式实战：构建按需加载知识的 SQL 助手
Katika makala iliyotangulia, tulichunguza jinsi ya kuiga muundo wa Deep Agent kwa kutumia Skills kupitia Deep Agents CLI. Hivi sasa, LangChain inasaidia kipengele hiki kiasili, na kurahisisha sana mchakato wa maendeleo. Makala hii itakuongoza kupitia uzoefu wa kina wa kipengele hiki, na kujenga msaidizi mahiri zaidi wa SQL.
Unapojenga AI Agent changamano, watengenezaji mara nyingi huangukia katika hali ngumu: Je, waingize muktadha wote (muundo wa jedwali la hifadhidata, hati za API, sheria za biashara) katika Prompt ya Mfumo mara moja, na kusababisha dirisha la muktadha (Context Window) kufurika na kutawanya umakini wa mfumo? Au uchague urekebishaji mzuri wa gharama kubwa (Fine-tuning)?
Muundo wa Skills hutoa njia ya kati ya kifahari. Inafanikisha matumizi bora ya muktadha kwa kupakia maarifa yanayohitajika kwa nguvu. Usaidizi asili wa LangChain kwa muundo huu unamaanisha kuwa tunaweza kujenga kwa urahisi zaidi Agent zenye uwezo wa "kujifunza inapohitajika".
Makala hii, pamoja na hati rasmi Jenga msaidizi wa SQL na ujuzi wa mahitaji, itawaongoza wasomaji kujenga Msaidizi wa SQL anayeunga mkono "upakiaji wa maarifa unapohitajika" kutoka mwanzo.
1. Dhana Muhimu: Kwa nini uchague muundo wa Skills?
Mapungufu ya Wakala wa Jadi wa SQL
Katika usanifu wa jadi wa Wakala wa SQL, kwa kawaida tunahitaji kutoa Schema kamili ya Hifadhidata katika Prompt ya Mfumo. Kadiri biashara inavyokua, wakati idadi ya jedwali inapanuka hadi mamia, njia hii itasababisha matatizo makubwa:
-
Matumizi makubwa ya Token: Kila mazungumzo hubeba idadi kubwa ya miundo ya jedwali isiyohusiana, na kusababisha upotevu wa rasilimali.
-
Hatari ya udanganyifu huongezeka: Habari nyingi za kuingilia zisizohusiana zitapunguza usahihi wa hoja ya mfumo.
-
Ugumu wa matengenezo: Maarifa yote ya mistari ya biashara yameunganishwa kwa karibu, na ni vigumu kurudia kwa kujitegemea.
Muundo wa Skills: Suluhisho kulingana na ufichuzi wa maendeleo
Muundo wa Skills unategemea kanuni ya ufichuzi wa maendeleo, na hutenganisha mchakato wa upataji wa maarifa:
-
Hali ya awali ya Wakala: Anajua tu "ujuzi" (Skills) ulipo na maelezo mafupi (Description), na kubaki mwepesi.
-
Upakiaji wa wakati wa utekelezaji: Wakati wa kukabiliana na tatizo maalum (kama vile "swali la hesabu"), Wakala huchukua hatua ya kupiga simu kwa zana (
load_skill) ili kupakia muktadha wa kina wa ujuzi (Schema + Prompt). -
Tekeleza kazi: Kulingana na muktadha sahihi uliopakiwa, tekeleza kazi maalum (kama vile kuandika na kutekeleza SQL).
Muundo huu unaunga mkono kwa ufanisi upanuzi usio na kikomo na utengano wa timu, kuwezesha Wakala kukabiliana na matukio ya biashara yanayozidi kuwa magumu.
2. Ubunifu wa Usanifu wa Mfumo
Mradi huu wa vitendo utajenga Msaidizi wa SQL ambaye anajumuisha Skills mbili muhimu ili kuonyesha matumizi halisi ya muundo:
-
Sales Analytics: Inawajibika kwa jedwali la
sales_data, kushughulikia takwimu za mapato, uchambuzi wa mwelekeo wa agizo, nk. -
Inventory Management: Inawajibika kwa jedwali la
inventory_items, kushughulikia ufuatiliaji wa kiwango cha hesabu, swali la eneo, nk.
3. Kuanzisha Mazingira ya Maendeleo
Mradi huu unatumia Pythonuv kwa usimamizi mzuri wa utegemezi.
Ufungaji wa utegemezi muhimu
uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community
Usanidi wa mazingira ya PostgreSQL
Anzisha mfano wa Postgres ndani ya nchi, na uunde hifadhidata ya agent_platform. Tumetoa hati ya setup_db.py ili kuanzisha kiotomatiki muundo wa jedwali na data ya majaribio (tazama msimbo wa chanzo mwishoni mwa makala).
4. Ufafanuzi wa Hatua Muhimu za Utekelezaji### Hatua ya Kwanza: Kufafanua Ujuzi wa Kikoa (Maarifa)
Tutaelezea ujuzi kama muundo wa kamusi, kuiga mchakato wa kupakia kutoka kwa mfumo wa faili au hifadhidata. Tafadhali kumbuka kutofautisha kati ya description (inayotumiwa na Wakala kwa uteuzi wa uamuzi) na content (mukhtasari halisi uliopakiwa).
SKILLS = {"sales_analytics": {"description":"Muhimu kwa kuchambua mapato ya mauzo, mitindo...","content":"""... Schema ya Jedwali: sales_data ..."" },"inventory_management": {"description":"Muhimu kwa kuangalia viwango vya hisa...","content":"""... Schema ya Jedwali: inventory_items ..."" }}
Hatua ya Pili: Kutekeleza Zana Muhimu (Uwezo)
Wakala anategemea zana mbili muhimu kukamilisha kazi:
-
load_skill(skill_name)**: Hupakia maelezo ya ujuzi maalum kwa nguvu wakati wa utekelezaji. ** -
run_sql_query(query)**: Hufanya taarifa maalum za SQL. **
Hatua ya Tatu: Kupanga Mantiki ya Wakala (Ubongo)
Tumia LangGraph kujenga Wakala wa ReAct. Kidokezo cha Mfumo kina jukumu muhimu hapa, kuelekeza Wakala kufuata kwa ukali utaratibu wa kawaida wa uendeshaji (SOP) wa Identify -> Load -> Query.
system_prompt ="""1. Tambua ujuzi unaofaa.2. Tumia 'load_skill' kupata schema.3. Andika na utekeleze SQL ukitumia 'run_sql_query'....Usikisie majina ya jedwali. Daima pakia ujuzi kwanza.""
5. Uthibitishaji wa Athari za Uendeshaji
Kwa kuendesha test_agent.py, tulijaribu maswali katika nyanja mbili tofauti: Mauzo na Hesabu. Hapa kuna kumbukumbu halisi za pato la koni, zinazoonyesha jinsi Wakala anavyopakia ujuzi kwa nguvu kulingana na swali:
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. Marejeleo Kamili ya Msimbo Chanzo
Ifuatayo ni msimbo kamili wa chanzo wa mradi, unaojumuisha hati ya uanzishaji wa hifadhidata na programu kuu ya Agent.
1. Uanzishaji wa Hifadhidata (setup_db.py)
`import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT import os from dotenv import load_dotenv
load_dotenv()
Tafadhali hakikisha kuwa umesanidi maelezo ya muunganisho wa hifadhidata katika .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") # Tafadhali badilisha na nenosiri halisi DB_NAME = os.getenv("DB_NAME", "agent_platform")
def create_database(): try: # Unganisha kwenye hifadhidata chaguo-msingi ya 'postgres' ili kuunda hifadhidata mpya 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()
# Angalia ikiwa hifadhidata ipo
cur.execute(f"SELECT 1 FROM pg_catalog.pg_database WHERE datname = '{DB_NAME}'")
exists = cur.fetchone()
if not exists:
print(f"Inaunda hifadhidata {DB_NAME}...")
cur.execute(f"CREATE DATABASE {DB_NAME}")
else:
print(f"Hifadhidata {DB_NAME} tayari ipo.")
cur.close()
conn.close()
except Exception as e:
print(f"Kosa wakati wa kuunda hifadhidata: {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()
# Unda Jedwali la Mauzo
print("Inaunda jedwali la sales_data...")
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)
)
"""
)
# Unda Jedwali la Hesabu
print("Inaunda jedwali la inventory_items...")
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)
)
"""
)
# Ingiza Data Bandia
print("Inaingiza data bandia...")
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("Usanidi wa hifadhidata umekamilika.")
except Exception as e:
print(f"Kosa wakati wa kusanidi majedwali: {e}")
if name == "main": create_database() create_tables_and_data() `### 2. Agent 主程序 (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()
--- 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""" Jedwali. Schema ya Jedwali:
- id: integer (ufunguo msingi)
- product_id: varchar(50)
- product_name: varchar(100)
- stock_count: integer
- warehouse_location: varchar(50) Maswali ya kawaida:
- Angalia hisa: WHERE product_name = '...'
- Hisa ndogo: WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:"""
Pakia kidokezo cha kina na schema kwa ujuzi maalum.
Ujuzi unaopatikana:
- sales_analytics: Kwa uchambuzi wa mauzo, mapato na miamala.
- inventory_management: Kwa maswali ya hisa, bidhaa na ghala. """ ujuzi = SKILLS.get(skill_name) kama sio ujuzi: rudisha f"Hitilafu: Ujuzi '{skill_name}' haupatikani. Ujuzi unaopatikana:{list(SKILLS.keys())}" rudisha ujuzi["content"] @tool def run_sql_query(query: str) -> str: """Tekeleza swali la SQL dhidi ya hifadhidata. Tumia tu zana hii BAADA ya kupakia ujuzi unaofaa ili kuelewa schema. """ jaribu: rudisha db.run(query) isipokuwa Exception kama e: rudisha f"Hitilafu wakati wa kutekeleza SQL:{e}" @tool def list_tables() -> str: """Orodhesha majedwali yote yanayopatikana kwenye hifadhidata.""" rudisha str(db.get_usable_table_names()) zana = [load_skill, run_sql_query, list_tables]
--- Usanidi wa Wakala ---
llm = ChatOpenAI( base_url=BASE_URL, api_key=API_KEY, model=MODEL_NAME, temperature=0 ) llm_with_tools = llm.bind_tools(zana)
--- Ufafanuzi wa Grafu ---
class AgentState(MessagesState):
Tunaweza kuongeza hali maalum ikiwa inahitajika, lakini MessagesState inatosha kwa mazungumzo rahisi
pitisha
def agent_node(state: AgentState): ujumbe = state["messages"] jibu = llm_with_tools.invoke(ujumbe) rudisha {"messages": [jibu]}
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()# --- Utekelezaji Mkuu ---if__name__ =="main": system_prompt ="""Wewe ni Msaidizi wa SQL msaidizi. Una ufikiaji wa ujuzi maalum ambao una schemata za hifadhidata na maarifa ya kikoa. Ili kujibu swali la mtumiaji:
- Tambua ujuzi unaohusika (sales_analytics au inventory_management).
- Tumia zana ya 'load_skill' kupata schema na maagizo.
- Kulingana na ujuzi uliopakiwa, andika na utekeleze swali la SQL kwa kutumia 'run_sql_query'.
- Jibu swali la mtumiaji kulingana na matokeo ya swali.
Usikisie majina ya jedwali. Daima pakia ujuzi kwanza.""" print("Msaidizi wa SQL ameanzishwa. Andika 'quit' ili kutoka.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Angalia muunganisho wa awali kabla ya matumizi
try:
print(f"Imeunganishwa kwenye hifadhidata:{DB_URI.split('@')[-1]}")
exceptExceptionase:
print(f"Onyo la muunganisho wa hifadhidata:{e}")
whileTrue:
try:
user_input = input("Mtumiaji: ")
ifuser_input.lower()in["quit","exit"]:
break
messages.append(HumanMessage(content=user_input))
Tiririsha utekelezaji
print("Agent: ", end="", flush=True) final_response =None foreventinapp.stream({"messages": messages}, stream_mode="values"): # Katika hali ya 'values', tunapata hali kamili. Tunataka tu kuona ujumbe wa mwisho ikiwa ni mpya. last_message = event["messages"][-1] # Sasisha historia yetu ya ujumbe na hali ya hivi karibuni passBaada ya mtiririko kumalizika, hali ya mwisho ina jibu la mwisho
final_state = app.invoke({"messages": messages}) last_msg = final_state["messages"][-1]ifisinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]# Sasisha historia print("-"*50)exceptExceptionase: print(f"\nHitilafu:{e}")break`
