LangChain Skills 模式实战：构建按需加载知识的 SQL 助手
Dalam artikel sebelumnya, kita telah meneroka cara meniru corak penggunaan Skills oleh Deep Agent melalui Deep Agents CLI. Kini, LangChain telah menyokong ciri ini secara natif, yang sangat memudahkan proses pembangunan. Artikel ini akan membawa anda untuk mengalami fungsi ini secara mendalam dan membina pembantu SQL yang lebih pintar.
Apabila membina AI Agent yang kompleks, pembangun sering menghadapi dilema: adakah untuk menyuntik semua konteks (struktur jadual pangkalan data, dokumentasi API, peraturan perniagaan) ke dalam System Prompt sekaligus, menyebabkan tetingkap konteks (Context Window) melimpah dan mengganggu tumpuan model? Atau adakah untuk memilih penalaan halus (Fine-tuning) yang kerap dan mahal?
Skills 模式（Skills Pattern） menyediakan jalan tengah yang elegan. Ia mencapai penggunaan konteks yang cekap dengan memuatkan pengetahuan yang diperlukan secara dinamik. Sokongan natif LangChain untuk corak ini bermakna kita boleh membina Agent yang mempunyai keupayaan "pembelajaran atas permintaan" dengan lebih mudah.
Artikel ini akan menggabungkan dokumentasi rasmi Build a SQL assistant with on-demand skills, membimbing pembaca dari awal untuk membina SQL Assistant yang menyokong "pemuatan pengetahuan atas permintaan".
1. 核心概念：为何选择 Skills 模式？
传统 SQL Agent 的局限性
Dalam seni bina SQL Agent tradisional, kita biasanya perlu menyediakan Skema Pangkalan Data yang lengkap dalam System Prompt. Apabila bilangan jadual berkembang kepada beratus-ratus dengan perkembangan perniagaan, kaedah ini akan membawa masalah yang ketara:
-
Penggunaan Token yang besar: Setiap perbualan membawa sejumlah besar struktur jadual yang tidak berkaitan, menyebabkan pembaziran sumber.
-
Peningkatan risiko halusinasi: Terlalu banyak maklumat gangguan yang tidak berkaitan akan mengurangkan ketepatan penaakulan model.
-
Kesukaran penyelenggaraan: Pengetahuan semua barisan perniagaan digabungkan rapat, menjadikannya sukar untuk diulang secara bebas.
Skills 模式：基于渐进式披露的解决方案
Skills 模式基于**渐进式披露（Progressive Disclosure）**原则，将知识获取过程分层处理：
-
Keadaan awal Agent: Hanya menguasai "kemahiran" (Skills) yang ada dan penerangan ringkas (Description), mengekalkan ringan.
-
Pemuatan masa jalan: Apabila menghadapi masalah khusus (seperti "pertanyaan inventori"), Agent secara aktif memanggil alat (
load_skill) untuk memuatkan konteks terperinci kemahiran (Skema + Prompt). -
Melaksanakan tugas: Berdasarkan konteks tepat yang dimuatkan, melaksanakan tugas khusus (seperti menulis dan melaksanakan SQL).
Corak ini menyokong pengembangan tanpa had dan penyahgandingan pasukan dengan berkesan, membolehkan Agent menyesuaikan diri dengan senario perniagaan yang semakin kompleks.
2. 系统架构设计
Projek amali ini akan membina SQL Assistant yang mengandungi dua Skills teras untuk menunjukkan aplikasi praktikal corak ini:
-
Sales Analytics（销售分析）：负责
sales_data表，处理收入统计、订单趋势分析等。 -
Inventory Management（库存管理）：负责
inventory_items表，处理库存水平监控、位置查询等。
3. 开发环境搭建
Projek ini menggunakan Pythonuv untuk pengurusan pergantungan yang cekap.
核心依赖安装
uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community
PostgreSQL 环境配置
Mulakan contoh Postgres tempatan dan buat pangkalan data agent_platform. Kami menyediakan skrip setup_db.py untuk memulakan struktur jadual dan data ujian secara automatik (lihat kod sumber di akhir artikel).
4. 核心实现步骤详解### Langkah Satu: Mentakrifkan Kemahiran Domain (The Knowledge)
Kita akan mentakrifkan kemahiran sebagai struktur kamus, mensimulasikan proses memuatkan daripada sistem fail atau pangkalan data. Sila ambil perhatian perbezaan antara description (untuk Agent membuat keputusan pemilihan) dan content (konteks terperinci yang dimuatkan sebenar).
SKILLS = {"sales_analytics": {"description":"Berguna untuk menganalisis hasil jualan, trend...","content":"""... Skema Jadual: sales_data ..."" },"inventory_management": {"description":"Berguna untuk menyemak tahap stok...","content":"""... Skema Jadual: inventory_items ..."" }}
Langkah Dua: Melaksanakan Alat Teras (The Capabilities)
Agent bergantung pada dua alat penting untuk menyelesaikan tugas:
-
load_skill(skill_name)**: Memuatkan butiran kemahiran yang ditentukan secara dinamik semasa masa jalan. ** -
run_sql_query(query)**: Melaksanakan pernyataan SQL tertentu. **
Langkah Tiga: Menyusun Logik Agent (The Brain)
Manfaatkan LangGraph untuk membina ReAct Agent. System Prompt memainkan peranan penting di sini, ia membimbing Agent untuk mematuhi dengan ketat prosedur operasi standard (SOP) Identify -> Load -> Query.
system_prompt ="""1. Kenal pasti kemahiran yang berkaitan.2. Gunakan 'load_skill' untuk mendapatkan skema.3. Tulis dan laksanakan SQL menggunakan 'run_sql_query'....Jangan teka nama jadual. Sentiasa muatkan kemahiran dahulu.""
5. Pengesahan Kesan Pelaksanaan
Dengan menjalankan test_agent.py, kami menguji pertanyaan dalam dua domain berbeza iaitu Jualan dan Inventori. Berikut ialah log output sebenar daripada konsol, yang menunjukkan cara Agent memuatkan kemahiran secara dinamik berdasarkan soalan:
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. Rujukan Kod Sumber Lengkap
Berikut ialah kod sumber lengkap projek, termasuk skrip permulaan pangkalan data dan program utama Agent.
1. Permulaan Pangkalan Data (setup_db.py)
`import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT import os from dotenv import load_dotenv
load_dotenv()
Sila pastikan maklumat sambungan pangkalan data dikonfigurasikan dalam .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") # Sila gantikan dengan kata laluan sebenar DB_NAME = os.getenv("DB_NAME", "agent_platform")
def create_database(): try: # Sambung ke pangkalan data 'postgres' lalai untuk mencipta db baharu 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()
# Semak sama ada pangkalan data wujud
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()
# Cipta Jadual Jualan
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)
)
"""
)
# Cipta Jadual Inventori
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)
)
"""
)
# Masukkan Data Mock
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":"Berguna untuk menganalisis hasil jualan, trend, dan prestasi serantau.","content":"""Anda ialah Pakar Analisis Jualan.Anda mempunyai akses kepada jadual 'sales_data'.Skema Jadual:- id: integer (kunci utama)- transaction_date: date- product_id: varchar(50)- amount: decimal(10, 2)- region: varchar(50)Pertanyaan biasa:- Jumlah hasil: SUM(amount)- Hasil mengikut rantau: GROUP BY region- Trend jualan: GROUP BY transaction_date""" },"inventory_management": {"description":"Berguna untuk menyemak tahap stok, lokasi produk, dan pengurusan gudang.","content":"""Anda ialah Pakar Pengurusan Inventori.Anda mempunyai akses kepada jadual '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:""" 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) if not skill: return f"Error: Skill '{skill_name}' not found. Available skills: {list(SKILLS.keys())}" return skill["content"]
@tool def run_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: return db.run(query) except Exception as e: return f"Error executing SQL: {e}"
@tool def list_tables() -> str: """List all available tables in the database.""" 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):
We can add custom state if needed, but MessagesState is sufficient for simple chat
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()# --- Pelaksanaan Utama ---if__name__ =="main": system_prompt ="""Anda adalah Pembantu SQL yang berguna.Anda mempunyai akses kepada kemahiran khusus yang mengandungi skema pangkalan data dan pengetahuan domain.Untuk menjawab soalan pengguna:1. Kenal pasti kemahiran yang berkaitan (analisis_jualan atau pengurusan_inventori).2. Gunakan alat 'load_skill' untuk mendapatkan skema dan arahan.3. Berdasarkan kemahiran yang dimuatkan, tulis dan laksanakan pertanyaan SQL menggunakan 'run_sql_query'.4. Jawab soalan pengguna berdasarkan hasil pertanyaan.Jangan teka nama jadual. Sentiasa muatkan kemahiran terlebih dahulu.""" print("Pembantu SQL dimulakan. Taip 'quit' untuk keluar.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Semakan sambungan pra-panastry: print(f"Bersambung ke pangkalan data:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Amaran sambungan pangkalan data:{e}")whileTrue:try: user_input = input("Pengguna: ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Alirkan pelaksanaan print("Ejen: ", end="", flush=True) final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# Dalam mod 'values', kita mendapat keadaan penuh. Kita hanya mahu melihat mesej terakhir jika ia baharu. last_message = event["messages"][-1]# Kemas kini sejarah mesej kita dengan keadaan terkinipass# Selepas aliran selesai, keadaan terakhir mempunyai jawapan akhir 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"]
Kemas kini sejarah
print("-"*50) except Exception as e: print(f"\nError:{e}") break`
