advanced⏱40 minutesLição 5 de 10

Agentes SQL

Construa agentes de consulta SQL com LangGraph — vinculação de ferramentas de banco de dados, geração texto-para-SQL, validação e execução de consultas e interpretação de resultados.

Agentes SQL

SQL agents let users query databases using natural language. The agent converts the question to SQL, validates it, executes it, and interprets the results. This lesson covers safe and reliable text-to-SQL patterns.

Architecture

Natural Language Query
    ↓
SQL Agent
    ├── 1. Understand schema → Describe tables and columns
    ├── 2. Generate SQL → LLM converts question to SQL
    ├── 3. Validate SQL → Syntax check, safety checks
    ├── 4. Execute SQL → Run against database
    └── 5. Interpret results → LLM explains the results
    ↓
Natural Language Answer

Conexão com Banco de Dados

python

from sqlalchemy import create_engine, text, inspect
from sqlalchemy.engine import Engine
from langchain_core.tools import tool

# Database connection
DATABASE_URL = "postgresql://user:pass@localhost:5432/mydb"
engine = create_engine(DATABASE_URL)

[!AVISO] Never expose your database directly to users. Use a read-only database user and whitelist allowed queries. Always validate SQL before execution.

Ferramenta de Descoberta de Schema

python

@tool
def get_schema() -> str:
    """Get the database schema (tables, columns, types, and relationships)."""
    inspector = inspect(engine)
    schema = []

    for table_name in inspector.get_table_names():
        columns = inspector.get_columns(table_name)
        cols = [f"  - {c['name']}: {c['type']}" for c in columns]
        schema.append(f"Table: {table_name}\n" + "\n".join(cols))

        # Foreign keys
        fks = inspector.get_foreign_keys(table_name)
        for fk in fks:
            schema.append(f"  FK: {fk['constrained_columns']} → {fk['referred_table']}.{fk['referred_columns']}")

    return "\n\n".join(schema)

Geração Texto-para-SQL

python

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o")

sql_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a SQL expert. Convert natural language questions to SQL queries.\n"
               "Database schema:\n{schema}\n\n"
               "Rules:\n"
               "1. Use only SELECT statements (no INSERT/UPDATE/DELETE)\n"
               "2. Use LIMIT to restrict results (max 100 rows)\n"
               "3. Use proper JOIN syntax\n"
               "4. Return ONLY the SQL query, no explanations"),
    ("human", "{question}")
])

def generate_sql(question: str, schema: str) -> str:
    chain = sql_prompt | llm | StrOutputParser()
    sql = chain.invoke({"question": question, "schema": schema})
    # Clean the SQL (remove markdown fences if present)
    sql = sql.replace("```sql", "").replace("```", "").strip()
    return sql

Validação SQL

python

import re
from typing import Tuple

FORBIDDEN_PATTERNS = [
    r'\bINSERT\b', r'\bUPDATE\b', r'\bDELETE\b',
    r'\bDROP\b', r'\bALTER\b', r'\bCREATE\b',
    r'\bTRUNCATE\b', r'\bGRANT\b', r'\bREVOKE\b',
    r'\bEXEC\b', r'\bEXECUTE\b', r';.*;'  # Multiple statements
]

def validate_sql(sql: str) -> Tuple[bool, str]:
    """Validate SQL for safety and syntax."""

    # Check for forbidden operations
    for pattern in FORBIDDEN_PATTERNS:
        if re.search(pattern, sql, re.IGNORECASE):
            return False, f"Forbidden SQL operation detected: {pattern.strip('\\\\b')}"

    # Must be a SELECT statement
    if not sql.strip().upper().startswith("SELECT"):
        return False, "Only SELECT queries are allowed"

    # Syntax check (try to explain the query)
    try:
        with engine.connect() as conn:
            conn.execute(text(f"EXPLAIN {sql}"))
    except Exception as e:
        return False, f"SQL syntax error: {str(e)}"

    return True, "Valid"

[!NOTA] The EXPLAIN command tests syntax without executing the query, making it a safe validation step before running the actual query.

Ferramenta de Execução SQL

python

@tool
def execute_sql(sql: str) -> str:
    """Execute a SQL SELECT query and return results as formatted text."""
    # Validate first
    valid, msg = validate_sql(sql)
    if not valid:
        return f"Validation error: {msg}"

    # Execute
    try:
        with engine.connect() as conn:
            result = conn.execute(text(sql))
            rows = result.fetchmany(100)
            columns = result.keys()

            # Format as text table
            output = []
            output.append(" | ".join(columns))
            output.append("-" * len(output[0]))

            for row in rows:
                output.append(" | ".join(str(val)[:50] for val in row))

            return "\n".join(output) if len(output) > 2 else "No results found."
    except Exception as e:
        return f"Execution error: {str(e)}"

