Building Conversational SQL Agents with LangChain

Written by

jigneshmayani90@gmail.com

Post Date

September 10, 2025

comments

No Comments

Building Conversational SQL Agents with LangChain

Written by

Debakshi B.

Post Date

Sep 10, 2025

556 Views

Ever wanted to build a chatbot that can answer specialized questions directly from your database? We had this exact requirement in one of our projects.

Initially, we used standard LangChain with custom tool calling — creating separate functions for each database table and manually crafting SQL queries. The result? High latency, an unmaintainable codebase that grew exponentially with our database, and memory management nightmares.

The Problems We Faced:

Codebase exploded as new tables were added
Manual SQL generation slowed everything down and introduced errors
Supporting new queries required heavy dev effort
Adding new features on top of the chatbot became difficult
Implementing context management (bot memory) was tedious and error-prone

Then we discovered LangChain’s SQL agent, and how PostgreSQL and LangChain can take care of most of our needs readily. In this tutorial, we’ll show you how to build the solution that reduced our codebase complexity by 70% while dramatically improving performance.

What We’ll Build

A complete conversational SQL agent using LangChain, OpenAI, and FastAPI that can handle queries from a book database, like:

The system maintains conversation context, so follow-up questions work naturally — all without the maintenance headaches of our previous approach.

You can find all the code for this guide on our GitHub page: LangChain SQL Agent Demo (GitHub)

Prerequisites

Setup

First, install the required packages:

pip install fastapi uvicorn langchain-openai langchain-community sqlalchemy psycopg2-binary langchain-postgres

For our example, let’s assume we have a simple bookstore database with two tables:

-- Authors table
CREATE TABLE authors (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    birth_year INTEGER,
    nationality VARCHAR(100)
);

-- Books table
CREATE TABLE books (
    id SERIAL PRIMARY KEY,
    title VARCHAR(255) NOT NULL,
    author_id INTEGER REFERENCES authors(id),
    genre VARCHAR(100),
    publication_year INTEGER,
    rating DECIMAL(3,2)
);

Once you create these tables and insert some sample data, it helps to also define a view called books_with_authors. This view joins both tables into one unified dataset so that queries can be simplified. Instead of writing complex SQL joins every time, the agent can query the view directly to get books along with their authors.

CREATE VIEW books_with_authors AS
SELECT
    b.id AS book_id,
    b.title,
    b.genre,
    b.publication_year,
    b.rating,
    a.name AS author_name,
    a.birth_year,
    a.nationality
FROM books b
JOIN authors a ON b.author_id = a.id;

Step 1: Basic SQL Agent

Now that the database is ready, we’ll set up a SQL agent powered by LangChain and OpenAI.

import os
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent, SQLDatabaseToolkit
from sqlalchemy import create_engine

# Setup
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
DB_URI = "postgresql+psycopg2://username:password@localhost:5432/bookstore"

# Create database connection
engine = create_engine(DB_URI)

# Define custom table info for better LLM context
custom_table_info = {
    "authors": (
        "A table of authors.\n"
        "- id (SERIAL PRIMARY KEY): Unique ID of author\n"
        "- name (VARCHAR): Name of the author\n"
        "- birth_year (INTEGER): Year of birth\n"
        "- nationality (VARCHAR): Nationality of the author\n"
    ),
    "books": (
        "A table of books.\n"
        "- id (SERIAL PRIMARY KEY): Unique ID of book\n"
        "- title (VARCHAR): Title of the book\n"
        "- author_id (INTEGER): References authors(id)\n"
        "- genre (VARCHAR): Genre of the book\n"
        "- publication_year (INTEGER): Year of publication\n"
        "- rating (DECIMAL): Book rating (0–5)\n"
    ),
    "books_with_authors": (
        "A view combining books and authors.\n"
        "- book_id (INTEGER): ID of the book\n"
        "- title (VARCHAR): Title of the book\n"
        "- genre (VARCHAR): Genre of the book\n"
        "- publication_year (INTEGER): Year of publication\n"
        "- rating (DECIMAL): Rating of the book\n"
        "- author_name (VARCHAR): Name of the author\n"
        "- birth_year (INTEGER): Birth year of the author\n"
        "- nationality (VARCHAR): Nationality of the author\n"
    ),
}

