Retrieving from SQL and Structured Data

Not all knowledge lives in text files. Orders, customers, prices and logs sit in SQL tables and spreadsheets. Vector search is poor at questions like total paid orders per city, because that needs exact filtering and arithmetic, not similarity. In this lesson you will learn two ways to bring structured data into a RAG app: let an LLM write SQL, or convert rows into documents with metadata.

Overview: how structured retrieval works

Unstructured RAG finds passages that are similar to the question. Structured retrieval asks a database to compute the answer. The database is exact, fast and always up to date, so you should use it for counts, sums, joins, date ranges and lookups by ID.

The most common pattern is called text-to-SQL. It has these steps:

1. Read the database schema (table names, columns, types, a few sample rows).
2. Put the schema and the user question into a prompt.
3. The LLM writes a SQL query.
4. Your code validates the query, then runs it on the database.
5. The rows come back, and (optionally) a second LLM call turns them into a natural-language answer.

The LLM never touches the data directly. It only produces text, and your code decides whether that text is safe to run. This is the key security idea of the whole lesson: LLM output is untrusted input.

The second pattern is rows as documents. You turn each row (or group of rows) into a Document with readable text and metadata such as city or id. You embed the text, store it in a vector store, and at query time you combine similarity search with a metadata filter. This works well when the text columns are descriptive (product descriptions, support tickets) and the user question is fuzzy.

Key concepts and building blocks

These pieces from LangChain are used in this lesson:

• SQLDatabase (from langchain_community.utilities) wraps a SQLAlchemy engine. SQLDatabase.from_uri(url) connects. db.get_table_info() returns CREATE TABLE statements plus sample rows. db.run(sql) executes SQL and returns the result as a string. db.dialect names the SQL flavour.
• ChatPromptTemplate builds the prompt with the schema and question.
• RunnableLambda and RunnablePassthrough.assign let you plug your own Python steps into an LCEL chain with the | operator.
• Document, an embeddings class, and a vector store (InMemoryVectorStore) for the rows-as-documents pattern.

The general shape of a text-to-SQL chain is:

question -> prompt(schema + question) -> LLM -> extract SQL -> validate -> db.run -> rows -> (LLM answer)

Install the packages first. SQLDatabase needs SQLAlchemy, which comes with langchain-community.

pip install langchain langchain-core langchain-community

Examples

Example 1: Connect to a database and read it

We create a small SQLite shop database with the standard library, then open it through LangChain. SQLite needs no server, so everyone can follow along.

import sqlite3
from langchain_community.utilities import SQLDatabase

conn = sqlite3.connect('shop.db')
conn.execute('DROP TABLE IF EXISTS orders')
conn.execute('DROP TABLE IF EXISTS customers')
conn.execute('CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, city TEXT)')
conn.execute('CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL, status TEXT)')
conn.executemany('INSERT INTO customers VALUES (?, ?, ?)', [
    (1, 'Asha', 'Pune'), (2, 'Ravi', 'Delhi'), (3, 'Meena', 'Pune'),
])
conn.executemany('INSERT INTO orders VALUES (?, ?, ?, ?)', [
    (1, 1, 250.0, 'paid'), (2, 1, 100.0, 'paid'),
    (3, 2, 75.5, 'refunded'), (4, 3, 300.0, 'paid'),
])
conn.commit()
conn.close()

db = SQLDatabase.from_uri('sqlite:///shop.db')
print(db.dialect)
print(sorted(db.get_usable_table_names()))
print(db.run('SELECT name, city FROM customers ORDER BY id'))

Output:

sqlite
['customers', 'orders']
[('Asha', 'Pune'), ('Ravi', 'Delhi'), ('Meena', 'Pune')]

Notice that db.run returns a string, not a list. That is deliberate: the string can be pasted straight into a prompt for the answer step.

Example 2: A text-to-SQL chain with a safety guard

To keep the example runnable without an API key, the LLM is replaced by a fake function that returns a fixed reply. In a real app you swap it for a chat model (shown later). Everything else, including the guard, stays the same.

import re
from langchain_community.utilities import SQLDatabase
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough

db = SQLDatabase.from_uri('sqlite:///shop.db')

