Na anterior publicação, exploramos como simular o padrão de uso de Skills pelo Deep Agent através do Deep Agents CLI. Atualmente, o LangChain já suporta nativamente essa funcionalidade, simplificando enormemente o processo de desenvolvimento. Este artigo irá guiá-lo através de uma experiência aprofundada desta funcionalidade, construindo um assistente SQL mais inteligente.

Ao construir Agentes de IA complexos, os desenvolvedores frequentemente se encontram em um dilema: injetar todo o contexto (estrutura da tabela do banco de dados, documentação da API, regras de negócios) no System Prompt de uma só vez, resultando em estouro da janela de contexto (Context Window) e dispersão da atenção do modelo? Ou escolher um ajuste fino (Fine-tuning) frequente e dispendioso?

O Padrão de Skills (Skills Pattern) oferece uma rota intermediária elegante. Ele realiza a utilização eficiente do contexto através do carregamento dinâmico do conhecimento necessário. O suporte nativo do LangChain para este padrão significa que podemos construir mais facilmente um Agente com capacidade de "aprender sob demanda".

Este artigo, em conjunto com a documentação oficial Build a SQL assistant with on-demand skills, guiará o leitor desde o início, construindo um Assistente SQL que suporta o "carregamento de conhecimento sob demanda".

1. Conceitos Centrais: Por que escolher o Padrão de Skills?

Limitações do Agente SQL Tradicional

Na arquitetura tradicional do Agente SQL, geralmente precisamos fornecer o Schema completo do Banco de Dados no System Prompt. Com o desenvolvimento dos negócios, quando o número de tabelas se expande para centenas, essa abordagem traz problemas significativos:

• Consumo enorme de Token: Cada diálogo carrega uma grande quantidade de estruturas de tabela irrelevantes, causando desperdício de recursos.

• Aumento do risco de alucinações: Muita informação de interferência irrelevante reduzirá a precisão do raciocínio do modelo.

• Dificuldade de manutenção: Todo o conhecimento das linhas de negócios está intimamente acoplado, tornando difícil a iteração independente.

Padrão de Skills: Uma solução baseada na divulgação progressiva

O padrão de Skills é baseado no princípio da divulgação progressiva (Progressive Disclosure), que processa o processo de aquisição de conhecimento em camadas:

• Estado inicial do Agente: Apenas domina quais são as "habilidades" (Skills) e suas breves descrições (Description), mantendo-se leve.

• Carregamento em tempo de execução: Ao enfrentar um problema específico (como "consultar o estoque"), o Agente chama ativamente a ferramenta (load_skill) para carregar o contexto detalhado da habilidade (Schema + Prompt).

• Execução da tarefa: Com base no contexto preciso carregado, executa a tarefa específica (como escrever e executar SQL).

Este padrão suporta efetivamente a expansão ilimitada e o desacoplamento da equipe, permitindo que o Agente se adapte a cenários de negócios cada vez mais complexos.

2. Projeto da Arquitetura do Sistema

Este projeto prático construirá um Assistente SQL contendo duas Skills centrais para demonstrar a aplicação prática deste padrão:

• Sales Analytics (Análise de Vendas): Responsável pela tabelasales_data, lidando com estatísticas de receita, análise de tendências de pedidos, etc.

• Inventory Management (Gestão de Estoque): Responsável pela tabelainventory_items, lidando com o monitoramento do nível de estoque, consulta de localização, etc.

3. Configuração do Ambiente de Desenvolvimento

Este projeto usa Pythonuvpara gerenciamento eficiente de dependências.

Instalação das Dependências Centrais

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

Configuração do Ambiente PostgreSQL

Inicie uma instância Postgres localmente e crie o banco de dadosagent_platform. Fornecemos o scriptsetup_db.pypara inicializar automaticamente a estrutura da tabela e os dados de teste (veja o código-fonte no final do artigo).

4. Explicação Detalhada das Etapas de Implementação Central### Passo 1: Definindo Habilidades de Domínio (O Conhecimento)

Definiremos as habilidades como uma estrutura de dicionário, simulando o processo de carregamento de um sistema de arquivos ou banco de dados. Observe a distinção entre description (usada pelo Agente para seleção de decisão) e content (o contexto detalhado real carregado).

SKILLS = {"sales_analytics": {"description":"Útil para analisar receita de vendas, tendências...","content":"""... Esquema da Tabela: sales_data ..."" },"inventory_management": {"description":"Útil para verificar os níveis de estoque...","content":"""... Esquema da Tabela: inventory_items ..."" }}

Passo 2: Implementando Ferramentas Essenciais (As Capacidades)

O Agente depende de duas ferramentas cruciais para concluir as tarefas:

• load_skill(skill_name): Carrega dinamicamente os detalhes da habilidade especificada em tempo de execução.

• run_sql_query(query): Executa instruções SQL específicas.

Passo 3: Orquestrando a Lógica do Agente (O Cérebro)

Use LangGraph para construir o ReAct Agent. O System Prompt desempenha um papel fundamental aqui, orientando o Agente a seguir estritamente o Procedimento Operacional Padrão (SOP) de Identify -> Load -> Query.