# Initialize SQLDatabase with view support and custom info
db = SQLDatabase(
    engine=engine,
    include_tables=list(custom_table_info.keys()),
    custom_table_info=custom_table_info,
    view_support=True
)

# Initialize LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# Create toolkit and agent
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
agent = create_sql_agent(
    toolkit=toolkit,
    llm=llm,
    agent_type="tool-calling",
    verbose=True
)

# Test it out
response = agent.invoke({"input": "List all books with their authors and ratings"})
print(response["output"])

Here’s what happens under the hood:

Natural Language to SQL — The agent takes a plain English question like “List all books with their authors and ratings”.
SQL Generation — It automatically generates the appropriate SQL query (for example, selecting from the books_with_authors view).
Execution — The query is run against the PostgreSQL database.
Readable Output — The agent then returns the result in a human-friendly format.

We also enhance the agent’s reasoning by providing custom table information. This metadata describes each table and the view in natural language, so the model has context about what fields mean without repeatedly inspecting the schema. For example, we specify that rating is a decimal between 0–5, or that author_id references the authors table.

By enabling view support, we tell the agent that views (like books_with_authors) should be treated as first-class citizens. This allows it to prefer querying the view instead of re-creating the join logic each time, which makes the queries cleaner and more reliable.

Console Output:

Step 2: Adding Callbacks for Raw Results

Sometimes you need access to the raw SQL results for additional processing. Let’s add a callback handler to capture this information:

from langchain.callbacks.base import BaseCallbackHandler

class SQLResultHandler(BaseCallbackHandler):
    """Callback handler to capture raw SQL query results"""
    
    def __init__(self):
        self.latest_sql_result = None
        self.sql_run_ids = set()
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        """Track SQL tool starts"""
        tool_name = serialized.get('name', 'unknown') if isinstance(serialized, dict) else str(serialized)
        if tool_name == "sql_db_query":
            run_id = kwargs.get('run_id')
            self.sql_run_ids.add(run_id)
    
    def on_tool_end(self, output, **kwargs):
        """Capture SQL tool output"""
        run_id = kwargs.get('run_id')
        parent_run_id = kwargs.get('parent_run_id')
        
        # Check if this is a SQL tool end
        if run_id in self.sql_run_ids or parent_run_id in self.sql_run_ids:
            self.latest_sql_result = output
        
        # Clean up run IDs
        self.sql_run_ids.discard(run_id)
        self.sql_run_ids.discard(parent_run_id)
    
    def on_tool_error(self, error, **kwargs):
        """Clean up on SQL tool errors"""
        run_id = kwargs.get('run_id')
        self.sql_run_ids.discard(run_id)
    
    def get_latest_result(self):
        """Get the most recent SQL result"""
        return self.latest_sql_result
    
    def reset(self):
        """Reset for next query"""
        self.latest_sql_result = None
        self.sql_run_ids = set()

# Usage with callback
sql_handler = SQLResultHandler()

response = agent.invoke(
    {"input": "Show me all science fiction books"},
    {"callbacks": [sql_handler]}
)

print("Agent Response:", response["output"])
print("Raw SQL Result:", sql_handler.get_latest_result())

The callback system in LangChain allows you to hook into different stages of the agent’s execution. Our SQLResultHandler specifically captures the output from SQL database queries, giving us access to both the agent’s natural language response and the raw data.

Step 3: Adding Conversation Memory

Now let’s add memory so our agent can handle follow-up questions and maintain context:

from langchain_postgres import PostgresChatMessageHistory
from langchain.memory import ConversationBufferMemory
import psycopg

# Connection for chat history (separate from main DB)
CHAT_HISTORY_DB = "postgresql://username:password@localhost:5432/bookstore"
CHAT_HISTORY_TABLE = "chat_history" #the table that will store our hisory
CHAT_HISTORY_CONN = psycopg.connect(CHAT_HISTORY_DB)

We also have to tell LangChain to create the table that will store the chat history. This will be done only once, you can comment out or remove the snippet later.

