2026年人工智能 - AI圈内卷？天池团聚请来专家集体“问诊”

大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。

自从10月份Claude提出Skills概念后，给智能体开发带来一个全新的视角；今天我们就基于企业一个非常实用的NL2SQL场景出发，结合langchain1.X+Skills实现一个企业级稳定高效的问数智能体；下面我们正式开始。

在数字化转型的今天，企业并不缺数据，缺的是获取数据的效率。

现状：业务人员想看一个维度，需要提需求给 IT，IT 写 SQL，反复确认后再出报表。这个周期通常以“天”为单位。
愿景：NL2SQL 的核心价值在于：打破技术壁垒，让数据“说人话”，实现数据查询的即问即答。

很多企业初试 AI 时，倾向于把所有数据库表结构塞给大模型（LLM），结果往往不尽如人意：

上下文过载（Context Overflow）：企业数据库动辄几百张表，超出模型处理极限，导致回复变慢、成本飙升。
意图稀释（Intent Dilution）：表越多，模型越容易“张冠李戴”，把“销售额”算成“利润额”。
幻觉风险：模型在不确定的情况下会“盲目自信”，编造不存在的字段或计算逻辑。

这就是为什么我们要在 LangChain 1.x 中引入 Skill-based（基于技能） 的模式。这不仅仅是技术实现，更是一种管理哲学。

1. 知识的“分封制”：解决认知负担

通过 Skill 模式，我们将庞大的数据库拆解为一个个“垂直领域专家”（如：销售专家、库存专家）。

意义：Agent 在思考时，只需加载与当前问题相关的 1% 的知识。这像极了前端的按需加载 (Lazy Loading)，既快又准。

2. 确定性的执行：从“猜测”到“指令”

Skill 模式要求 Agent 先通过 get_skill_details 确认“说明书”。

意义：它强制模型先看“规则”再写“代码”，极大地降低了模型凭记忆编造数据（幻觉）的概率。

1. 复杂业务逻辑的沉淀

企业级查询不只是 SELECT *，它包含复杂的计算逻辑（如：毛利如何计算、逾期如何定义）。

价值：我们将这些逻辑固化在 Skill 的 Prompt 中。Skill 成了企业业务知识的数字资产包，不再随人员离职而流失。

2. 安全与权限的微隔离

在企业中，不是所有人都能查薪资。

价值：基于 Skill 的架构天然支持权限挂钩。我们可以根据用户的身份，动态决定哪些“技能”对该用户可见。

3. 系统自我修复的闭环

基于 LangGraph 的状态机架构，如果 SQL 执行失败，Agent 能根据报错信息自动重试。

价值：这减少了人工介入的频率，让系统具备了真正的“自愈”能力。

构建基于 LangChain 1.x 的数据智能体，最终为企业带来的是：

敏捷响应：决策层获取数据的时间从“天”降至“秒”，应对市场变化更快。
人力解放：将 DBA 和数据分析师从重复性的手工写代码中解放出来，去处理更具战略意义的数据治理工作。
数据普惠：让不懂技术的一线业务员也能具备数据驱动的能力，真正实现“人人都是分析师”。

理论知识讲完了，接下来我们通过代码完成从数库创建到智能体构建整个过程；实实在在感觉通过skills这种模型带来的变化；

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, .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 conndb_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.定义工具

@tooldef get_skill_details(skill_name: str): “”“根据技能名称获取详细的技能配置（包含表结构和描述）。”“” print(f” [工具调用] 正在加载技能详情: {skill_name}“) skill = SKILLS_STORE.get(skill_name) return json.dumps(skill, ensure_ascii=False) if skill else “未找到该技能”@tooldef execute_sql(sql: str): “”“执行 SQL 查询并返回 JSON 格式的结果。”“” print(f” [工具调用] 正在执行 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 = ” “.join([f”- {s[‘name’]}: {s[‘description’]}” for s in skills]) def _get_addendum(self) -> str: # 实时获取最新描述，支持热更新 skills = load_skills_from_json() prompt = “ “.join([f”- {s[‘name’]}: {s[‘description’]}” for s in skills]) return ( f”

当前可用专业知识模块

{prompt}

” “注意：如果你需要编写具体的 SQL 或了解业务细节，请务必先使用 load_skill 工具获取详细信息。” ) def wrap_model_call(self, request, handler): addendum = f”

可用技能模块: {self.skills_prompt}

请先 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) + [ ] 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.pyfrom typing import Dict, Callablefrom langchain_core.runnables import Runnablefrom examples.multiAgent.skills_mysql import get_skills_agent# 具备联网搜索的deep agentAGENT_REGISTRY: Dict[str, Callable[[], Runnable]] =

7.效果展示

我这边中通过cherry studio客户端进行显示的，配置如下：

人工智能 - AI圈内卷？天池团聚请来专家集体“问诊”_#架构

配置模型

下面是回复效果：

人工智能 - AI圈内卷？天池团聚请来专家集体“问诊”_#架构_02

人工智能 - AI圈内卷？天池团聚请来专家集体“问诊”_#架构_03

人工智能 - AI圈内卷？天池团聚请来专家集体“问诊”_#gpt_04

在本项目中，Skills 是智能体的专业工具箱。它让 Agent 告别了万金油 式的模糊回答，成为了一个懂业务规矩、看说明书办事、且边界清晰的专业数据官；不仅降低了成本还提升了效率