prompt = ChatPromptTemplate.from_messages([
    ('system', 'You are a SQLite expert. Write ONE read-only SELECT query. Return only SQL. Schema: {schema}'),
    ('human', '{question}'),
])

FAKE_REPLY = '''```sql
SELECT c.city, SUM(o.amount) AS total FROM orders o JOIN customers c ON c.id = o.customer_id WHERE o.status = 'paid' GROUP BY c.city ORDER BY total DESC;
```'''


def fake_llm(prompt_value):
    assert 'CREATE TABLE' in prompt_value.to_string()
    return FAKE_REPLY


def extract_sql(text):
    m = re.search(r'```(?:sql)?\s*(.*?)```', text, re.S)
    return (m.group(1) if m else text).strip()


FORBIDDEN = re.compile(r'\b(insert|update|delete|drop|alter|create|attach|pragma|replace)\b', re.I)


def check_sql(sql):
    body = sql.strip().rstrip(';')
    if ';' in body:
        raise ValueError('multiple statements')
    if not body.lower().startswith('select'):
        raise ValueError('only SELECT')
    if FORBIDDEN.search(body):
        raise ValueError('forbidden keyword')
    return body


sql_chain = (
    RunnablePassthrough.assign(schema=lambda _: db.get_table_info())
    | prompt
    | RunnableLambda(fake_llm)
    | RunnableLambda(extract_sql)
    | RunnableLambda(check_sql)
)

sql = sql_chain.invoke({'question': 'Total paid order value per city?'})
print('SQL:', sql)
print('Result:', db.run(sql))

for bad in ['DROP TABLE orders', 'SELECT 1; DELETE FROM orders']:
    try:
        check_sql(bad)
    except ValueError as e:
        print('rejected:', e)

Output:

SQL: SELECT c.city, SUM(o.amount) AS total FROM orders o JOIN customers c ON c.id = o.customer_id WHERE o.status = 'paid' GROUP BY c.city ORDER BY total DESC
Result: [('Pune', 650.0)]
rejected: only SELECT
rejected: multiple statements

Pune has paid orders of 250 + 100 + 300 = 650. Delhi has only a refunded order, so it is filtered out. The guard rejects anything that is not a single SELECT. The LLM often wraps SQL in a Markdown fence, so extract_sql removes it before validation.

Example 3: Use a real LLM

To use a real model, replace RunnableLambda(fake_llm) with a chat model and read the key from an environment variable. Model names change often, so check your provider’s documentation for the current name.

import os
from langchain.chat_models import init_chat_model

assert os.environ.get('ANTHROPIC_API_KEY'), 'set your API key first'
llm = init_chat_model('anthropic:claude-sonnet-5-5', temperature=0)

sql_chain = (
    RunnablePassthrough.assign(schema=lambda _: db.get_table_info())
    | prompt
    | llm
    | StrOutputParser()
    | RunnableLambda(extract_sql)
    | RunnableLambda(check_sql)
)

A chat model returns a message object, so StrOutputParser (from langchain_core.output_parsers) converts it to plain text before cleaning. Use temperature=0 so the SQL is as repeatable as possible.

Example 4: Rows as documents with a metadata filter

Now the second pattern. Each customer row becomes a Document. We use DeterministicFakeEmbedding so no network is needed. With real embeddings, the same code gives meaningful similarity.

import sqlite3
from langchain_core.documents import Document
from langchain_core.embeddings import DeterministicFakeEmbedding
from langchain_core.vectorstores import InMemoryVectorStore

conn = sqlite3.connect('shop.db')
rows = conn.execute('SELECT id, name, city FROM customers').fetchall()
conn.close()

docs = [
    Document(
        page_content=f'Customer {name} lives in {city}.',
        metadata={'id': i, 'city': city},
    )
    for i, name, city in rows
]

store = InMemoryVectorStore(DeterministicFakeEmbedding(size=64))
store.add_documents(docs)

hits = store.similarity_search(
    'who lives in Pune?',
    k=3,
    filter=lambda d: d.metadata['city'] == 'Pune',
)
print(sorted(d.metadata['id'] for d in hits))

Output:

[1, 3]

The filter removes every non-Pune document before ranking, so only IDs 1 and 3 can come back. The id in the metadata lets you go back to the database for fresh details. In production vector stores (Chroma, pgvector and others) the filter is usually a dictionary or expression instead of a Python function, so check your store’s documentation.