# Run this only once
try:
    PostgresChatMessageHistory.create_tables(CHAT_HISTORY_CONN, CHAT_HISTORY_TABLE)
    print(f"Chat history table '{CHAT_HISTORY_TABLE}' created or already exists")
except Exception as e:
    print(f"Note: {e}")

After that we fetch the conversation history and convert it into a readable format for the agent to understand easily. The PostgresChatMessageHistory requires the session_id to be a UUID string.

async def get_session_history(session_id: str) -> PostgresChatMessageHistory:
    """Get chat history for a session"""
    async_conn = await psycopg.AsyncConnection.connect(CHAT_HISTORY_CONN)
    return PostgresChatMessageHistory(
        CHAT_HISTORY_TABLE,
        session_id,
        async_connection=async_conn
    )

async def get_memory(session_id: str) -> ConversationBufferMemory:
    """Create memory with PostgreSQL backing"""
    chat_history = await get_session_history(session_id)
    return ConversationBufferMemory(
        chat_memory=chat_history,
        memory_key="history", 
        return_messages=True
    )

async def format_history(chat_history, max_messages: int = 6):
    """Format recent chat history for context"""
    messages = await chat_history.aget_messages()
    recent_messages = messages[-max_messages:] if len(messages) > max_messages else messages
    
    formatted = []
    for msg in recent_messages:
        role = "User" if msg.type == "human" else "Assistant"
        formatted.append(f"{role}: {msg.content}")
    
    return "\n".join(formatted)

async def create_agent_with_memory(session_id: str):
    """Create agent with conversation memory"""
    memory = await get_memory(session_id)
    
    # Get formatted history for context
    readable_history = await format_history(memory.chat_memory, 6)
    
    # Custom prompt with history
    custom_prefix = f"""
    You are a helpful assistant that can answer questions about a bookstore database.
    You have access to information about books and authors.
    
    Previous conversation context:
    {readable_history}
    
    Be concise and helpful in your responses.
    """
    
    return create_sql_agent(
        toolkit=toolkit,
        llm=llm,
        agent_type="tool-calling",
        prefix=custom_prefix,
        agent_executor_kwargs={"memory": memory},
        verbose=True
    )

# Usage with memory
import asyncio

async def chat_example():
    agent = await create_agent_with_memory("3dc035ae-bc72-4d5a-8569-c87c10aab97f") # Must be a UUID
    
    # First question
    response1 = await agent.ainvoke({"input": "How many books by Jane Austen do we have?"})
    print("Response 1:", response1["output"])
    
    # Follow-up question (will remember context)
    response2 = await agent.ainvoke({"input": "What genres are they?"})
    print("Response 2:", response2["output"])

# Run the example
asyncio.run(chat_example())

The memory system stores conversation history in PostgreSQL, allowing the agent to:

Step 4: FastAPI Web Service

Finally, let’s wrap everything in a FastAPI application for easy deployment:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI(title="SQL Chat Agent", version="1.0.0")

class ChatRequest(BaseModel):
    message: str
    user_id: str

class ChatResponse(BaseModel):
    reply: str
    raw_sql_result: Optional[str] = None

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    """Chat with the SQL agent"""
    
    # Create handler for raw results
    sql_handler = SQLResultHandler()
    
    # Create agent with memory for this user
    agent = await create_agent_with_memory(request.user_id)
    
    # Process the question
    response = await agent.ainvoke(
        {"input": request.message},
        {"callbacks": [sql_handler]}
    )
    
    return ChatResponse(
        reply=response["output"],
        raw_sql_result=sql_handler.get_latest_result()
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Running the Application

Save your code to main.py and run:

uvicorn main:app --reload

Your API will be available at http://localhost:8000. You can test it with:

curl -X POST "http://localhost:8000/chat" \
  -H "Content-Type: application/json" \
  -d '{
  "message": "How many authors do we have?",
  "user_id": "3dc035ae-bc72-4d5a-8569-c87c10aab97f"
}'

Note: The user_id field must always be a valid UUID, otherwise LangChain will throw us an error.

Testing Your Agent

Try these example queries:

