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 bastante o processo de desenvolvimento. Este artigo irá guiá-lo através de uma experiência aprofundada dessa funcionalidade, construindo um assistente SQL mais inteligente.

Ao construir um Agente de IA complexo, 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 caro?

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 suporte o "carregamento de conhecimento sob demanda".

1. Conceitos Essenciais: 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 dessa habilidade (Schema + Prompt).
Execução da tarefa: Com base no contexto preciso carregado, executa tarefas específicas (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 principais 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 monitoramento do nível de estoque, consulta de localização, etc.

3. Configuração do Ambiente de Desenvolvimento

Este projeto usa Python uv para gerenciamento eficiente de dependências.

Instalação de dependências principais

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 dados agent_platform. Fornecemos o script setup_db.py para 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 principais da implementação### Passo 1: Definir 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 tomada 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: Implementar 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: Orquestrar a Lógica do Agente (O Cérebro)

Use LangGraph para construir o Agente ReAct. O System Prompt desempenha um papel fundamental aqui, orientando o Agente a seguir estritamente o Procedimento Operacional Padrão (POP) Identificar -> Carregar -> Consultar.

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 na pergunta:

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 Agente.

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 de Simulação
    print("Inserindo dados de simulação...")
    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""" Table Schema:
id: integer (primary key)
product_id: varchar(50)
product_name: varchar(100)
stock_count: integer
warehouse_location: varchar(50)

Common queries:

Check stock: WHERE product_name = '...'
Low stock: WHERE stock_count < threshold""" }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:""" Load the detailed prompt and schema for a specific skill. Available skills:
sales_analytics: For sales, revenue, and transaction analysis.
inventory_management: For stock, products, and warehouse queries. """ skill = SKILLS.get(skill_name) if not skill: return f"Error: Skill '{skill_name}' not found. Available skills: {list(SKILLS.keys())}" return skill["content"]

@tool def run_sql_query(query: str) -> str: """ Execute a SQL query against the database. Only use this tool AFTER loading the appropriate skill to understand the schema. """ try: return db.run(query) except Exception as e: return f"Error executing SQL: {e}"

@tool def list_tables() -> str: """List all available tables in the database.""" return str(db.get_usable_table_names())

tools = [load_skill, run_sql_query, list_tables]

--- Agent Setup ---

llm = ChatOpenAI( base_url=BASE_URL, api_key=API_KEY, model=MODEL_NAME, temperature=0 ) llm_with_tools = llm.bind_tools(tools)

--- Graph Definition ---

class AgentState(MessagesState):

We can add custom state if needed, but MessagesState is sufficient for simple chat

pass

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

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 do 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 os nomes das 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 de pré-aquecimentotry: print(f"Conectado ao banco de dados:{DB_URI.split('@')[-1]}")exceptExceptionase: print(f"Aviso de conexão do 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# Após o término do fluxo, 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"]