En el artículo anterior, exploramos cómo simular el patrón de uso de Skills por parte de Deep Agent a través de Deep Agents CLI. Hoy en día, LangChain ya admite esta característica de forma nativa, lo que simplifica enormemente el proceso de desarrollo. Este artículo lo guiará a través de esta función, construyendo un asistente de SQL más inteligente.

Al construir agentes de IA complejos, los desarrolladores a menudo se enfrentan a un dilema: ¿inyectar todo el contexto (estructura de la tabla de la base de datos, documentación de la API, reglas de negocio) en el System Prompt de una sola vez, lo que provoca el desbordamiento de la ventana de contexto (Context Window) y la dispersión de la atención del modelo? ¿O elegir un ajuste fino (Fine-tuning) frecuente y costoso?

El Modo Skills (Skills Pattern) proporciona una ruta intermedia elegante. Al cargar dinámicamente el conocimiento requerido, se logra una utilización eficiente del contexto. El soporte nativo de LangChain para este patrón significa que podemos construir más fácilmente agentes con la capacidad de "aprender bajo demanda".

Este artículo, en combinación con la documentación oficial Build a SQL assistant with on-demand skills, guiará a los lectores desde cero para construir un asistente de SQL que admita la "carga de conocimiento bajo demanda".

1. Conceptos clave: ¿Por qué elegir el modo Skills?

Limitaciones del agente SQL tradicional

En la arquitectura tradicional del agente SQL, normalmente necesitamos proporcionar el esquema completo de la base de datos en el System Prompt. A medida que el negocio se desarrolla, cuando el número de tablas se expande a cientos, este método traerá problemas significativos:

• Gran consumo de tokens: cada diálogo lleva una gran cantidad de estructuras de tabla irrelevantes, lo que provoca un desperdicio de recursos.

• Mayor riesgo de alucinaciones: demasiada información de interferencia irrelevante reducirá la precisión del razonamiento del modelo.

• Dificultad de mantenimiento: el conocimiento de todas las líneas de negocio está estrechamente acoplado, lo que dificulta la iteración independiente.

Modo Skills: una solución basada en la divulgación progresiva

El modo Skills se basa en el principio de divulgación progresiva (Progressive Disclosure), que procesa el proceso de adquisición de conocimiento en capas:

• Estado inicial del agente: solo domina qué "habilidades" (Skills) y sus breves descripciones (Description), manteniéndose ligero.

• Carga en tiempo de ejecución: cuando se enfrenta a un problema específico (como "consultar el inventario"), el agente llama activamente a la herramienta (load_skill) para cargar el contexto detallado de la habilidad (Schema + Prompt).

• Ejecutar tareas: basado en el contexto preciso cargado, ejecutar tareas específicas (como escribir y ejecutar SQL).

Este modo admite eficazmente la expansión ilimitada y el desacoplamiento del equipo, lo que permite al agente adaptarse a escenarios de negocio cada vez más complejos.

2. Diseño de la arquitectura del sistema

Este proyecto práctico construirá un asistente de SQL que contiene dos Skills centrales para demostrar la aplicación práctica de este modo:

• Sales Analytics (Análisis de ventas): responsable de la tablasales_data, que maneja estadísticas de ingresos, análisis de tendencias de pedidos, etc.

• Inventory Management (Gestión de inventario): responsable de la tablainventory_items, que maneja el monitoreo del nivel de inventario, la consulta de ubicación, etc.

3. Configuración del entorno de desarrollo

Este proyecto utiliza Python uv para una gestión de dependencias eficiente.

Instalación de dependencias centrales

uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community

Configuración del entorno PostgreSQL

Inicie una instancia de Postgres localmente y cree la base de datos agent_platform. Proporcionamos el script setup_db.py para inicializar automáticamente la estructura de la tabla y los datos de prueba (ver el código fuente al final del artículo).

4. Explicación detallada de los pasos de implementación principales### Paso 1: Definir Habilidades de Dominio (El Conocimiento)

Definiremos las habilidades como una estructura de diccionario, simulando el proceso de carga desde un sistema de archivos o base de datos. Tenga en cuenta la distinción entre description (utilizada por el Agente para la selección de opciones) y content (el contexto detallado real cargado).

SKILLS = {"sales_analytics": {"description":"Útil para analizar ingresos por ventas, tendencias...","content":"""... Esquema de la tabla: sales_data ..."" },"inventory_management": {"description":"Útil para verificar los niveles de stock...","content":"""... Esquema de la tabla: inventory_items ..."" }}

Paso 2: Implementar Herramientas Centrales (Las Capacidades)

El Agente depende de dos herramientas clave para completar las tareas:

• load_skill(skill_name): Carga dinámicamente los detalles de la habilidad especificada en tiempo de ejecución.

• run_sql_query(query): Ejecuta la instrucción SQL específica.

Paso 3: Orquestar la Lógica del Agente (El Cerebro)

Utilice LangGraph para construir el Agente ReAct. El System Prompt juega un papel crucial aquí, guiando al Agente para que siga estrictamente el procedimiento operativo estándar (SOP) de Identify -> Load -> Query.