Complete Working Example

Here’s the full code putting it all together:

import os
import asyncio
from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import create_engine
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent, SQLDatabaseToolkit
from langchain_postgres import PostgresChatMessageHistory
from langchain.memory import ConversationBufferMemory
from langchain.callbacks.base import BaseCallbackHandler
import psycopg

# Configuration
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
DB_URI = "postgresql+psycopg2://username:password@localhost:5432/bookstore"
CHAT_HISTORY_CONN = "postgresql://username:password@localhost:5432/bookstore"
CHAT_HISTORY_TABLE = "chat_history"

# Database setup
engine = create_engine(DB_URI)

# Define custom table info for better LLM context
custom_table_info = {
    "authors": (
        "A table of authors.\n"
        "- id (SERIAL PRIMARY KEY): Unique ID of author\n"
        "- name (VARCHAR): Name of the author\n"
        "- birth_year (INTEGER): Year of birth\n"
        "- nationality (VARCHAR): Nationality of the author\n"
    ),
    "books": (
        "A table of books.\n"
        "- id (SERIAL PRIMARY KEY): Unique ID of book\n"
        "- title (VARCHAR): Title of the book\n"
        "- author_id (INTEGER): References authors(id)\n"
        "- genre (VARCHAR): Genre of the book\n"
        "- publication_year (INTEGER): Year of publication\n"
        "- rating (DECIMAL): Book rating (0–10)\n"
    ),
    "books_with_authors": (
        "A view combining books and authors.\n"
        "- book_id (INTEGER): ID of the book\n"
        "- title (VARCHAR): Title of the book\n"
        "- genre (VARCHAR): Genre of the book\n"
        "- publication_year (INTEGER): Year of publication\n"
        "- rating (DECIMAL): Rating of the book\n"
        "- author_name (VARCHAR): Name of the author\n"
        "- birth_year (INTEGER): Birth year of the author\n"
        "- nationality (VARCHAR): Nationality of the author\n"
    ),
}

# Initialize SQLDatabase with view support and custom info
db = SQLDatabase(
    engine=engine,
    include_tables=list(custom_table_info.keys()),
    custom_table_info=custom_table_info,
    view_support=True
)
llm = ChatOpenAI(model="gpt-4", temperature=0)
toolkit = SQLDatabaseToolkit(db=db, llm=llm)

# Basic Callback Handler
class SQLResultHandler(BaseCallbackHandler):
    def __init__(self):
        self.latest_sql_result = None
        self.sql_run_ids = set()
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        tool_name = serialized.get('name', 'unknown') if isinstance(serialized, dict) else str(serialized)
        if tool_name == "sql_db_query":
            self.sql_run_ids.add(kwargs.get('run_id'))
    
    def on_tool_end(self, output, **kwargs):
        run_id = kwargs.get('run_id')
        if run_id in self.sql_run_ids:
            self.latest_sql_result = output
            self.sql_run_ids.discard(run_id)
    
    def get_latest_result(self):
        return self.latest_sql_result

# Memory Handling
async def get_session_history(session_id: str):
    async_conn = await psycopg.AsyncConnection.connect(CHAT_HISTORY_CONN)
    return PostgresChatMessageHistory(CHAT_HISTORY_TABLE, session_id, async_connection=async_conn)

async def get_memory(session_id: str):
    chat_history = await get_session_history(session_id)
    return ConversationBufferMemory(chat_memory=chat_history, memory_key="history", return_messages=True)

# Agent Creation
async def create_agent_with_memory(session_id: str):
    memory = await get_memory(session_id)
    return create_sql_agent(
        toolkit=toolkit,
        llm=llm,
        agent_type="tool-calling",
        agent_executor_kwargs={"memory": memory},
        verbose=True
    )

# FastAPI app
app = FastAPI(title="SQL Chat Agent")

# Models
class ChatRequest(BaseModel):
    message: str
    user_id: str

class ChatResponse(BaseModel):
    reply: str
    raw_sql_result: Optional[str] = None