Agente SQL Completo

python

from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolExecutor
from langchain_core.messages import SystemMessage, ToolMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any

tools = [get_schema, execute_sql]
llm_with_tools = llm.bind_tools(tools)
tool_executor = ToolExecutor(tools)

class SQLState(TypedDict):
    messages: Annotated[List[Any], add_messages]
    schema: str

def agent_node(state: SQLState) -> dict:
    system = """You are a SQL assistant. Follow these steps:
1. Use get_schema to understand the database structure
2. Generate a SQL query for the user's question
3. Use execute_sql to run it
4. Interpret the results for the user in plain English

Rules:
- Only use SELECT queries
- Explain results in natural language
- If a query returns an error, fix it and retry"""

    messages = [SystemMessage(system)] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def tools_node(state: SQLState) -> dict:
    last = state["messages"][-1]
    if not last.tool_calls:
        return {}
    return {"messages": [
        ToolMessage(content=str(tool_executor.invoke(tc)), tool_call_id=tc["id"])
        for tc in last.tool_calls
    ]}

def router(state: SQLState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "end"

builder = StateGraph(SQLState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tools_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", router, {"tools": "tools", "end": END})
builder.add_edge("tools", "agent")
app = builder.compile(checkpointer=MemorySaver())

# Example usage
result = app.invoke({
    "messages": [("human", "Show me the top 5 customers by total orders")],
    "schema": ""
})
print(result["messages"][-1].content)

[!SUCESSO] The agent autonomously discovers the schema, generates SQL, validates and executes it, then explains the results — all in a loop until the user's question is answered.

Conversas SQL de Múltiplos Turnos

python

# Follow-up questions maintain context
session = {"configurable": {"thread_id": "sql-session-1"}}

app.invoke({"messages": [("human", "What tables exist in the database?")]}, session)
# Agent: Uses get_schema, describes the tables

app.invoke({"messages": [("human", "How many customers are from New York?")]}, session)
# Agent: Already knows the schema, generates and executes the query

app.invoke({"messages": [("human", "Show me the details")]}, session)
# Agent: Understands context from conversation history

Recuperação de Erros

python

def resilient_agent(state: SQLState) -> dict:
    """Agent with error recovery — retries with fixed SQL on error."""
    last = state["messages"][-1]
    error_attempts = sum(1 for m in state["messages"]
                        if hasattr(m, 'content') and "Execution error:" in m.content)

    if error_attempts > 3:
        return {"messages": [AIMessage(
            "I'm having trouble with this query. Let me try a different approach."
        )]}

    # Continue with normal agent logic
    return agent_node(state)

[!DICA] Track error attempts in state. After N failures, try a different approach (simpler query, ask clarifying question) instead of retrying indefinitely.

Perguntas Práticas

Practice Question

What is the first step a SQL agent should take when processing a query?

Practice Question

What SQL operations should be forbidden in a SQL agent?

Practice Question

How can you validate SQL syntax without executing the query?

Practice Question

Why should the SQL agent use a read-only database user?

Practice Question

What does the get_schema tool return?

Practice Question

How should a SQL agent handle a query that returns an error?

Practice Question

What is a good maximum number of retry attempts for a SQL agent?

Practice Question

How does the agent in the SQL example decide when to stop executing tools?

Practice Question

What formatting does the execute_sql tool use for results?

Practice Question

Why is the schema discovery important for text-to-SQL generation?

[!SUCESSO]

Principais Conclusões

SQL agents follow: schema discovery → SQL generation → validation → execution → interpretation
Use a read-only database user and validate SQL before execution
EXPLAIN provides safe syntax validation without running the query
Error recovery: analyze the error, fix the SQL, and retry (up to a limit)
Multi-turn conversations maintain context about the database and prior queries
Format results as text tables for LLM interpretation
Proper schema discovery is essential for accurate text-to-SQL

Progresso50%

RAG Agentic

Padrão de Agente Reflexivo

Contéudo do Curso

1. Implantação 2. Observabilidade com LangSmith 3. Agentes Hierárquicos 4. RAG Agentic 5. Agentes SQL 6. Padrão de Agente Reflexivo 7. Padrão Planejar e Executar 8. Execução Paralela 9. Otimização 10. Projeto Final: Agente de Suporte ao Cliente de Produção