Skills 架构实现企业级 NL2SQL 智能体开发实战

代码实践

1. 数据库初始化

为演示方便，我们的使用的数据库为 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()

2. 外挂 Skills 并动态加载

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": "仓库名称"},
                    {"name": "stock_quantity", "type": "INTEGER", "description": "库存数量"}
                ]
            }
        ],
        "sample_queries": [
            "SELECT SUM(stock_quantity) FROM inventory WHERE product_id = 1;"
        ]
    }
}

3. 定义工具

@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)})

4. 通过中间件实现提示词注入

这是核心逻辑，也是 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 def awrap_model_call(
        self, request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
    ) -> ModelResponse:
        """异步实现：解决 LangGraph API/Server 的报错"""
        new_content = list(request.system_message.content_blocks) + [
            {"type": "text", "text": self._get_addendum()}
        ]
        new_system_message = SystemMessage(content=new_content)
        modified_request = request.override(system_message=new_system_message)
        # 注意这里需要 await 执行 handler
        return await handler(modified_request)

5. 智能体创建

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

6. 注册智能体

在 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 告别了万金油 式的模糊回答，成为了一个懂业务规矩、看说明书办事、且边界清晰的专业数据官；不仅降低了成本还提升了效率。