system_prompt ="""1. Identify the relevant skill.2. Use 'load_skill' to get schema.3. Write and execute SQL using 'run_sql_query'....Do not guess table names. Always load the skill first."""

5. Verificación del Efecto de Ejecución

Al ejecutar test_agent.py, probamos consultas en dos dominios diferentes: Ventas e Inventario. Los siguientes son los registros de salida reales de la consola, que muestran cómo el Agente carga dinámicamente las habilidades según la pregunta:

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. Referencia del código fuente completo\n\nA continuación, se muestra el código fuente completo del proyecto, que incluye el script de inicialización de la base de datos y el programa principal del Agente.\n\n### 1. Inicialización de la base de datos (setup_db.py)\n\n`importpsycopg2frompsycopg2.extensionsimportISOLATION_LEVEL_AUTOCOMMITimportosfromdotenvimportload_dotenvload_dotenv()# 请确保在 .env 中配置数据库连接信息DB_HOST = os.getenv(### 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": "Útil para analizar los ingresos por ventas, las tendencias y el rendimiento regional.", "content": """ Eres un Experto en Análisis de Ventas. Tienes acceso a la tabla 'sales_data'. Esquema de la tabla:

• id: integer (clave primaria)
• transaction_date: date
• product_id: varchar(50)
• amount: decimal(10, 2)
• region: varchar(50) Consultas comunes:
• Ingresos totales: SUM(amount)
• Ingresos por región: GROUP BY region
• Tendencia de ventas: GROUP BY transaction_date""" }, "inventory_management": { "description": "Útil para verificar los niveles de stock, las ubicaciones de los productos y la gestión del almacén.", "content": """ Eres un Experto en Gestión de Inventario. Tienes acceso a la tabla 'inventory_items' table.Table Schema:
• id: integer (clave primaria)
• product_id: varchar(50)
• product_name: varchar(100)
• stock_count: integer
• warehouse_location: varchar(50) Consultas comunes:
• Verificar stock: WHERE product_name = '...'
• Stock bajo: WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:""" Cargar el prompt detallado y el esquema para una habilidad específica. Habilidades disponibles:
• sales_analytics: Para análisis de ventas, ingresos y transacciones.
• inventory_management: Para consultas de stock, productos y almacén. """ habilidad = SKILLS.get(skill_name) sinohabilidad: return f"Error: La habilidad '{skill_name}' no se encontró. Habilidades disponibles: {list(SKILLS.keys())}" return habilidad["content"] @tool def run_sql_query(query: str) -> str: """ Ejecutar una consulta SQL contra la base de datos. Solo usar esta herramienta DESPUÉS de cargar la habilidad apropiada para entender el esquema. """ try: return db.run(query) except Exception as e: return f"Error al ejecutar SQL: {e}" @tool def list_tables() -> str: """Listar todas las tablas disponibles en la base de datos.""" return str(db.get_usable_table_names())

herramientas = [load_skill, run_sql_query, list_tables]

--- Configuración del Agente ---

llm = ChatOpenAI( base_url=BASE_URL, api_key=API_KEY, model=MODEL_NAME, temperature=0 )

llm_with_tools = llm.bind_tools(herramientas)

--- Definición del Gráfico ---

class AgentState(MessagesState):

Podemos agregar un estado personalizado si es necesario, pero MessagesState es suficiente para un chat simple

pass

def agent_node(state: AgentState): mensajes = state["messages"] respuesta = llm_with_tools.invoke(mensajes) return {"messages": [respuesta]}

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()# --- Ejecución Principal ---if__name__ =="main": system_prompt ="""Eres un Asistente SQL útil.Tienes acceso a habilidades especializadas que contienen esquemas de bases de datos y conocimiento del dominio.Para responder a la pregunta de un usuario:1. Identifica la habilidad relevante (sales_analytics o inventory_management).2. Usa la herramienta 'load_skill' para obtener el esquema y las instrucciones.3. Basándote en la habilidad cargada, escribe y ejecuta una consulta SQL usando 'run_sql_query'.4. Responde a la pregunta del usuario basándote en los resultados de la consulta.No adivines los nombres de las tablas. Siempre carga la habilidad primero.""" print("Asistente SQL inicializado. Escribe 'quit' para salir.") print("-"*50) messages = [SystemMessage(content=system_prompt)]# Verificación de conexión previa al inicio (pre-warm)try: print(f"Conectado a la base de datos:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Advertencia de conexión a la base de datos:{e}")whileTrue:try: user_input = input("Usuario: ")ifuser_input.lower()in["quit","exit"]:break messages.append(HumanMessage(content=user_input))# Transmite la ejecución print("Agente: ", end="", flush=True) final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# En el modo 'values', obtenemos el estado completo. Solo queremos ver el último mensaje si es nuevo. last_message = event["messages"][-1]# Actualiza nuestro historial de mensajes con el último estadopass# Después de que finaliza la transmisión, el último estado tiene la respuesta final final_state = app.invoke({"messages": messages}) last_msg = final_state["messages"][-1] ifisinstance(last_msg, AIMessage): print(last_msg.content) messages = final_state["messages"]# Actualizar historial print("-"*50) exceptExceptionase: print(f"\nError:{e}") break`