Under the hood: step by step

What happens when you call sql_chain.invoke(...) in Example 2?

1. RunnablePassthrough.assign keeps the input dictionary and adds a schema key by calling db.get_table_info().
2. The prompt template fills {schema} and {question} and produces a chat prompt value.
3. The model (or our fake) returns text containing SQL.
4. extract_sql strips the Markdown fence.
5. check_sql raises an error for anything unsafe. In LCEL, an exception stops the chain, so unsafe SQL never reaches the database.
6. Your code calls db.run(sql) and gets the rows as a string.

The schema text is what makes the LLM accurate. Real column names, types and two or three sample rows tell it that status holds values like 'paid'. Without sample rows, models often guess values such as 'PAID' or 'completed'.

Common Mistakes

Mistake 1: Running LLM SQL on a powerful connection

db = SQLDatabase.from_uri('postgresql://admin:secret@host/prod')
result = db.run(llm_output)  # runs whatever the model wrote

This is wrong because the model can be tricked by a prompt injection (for example, text inside a database row saying ignore your rules and drop the table). With an admin account, that could destroy data. Fix it with defence in depth: validate the query (as in Example 2) and connect with a user that only has SELECT rights on the needed tables. For SQLite you can open the file read-only:

import sqlite3

ro = sqlite3.connect('file:shop.db?mode=ro', uri=True)
print(ro.execute('SELECT COUNT(*) FROM orders').fetchone()[0])
try:
    ro.execute('DELETE FROM orders')
except sqlite3.OperationalError as e:
    print('blocked:', e)

4
blocked: attempt to write a readonly database

Even if the validator has a bug, the database itself refuses the write.

Mistake 2: Sending the whole schema of a big database

db = SQLDatabase.from_uri(url)  # 400 tables go into every prompt

Hundreds of tables waste tokens and confuse the model. Limit what it can see:

db = SQLDatabase.from_uri(
    url,
    include_tables=['customers', 'orders'],
    sample_rows_in_table_info=2,
)

Mistake 3: Embedding numeric rows and expecting maths

Putting rows like Order 4: 300.0 paid in a vector store and asking for the total will not work. Similarity search returns a few similar rows, not all matching rows, and the model may add them up wrongly. Use SQL for aggregation. Use embeddings for descriptive text.

Best Practices

• Use a read-only database user with access only to the tables the app needs. Never rely on the prompt alone for safety.
• Validate generated SQL: single statement, SELECT only, no dangerous keywords. A SQL parser library gives stronger checks than regular expressions.
• Add a LIMIT (or wrap the query) so one question cannot return millions of rows.
• Set a query timeout on the database side.
• Include schema, sample rows and short column descriptions in the prompt. Explain business rules such as revenue means status = paid.
• Use temperature=0 for SQL generation. Log the question, the generated SQL and the result for debugging.
• If the query fails, feed the error message back to the LLM for one or two retries, then give up politely.
• For rows-as-documents, store the primary key in metadata and re-sync the index when the table changes.
• Route questions: send exact or numeric questions to SQL, fuzzy questions to the vector store.
• Test with a golden set of questions and known correct answers, because SQL mistakes can look confident.

Practice Exercises

1. Add a LIMIT guard. Extend check_sql so that if the query has no LIMIT, it appends LIMIT 50. Test with one query that has a limit and one that does not. Hint: use re.search(r'\blimit\b', body, re.I).
2. Add a new question. Change FAKE_REPLY so the fake LLM answers How many customers live in each city? Check that the guard accepts it and that db.run returns the right counts.
3. Filter by two fields. Add a tier column to the customers table, put it in the document metadata, and change the filter so only Pune customers with tier gold are returned.

Summary

• Use SQL for exact, computed answers and vector search for fuzzy, descriptive text.
• Text-to-SQL means: schema into prompt, LLM writes SQL, you validate, then run it.
• SQLDatabase.from_uri, get_table_info and run are the core LangChain helpers.
• Treat LLM-written SQL as untrusted: validate it and use a read-only database account.
• Limit visible tables, include sample rows and cap result size.
• Rows can become Document objects with metadata, and metadata filters narrow similarity search to the right records.
• A hybrid app routes each question to the retriever that fits it best.