PythonAI
Skills 架构实现企业级 NL2SQL 智能体开发实战
介绍基于 LangChain 1.x 和 Skills 架构的企业级 NL2SQL 智能体方案。针对传统 NL2SQL 上下文过载、意图稀释及幻觉风险等问题，提出将数据库拆解为垂直领域专家的技能模式。通过中间件注入提示词、动态加载技能配置及权限隔离，实现数据查询的即问即答与系统自愈。该方案降低了开发成本，提升了响应效率，使业务人员能直接获取数据支持决策。
介绍基于 LangChain 1.x 和 Skills 架构的企业级 NL2SQL 智能体方案。针对传统 NL2SQL 上下文过载、意图稀释及幻觉风险等问题，提出将数据库拆解为垂直领域专家的技能模式。通过中间件注入提示词、动态加载技能配置及权限隔离，实现数据查询的即问即答与系统自愈。该方案降低了开发成本，提升了响应效率，使业务人员能直接获取数据支持决策。
自从 10 月份 Claude 提出 Skills 概念后，给智能体开发带来一个全新的视角；今天我们就基于企业一个非常实用的 NL2SQL 场景出发，结合 langchain1.X+Skills 实现一个企业级稳定高效的问数智能体；下面我们正式开始。
在数字化转型的今天，企业并不缺数据，缺的是获取数据的效率。
很多企业初试 AI 时，倾向于把所有数据库表结构塞给大模型（LLM），结果往往不尽如人意：
这就是为什么我们要在 LangChain 1.x 中引入 Skill-based（基于技能） 的模式。这不仅仅是技术实现，更是一种管理哲学。
通过 Skill 模式，我们将庞大的数据库拆解为一个个'垂直领域专家'（如：销售专家、库存专家）。
Skill 模式要求 Agent 先通过 get_skill_details 确认'说明书'。
企业级查询不只是 SELECT *，它包含复杂的计算逻辑（如：毛利如何计算、逾期如何定义）。
在企业中，不是所有人都能查薪资。
基于 LangGraph 的状态机架构，如果 SQL 执行失败，Agent 能根据报错信息自动重试。
构建基于 LangChain 1.x 的数据智能体，最终为企业带来的是：
为演示方便，我们的使用的数据库为 sqlite
# --- 1. 数据库初始化 (增加库存相关的表) ---
def setup_db():
conn = sqlite3.connect(":memory:", check_same_thread=False)
cursor = conn.cursor()
# 销售相关表
cursor.execute("CREATE TABLE products (product_id INT, product_name TEXT)")
cursor.execute("CREATE TABLE sales (sale_id INT, product_id INT, amount REAL)")
cursor.execute("INSERT INTO products VALUES (1, '超级智能手表 Pro'), (2, 'AI 音箱')")
cursor.execute("INSERT INTO sales VALUES (1, 1, 2857340.5), (2, 2, 500.0)")
# --- 新增：库存相关表 ---
cursor.execute("CREATE TABLE inventory (product_id INT, warehouse_name TEXT, stock_quantity INT)")
cursor.execute("INSERT INTO inventory VALUES (1, '北京一号仓', 150), (1, '上海二号仓', 300), (2, '深圳分发中心', 1000)")
conn.commit()
return conn
db_conn = setup_db()
SKILLS_STORE = {
"sales_skill": {
"name": "sales_skill",
"description": "销售数据查询技能，用于执行销售相关的 SQL 查询",
"tables": [
{"name": "products", "columns": [{"name": "product_id", "type": "INTEGER"}, {"name": "product_name", "type": "TEXT"}]},
{"name": "sales", "columns": [{"name": "sale_id", "type": "INTEGER"}, {"name": "amount", "type": "REAL"}]}
]
},
# --- 新增加的技能 ---
"inventory_skill": {
"name": "inventory_skill",
"description": "库存管理技能，用于查询各仓库的产品库存余量",
"tables": [
{
"name": "inventory",
"columns": [
{"name": "product_id", "type": "INTEGER", "description": "产品 ID"},
{"name": "warehouse_name", "type": "TEXT", "description": "仓库名称"},
{: , : , : }
]
}
],
: [
]
}
}
@tool
def get_skill_details(skill_name: str):
"""根据技能名称获取详细的技能配置（包含表结构和描述）。"""
print(f"\n[工具调用] 正在加载技能详情：{skill_name}")
skill = SKILLS_STORE.get(skill_name)
return json.dumps(skill, ensure_ascii=False) if skill else "未找到该技能"
@tool
def execute_sql(sql: str):
"""执行 SQL 查询并返回 JSON 格式的结果。"""
print(f"\n[工具调用] 正在执行 SQL: {sql}")
try:
cursor = db_conn.cursor()
cursor.execute(sql)
columns = [column[0] for column in cursor.description]
results = [dict(zip(columns, row)) for row in cursor.fetchall()]
return json.dumps({"result": results, "rows": len(results), "columns": columns}, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": str(e)})
这是核心逻辑，也是 langchain1.0 更新的最大亮点，通过中间件可以智能体执行过程中的黑盒变的可扩控；
class SkillMiddleware(AgentMiddleware):
def __init__(self):
# 修正点：初始化时直接调用函数获取数据，不再引用已删除的 SKILLS 变量
skills = load_skills_from_json()
print(f"✅ 获取技能列表成功：{skills}")
self.skills_prompt = "\n".join([f"- {s['name']}: {s['description']}" for s in skills])
def _get_addendum(self) -> str:
# 实时获取最新描述，支持热更新
skills = load_skills_from_json()
prompt = "\n".join([f"- {s['name']}: {s['description']}" for s in skills])
return (
f"\n\n## 当前可用专业知识模块\n\n{prompt}\n\n"
"注意：如果你需要编写具体的 SQL 或了解业务细节，请务必先使用 `load_skill` 工具获取详细信息。"
)
def wrap_model_call(self, request, handler):
addendum = f"\n\n可用技能模块:\n{self.skills_prompt}\n\n请先 load_skill 再 execute_sql。"
new_content = list(request.system_message.content_blocks) + [{"type": "text", "text": addendum}]
modified_request = request.override(system_message=SystemMessage(content=new_content))
return handler(modified_request)
async () -> ModelResponse:
new_content = (request.system_message.content_blocks) + [
{: , : ._get_addendum()}
]
new_system_message = SystemMessage(content=new_content)
modified_request = request.override(system_message=new_system_message)
handler(modified_request)
model = get_llm()
agent = create_agent(
model,
system_prompt="你是一个能够直接操作数据库的业务专家。请根据用户问题，先加载知识，再执行 SQL，并最后总结答案。",
tools=[load_skill, execute_sql],
# 赋予两种能力
middleware=[SkillMiddleware()],
# checkpointer=InMemorySaver(),
debug=True
)
def get_skills_agent():
return agent
在 agents/registry.py 将我们刚刚开发的智能体注册进来：
# agents/registry.py
from typing import Dict, Callable
from langchain_core.runnables import Runnable
from examples.multiAgent.skills_mysql import get_skills_agent
# 具备联网搜索的 deep agent
AGENT_REGISTRY: Dict[str, Callable[[], Runnable]] = {
"skills_agent": get_skills_agent
}
在本项目中，Skills 是智能体的专业工具箱。它让 Agent 告别了万金油 式的模糊回答，成为了一个懂业务规矩、看说明书办事、且边界清晰的专业数据官；不仅降低了成本还提升了效率。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online