Agentes SQL

Construye agentes de consulta SQL con LangGraph, vinculación de herramientas de base de datos, generación texto-a-SQL, validación y ejecución de consultas e interpretación de resultados.

Los agentes SQL permiten a los usuarios consultar bases de datos en lenguaje natural. El agente convierte la pregunta a SQL, la valida, la ejecuta e interpreta los resultados. Esta lección cubre patrones de text-to-SQL seguros y fiables.

Arquitectura

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

Conexión con Base de Datos

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)

⚠️Advertencia

Nunca expongas tu base de datos directamente a los usuarios. Usa un usuario de base de datos de solo lectura y una lista blanca de consultas permitidas. Valida siempre el SQL antes de ejecutarlo.

Herramienta de Descubrimiento de Esquema

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)

Generación Texto-a-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

Validación 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

El comando EXPLAIN prueba la sintaxis sin ejecutar la consulta, lo que lo convierte en un paso de validación seguro antes de ejecutar la consulta real.

Herramienta de Ejecución 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)

✅Éxito

El agente descubre el esquema de forma autónoma, genera el SQL, lo valida y lo ejecuta, y luego explica los resultados, todo en un bucle hasta que se responda la pregunta del usuario.

Conversaciones SQL de Múltiples 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

Recuperación de Errores

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)

💡Consejo

Lleva la cuenta de los intentos fallidos en el estado. Tras N fallos, prueba un enfoque distinto (consulta más simple, haz una pregunta de aclaración) en vez de reintentar indefinidamente.