# API Endpoint
@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    sql_handler = SQLResultHandler()
    agent = await create_agent_with_memory(request.user_id)
    
    response = await agent.ainvoke(
        {"input": request.message},
        {"callbacks": [sql_handler]}
    )
    
    return ChatResponse(
        reply=response["output"],
        raw_sql_result=sql_handler.get_latest_result()
    )

# Execution
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Example Request:

{
  "message": "List all books with their authors and ratings",
  "user_id": "44b11b50-9417-4fa5-8e5d-ea968c6dc7d1"
}

Agent Reply:

{
  "reply": "Here are some books with their authors and ratings:\n\n1. 'One Hundred Years of Solitude' by Gabriel García Márquez - Rating: 4.90\n2. '1984' by George Orwell - Rating: 4.80\n3. 'Pride and Prejudice' by Jane Austen - Rating: 4.70\n4. 'Animal Farm' by George Orwell - Rating: 4.60\n5. 'Half of a Yellow Sun' by Chimamanda Ngozi Adichie - Rating: 4.60\n6. 'Kafka on the Shore' by Haruki Murakami - Rating: 4.50\n7. 'Emma' by Jane Austen - Rating: 4.50\n8. 'Americanah' by Chimamanda Ngozi Adichie - Rating: 4.40\n9. 'Adventures of Huckleberry Finn' by Mark Twain - Rating: 4.40\n10. 'Norwegian Wood' by Haruki Murakami - Rating: 4.30",
  "raw_sql_result": "[('One Hundred Years of Solitude', 'Gabriel García Márquez', Decimal('4.90')), ('1984', 'George Orwell', Decimal('4.80')), ('Pride and Prejudice', 'Jane Austen', Decimal('4.70')), ('Animal Farm', 'George Orwell', Decimal('4.60')), ('Half of a Yellow Sun', 'Chimamanda Ngozi Adichie', Decimal('4.60')), ('Kafka on the Shore', 'Haruki Murakami', Decimal('4.50')), ('Emma', 'Jane Austen', Decimal('4.50')), ('Americanah', 'Chimamanda Ngozi Adichie', Decimal('4.40')), ('Adventures of Huckleberry Finn', 'Mark Twain', Decimal('4.40')), ('Norwegian Wood', 'Haruki Murakami', Decimal('4.30'))]"
}

Key Benefits

This approach gives you:

Natural Language Interface: Users can ask questions in plain English
Conversation Memory: Maintains context across multiple questions
Raw Data Access: Callbacks provide access to underlying SQL results, or literally anything else you want to do after agent call.
REST API: Easy integration with web apps, mobile apps, or other services
Scalable: Supports multiple concurrent users with session management

Working Example from GitHub

The complete working example is available in the GitHub repository: 👉 LangChain SQL Agent Demo (GitHub)

To run it locally:

git clone https://github.com/Silversky-Technology/langchain-sql-agent-guide.git
cd langchain-sql-agent-guide
python -m venv venv
source venv/bin/activate
pip install fastapi uvicorn langchain-openai langchain-community sqlalchemy psycopg2-binary langchain-postgres asyncio
uvicorn main:app --reload

Remember to add in your own API key and database credentials.

Your API will be available to test at http://localhost:8000!

Conclusion

You now have a complete conversational SQL agent that can handle complex database queries through natural language. The modular design makes it easy to extend with additional features like:

The combination of LangChain’s SQL agent capabilities with FastAPI’s modern web framework creates a powerful foundation for building intelligent database interfaces.

Full working example is available here: LangChain SQL Agent Demo (GitHub)

Building Conversational SQL Agents with LangChain

Building Conversational SQL Agents with LangChain

shares

The Problems We Faced:

What We’ll Build

Prerequisites

Setup

Step 1: Basic SQL Agent

Console Output:

Step 2: Adding Callbacks for Raw Results

Step 3: Adding Conversation Memory

Step 4: FastAPI Web Service

Running the Application

Testing Your Agent

Complete Working Example

Example Request:

Agent Reply:

Key Benefits

Working Example from GitHub

Conclusion

Innovative Software Solutions for Every Industry

Got a project in mind? Let’s build something exceptional, just for you!

Services

Explore

Inquiry

Location

© 2025 Silversky Technology