advanced⏱40 minutesLesson 5 of 10

SQL Agents

Build SQL query agents with LangGraph — database tool binding, text-to-SQL generation, query validation and execution, and result interpretation.

SQL Agents

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

Database Connection

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)

⚠️Warning

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

Schema Discovery Tool

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)

Text-to-SQL Generation

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

SQL Validation

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"

ℹ️Note

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

SQL Execution Tool

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)}"

Complete SQL Agent

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)

✅Success

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.

Multi-Turn SQL Conversations

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

Error Recovery

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)

💡Tip

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

Practice Questions

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?

✅Success

Key Takeaways

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

Progress50%

Agentic RAG

Reflection Agent Pattern

Course Content

1. Deployment 2. Observability with LangSmith 3. Hierarchical Agents 4. Agentic RAG 5. SQL Agents 6. Reflection Agent Pattern 7. Plan and Execute Pattern 8. Parallel Execution 9. Optimization 10. Capstone: Production-Grade Customer Support Agent