system_prompt ="""1. Identifique a habilidade relevante.2. Use 'load_skill' para obter o esquema.3. Escreva e execute SQL usando 'run_sql_query'....Não adivinhe os nomes das tabelas. Sempre carregue a habilidade primeiro.""

5. Verificação do Efeito da Execução

Ao executar test_agent.py, testamos consultas em dois domínios diferentes, Vendas e Estoque. A seguir, estão os logs de saída reais do console, mostrando como o Agente carrega habilidades dinamicamente com base nas perguntas:

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. Referência Completa do Código-Fonte

A seguir, o código-fonte completo do projeto, incluindo o script de inicialização do banco de dados e o programa principal do Agent.

1. Inicialização do Banco de Dados (setup_db.py)

`import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT import os from dotenv import load_dotenv

load_dotenv()

Por favor, certifique-se de configurar as informações de conexão do banco de dados em .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") # Por favor, substitua pela senha real DB_NAME = os.getenv("DB_NAME", "agent_platform")

def create_database(): try: # Conecte-se ao banco de dados padrão 'postgres' para criar um novo banco de dados 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()

    # Verifique se o banco de dados existe
    cur.execute(f"SELECT 1 FROM pg_catalog.pg_database WHERE datname = '{DB_NAME}'")
    exists = cur.fetchone()

    if not exists:
        print(f"Criando banco de dados {DB_NAME}...")
        cur.execute(f"CREATE DATABASE {DB_NAME}")
    else:
        print(f"Banco de dados {DB_NAME} já existe.")

    cur.close()
    conn.close()
except Exception as e:
    print(f"Erro ao criar o banco de dados: {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()

    # Criar Tabela de Vendas
    print("Criando tabela sales_data...")
    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)
        )
    """)

    # Criar Tabela de Inventário
    print("Criando tabela inventory_items...")
    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)
        )
    """)

    # Inserir Dados Mock
    print("Inserindo dados mock...")
    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("Configuração do banco de dados concluída.")
except Exception as e:
    print(f"Erro ao configurar as tabelas: {e}")

if name == "main": create_database() create_tables_and_data()`### 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": "Useful for analyzing sales revenue, trends, and regional performance.",
        "content": """You are a Sales Analytics Expert.
You have access to the 'sales_data' table.
Table Schema:
- id: integer (primary key)
- transaction_date: date
- product_id: varchar(50)
- amount: decimal(10, 2)
- region: varchar(50)
Common queries:
- Total revenue: SUM(amount)
- Revenue by region: GROUP BY region
- Sales trend: GROUP BY transaction_date"""
    },
    "inventory_management": {
        "description": "Useful for checking stock levels, product locations, and warehouse management.",
        "content": """You are an Inventory Management Expert.
You have access to the 'inventory_item
```Tabela Schema:
- id: integer (chave primária)
- product_id: varchar(50)
- product_name: varchar(100)
- stock_count: integer
- warehouse_location: varchar(50)
Consultas comuns:
- Verificar estoque: WHERE product_name = '...'
- Estoque baixo: WHERE stock_count < threshold"""  }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:"""
 Carrega o prompt detalhado e o esquema para uma habilidade específica.
 Habilidades disponíveis:
 - sales_analytics: Para análise de vendas, receita e transações.
 - inventory_management: Para consultas de estoque, produtos e armazém.
 """
habilidade = SKILLS.get(skill_name)
ifnot habilidade:
 return f"Erro: Habilidade '{skill_name}' não encontrada. Habilidades disponíveis: {list(SKILLS.keys())}"
return habilidade["content"]
@tool
def run_sql_query(query: str) -> str:
 """
 Executa uma consulta SQL no banco de dados.
 Use esta ferramenta SOMENTE APÓS carregar a habilidade apropriada para entender o esquema.
 """
try:
 return db.run(query)
except Exception as e:
 return f"Erro ao executar SQL: {e}"
@tool
def list_tables() -> str:
 """Lista todas as tabelas disponíveis no banco de dados."""
 return str(db.get_usable_table_names())
ferramentas = [load_skill, run_sql_query, list_tables]
# --- Configuração do Agente ---
llm = ChatOpenAI(
 base_url=BASE_URL,
 api_key=API_KEY,
 model=MODEL_NAME,
 temperature=0
)
llm_with_tools = llm.bind_tools(ferramentas)
# --- Definição do Gráfico ---
class AgentState(MessagesState):
 # Podemos adicionar um estado personalizado, se necessário, mas MessagesState é suficiente para um bate-papo simples
 pass

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

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()# --- Main Execution ---if__name__ =="__main__":  system_prompt ="""Você é um Assistente SQL útil.Você tem acesso a habilidades especializadas que contêm esquemas de banco de dados e conhecimento de domínio.Para responder à pergunta de um usuário:1. Identifique a habilidade relevante (sales_analytics ou inventory_management).2. Use a ferramenta 'load_skill' para obter o esquema e as instruções.3. Com base na habilidade carregada, escreva e execute uma consulta SQL usando 'run_sql_query'.4. Responda à pergunta do usuário com base nos resultados da consulta.Não adivinhe nomes de tabelas. Sempre carregue a habilidade primeiro."""  print("Assistente SQL inicializado. Digite 'quit' para sair.")  print("-"*50)  messages = [SystemMessage(content=system_prompt)]# Verificação de conexão pré-aquecidatry:    print(f"Conectado ao banco de dados:{DB_URI.split('@')[-1]}")exceptExceptionase:    print(f"Aviso de conexão com o banco de dados:{e}")whileTrue:try:      user_input = input("Usuário: ")ifuser_input.lower()in["quit","exit"]:break      messages.append(HumanMessage(content=user_input))# Transmita a execução      print("Agente: ", end="", flush=True)      final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# No modo 'values', obtemos o estado completo. Só queremos ver a última mensagem se for nova.        last_message = event["messages"][-1]# Atualize nosso histórico de mensagens com o estado mais recentepass# Depois que o fluxo termina, o último estado tem a resposta final      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"]
 # Update history
 print("-"*50)
 except Exception as e:
 print(f"\nError:{e}")
 break`