Text-to-SQL Agent

Use a seven-table schema with joins, aggregations, and nullable foreign keys, so the agent has to reason about real relationships instead of a single flat table.

from sqlalchemy import create_engine, text
from langchain_community.utilities import SQLDatabase

COMPLEX_DB_SCHEMA = """
CREATE TABLE users (user_id INTEGER PRIMARY KEY, username TEXT NOT NULL UNIQUE, email TEXT NOT NULL UNIQUE, first_name TEXT, last_name TEXT, registration_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, account_type TEXT CHECK (account_type IN ('standard', 'premium', 'admin')) DEFAULT 'standard');
CREATE TABLE product_categories (category_id INTEGER PRIMARY KEY, parent_category_id INTEGER, name TEXT NOT NULL, FOREIGN KEY (parent_category_id) REFERENCES product_categories(category_id) ON DELETE SET NULL);
CREATE TABLE products (product_id INTEGER PRIMARY KEY, sku TEXT NOT NULL UNIQUE, name TEXT NOT NULL, price DECIMAL(10, 2) NOT NULL, inventory_count INTEGER DEFAULT 0, is_active BOOLEAN DEFAULT TRUE);
CREATE TABLE product_category_mappings (product_id INTEGER NOT NULL, category_id INTEGER NOT NULL, PRIMARY KEY (product_id, category_id), FOREIGN KEY (product_id) REFERENCES products(product_id) ON DELETE CASCADE, FOREIGN KEY (category_id) REFERENCES product_categories(category_id) ON DELETE CASCADE);
CREATE TABLE orders (order_id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, status TEXT CHECK (status IN ('pending', 'processing', 'shipped', 'delivered', 'cancelled', 'refunded')) DEFAULT 'pending', total_amount DECIMAL(10, 2) NOT NULL, payment_status TEXT CHECK (payment_status IN ('pending', 'authorized', 'paid', 'refunded', 'failed')) DEFAULT 'pending', FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE RESTRICT);
CREATE TABLE order_items (order_item_id INTEGER PRIMARY KEY, order_id INTEGER NOT NULL, product_id INTEGER NOT NULL, quantity INTEGER NOT NULL, unit_price DECIMAL(10, 2) NOT NULL, FOREIGN KEY (order_id) REFERENCES orders(order_id) ON DELETE CASCADE, FOREIGN KEY (product_id) REFERENCES products(product_id) ON DELETE RESTRICT);
CREATE TABLE reviews (review_id INTEGER PRIMARY KEY, product_id INTEGER NOT NULL, user_id INTEGER NOT NULL, rating INTEGER NOT NULL CHECK (rating BETWEEN 1 AND 5), review_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products(product_id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE);
"""

# Rows keyed by table name, inserted in FK-safe order (users/categories/products first).
COMPLEX_SAMPLE_DATA = {
    "users": [
        {"user_id": 1, "username": "amy_t", "email": "amy@example.com", "first_name": "Amy", "last_name": "Tran", "account_type": "premium"},
        {"user_id": 2, "username": "ben_k", "email": "ben@example.com", "first_name": "Ben", "last_name": "Kim", "account_type": "standard"},
    ],
    "product_categories": [
        {"category_id": 1, "parent_category_id": None, "name": "Electronics"},
        {"category_id": 2, "parent_category_id": 1, "name": "Headphones"},
        {"category_id": 3, "parent_category_id": None, "name": "Home & Kitchen"},
    ],
    "products": [
        {"product_id": 1, "sku": "ELE-001", "name": "Noise-Cancelling Headphones", "price": 149.99, "inventory_count": 40, "is_active": True},
        {"product_id": 2, "sku": "ELE-002", "name": "Wireless Earbuds", "price": 79.99, "inventory_count": 120, "is_active": True},
        {"product_id": 3, "sku": "HOM-001", "name": "Stainless Steel Kettle", "price": 34.50, "inventory_count": 60, "is_active": True},
    ],
    "product_category_mappings": [
        {"product_id": 1, "category_id": 2},
        {"product_id": 2, "category_id": 2},
        {"product_id": 3, "category_id": 3},
    ],
    "orders": [
        {"order_id": 1, "user_id": 1, "status": "delivered", "total_amount": 229.98, "payment_status": "paid"},
        {"order_id": 2, "user_id": 2, "status": "pending", "total_amount": 239.97, "payment_status": "pending"},
    ],
    "order_items": [
        {"order_item_id": 1, "order_id": 1, "product_id": 1, "quantity": 1, "unit_price": 149.99},
        {"order_item_id": 2, "order_id": 1, "product_id": 2, "quantity": 1, "unit_price": 79.99},
        {"order_item_id": 3, "order_id": 2, "product_id": 2, "quantity": 3, "unit_price": 79.99},
    ],
    "reviews": [
        {"review_id": 1, "product_id": 1, "user_id": 1, "rating": 5},
        {"review_id": 2, "product_id": 2, "user_id": 2, "rating": 3},
    ],
}


def setup_database():
    """Creates an in-memory SQLite database with the schema and sample rows."""
    engine = create_engine("sqlite:///:memory:")

    with engine.connect() as conn:
        for statement in COMPLEX_DB_SCHEMA.split(";"):
            statement = statement.strip()
            if statement:
                conn.execute(text(statement))
        conn.commit()

        for table_name, rows in COMPLEX_SAMPLE_DATA.items():
            if not rows:
                continue
            columns = list(rows[0].keys())
            placeholders = ", ".join(f":{col}" for col in columns)
            insert_query = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})"
            for row in rows:
                conn.execute(text(insert_query), row)
            conn.commit()

    return SQLDatabase(engine=engine)

You should see no output here. setup_database() returns a SQLDatabase wrapping the populated SQLite engine, ready for the agent to query.

Build the agent with bounded retries, parsing-error handling, and intermediate-step capture, so you can extract the exact SQL it executes for each question.

from langchain_openai import ChatOpenAI
from langchain_community.agent_toolkits import create_sql_agent


def get_model(model_name: str) -> ChatOpenAI:
    return ChatOpenAI(model=model_name, temperature=0)


def create_improved_sql_agent(llm, db):
    """Bounded retries, parsing-error handling, and intermediate-step capture."""
    return create_sql_agent(
        llm=llm,
        db=db,
        agent_type="tool-calling",
        verbose=True,
        max_iterations=5,
        handle_parsing_errors=True,
        return_intermediate_steps=True,
    )

You should see no output here. create_improved_sql_agent() returns a LangChain AgentExecutor bound to the SQLite database from Step 1.

Register a trace provider with five EvalTag entries: four built-in evals (completeness, groundedness, text_to_sql, detect_hallucination) and one custom eval (table_checker) that checks the agent picked the right tables. Each EvalTag maps evaluation inputs to specific span attributes, so Future AGI knows what data to score.

from traceai_langchain import LangChainInstrumentor
from fi_instrumentation import register
from fi_instrumentation.fi_types import (
    EvalName,
    EvalSpanKind,
    EvalTag,
    EvalTagType,
    ModelChoices,
    ProjectType,
)

# Generated from a table instead of four near-identical EvalTag blocks.
BUILTIN_EVALS = [
    (EvalSpanKind.AGENT, EvalName.COMPLETENESS, "Completeness"),
    (EvalSpanKind.AGENT, EvalName.GROUNDEDNESS, "Groundedness"),
    (EvalSpanKind.TOOL, EvalName.TEXT_TO_SQL, "Text-to-SQL"),
    (EvalSpanKind.AGENT, EvalName.DETECT_HALLUCINATION, "Hallucination"),
]

eval_tags = [
    EvalTag(
        type=EvalTagType.OBSERVATION_SPAN,
        value=span_kind,
        eval_name=eval_name,
        config={},
        mapping={"input": "raw.input", "output": "raw.output"},
        custom_eval_name=label,
        model=ModelChoices.TURING_LARGE,
    )
    for span_kind, eval_name, label in BUILTIN_EVALS
]
eval_tags.append(
    EvalTag(
        type=EvalTagType.OBSERVATION_SPAN,
        value=EvalSpanKind.TOOL,
        eval_name="table_checker",
        config={},
        mapping={"query": "raw.input", "tables": "raw.output"},
        custom_eval_name="table_checker",
        model=ModelChoices.TURING_LARGE,
    )
)

trace_provider = register(
    project_type=ProjectType.EXPERIMENT,
    project_name="text-to-sql-agent",
    eval_tags=eval_tags,
)

LangChainInstrumentor().instrument(tracer_provider=trace_provider)

You should see no output here, but every LangChain call made after this point is captured as a span in the text-to-sql-agent project, with all five evals queued to run against it. A span records one operation (one tool call, one LLM call); a trace connects the spans for one full agent run.

Run one question through the instrumented agent to confirm tracing is wired up before running the full set:

model = get_model("gpt-4o")
db = setup_database()
agent_executor = create_improved_sql_agent(model, db)

first_run = agent_executor.invoke({"input": "How many products are in each category?"})
print(first_run["output"])

Expected output (SQL and phrasing vary by model):

Headphones has 2 products, and Home & Kitchen has 1 product.

This call flows through the LangChainInstrumentor from above and lands as a trace in your Future AGI project, with the completeness, groundedness, text-to-sql, hallucination, and table_checker scores attached to it.

Run the agent from Step 2 against a fixed set of questions, extracting the SQL it executed from each response and timing every call.

import time

TEXT2SQL_QUESTIONS = [
    "How many products are in each category?",
    "What is the total amount spent by each user?",
    "Which products have an average rating below 4?",
    "List all orders that still have a pending payment.",
    "What is the top-selling product by quantity ordered?",
]


def execute_sql_query(agent_executor, question):
    """Runs one question through the agent and extracts the SQL it executed."""
    start_time = time.time()
    try:
        agent_result = agent_executor.invoke({"input": question})
        sql_query = ""
        for step in agent_result.get("intermediate_steps", []):
            tool_input = step[0].tool_input
            if isinstance(tool_input, str) and any(kw in tool_input.upper() for kw in ["SELECT", "INSERT", "UPDATE"]):
                sql_query = tool_input
                break
        return {
            "execution_success": True,
            "sql_query": sql_query,
            "result": agent_result["output"],
            "error": "",
            "latency": time.time() - start_time,
        }
    except Exception as e:
        return {
            "execution_success": False,
            "sql_query": "",
            "result": "",
            "error": str(e),
            "latency": time.time() - start_time,
        }


def run_complex_text2sql_experiment(model_name):
    model = get_model(model_name)
    db = setup_database()
    agent_executor = create_improved_sql_agent(model, db)

    results = []
    for question in TEXT2SQL_QUESTIONS:
        query_result = execute_sql_query(agent_executor, question)
        results.append({"model": model_name, "question": question, **query_result})
    return results


results = run_complex_text2sql_experiment("gpt-4o")
print(f"Ran {len(results)} questions against the agent.")

Expected output:

Ran 5 questions against the agent.

Each agent_executor.invoke() call flows through the same instrumentation from Step 3 and lands as a trace in your Future AGI project.

Compute summary metrics from the raw results so you can compare agent versions or models without opening the dashboard for every run.

def collect_metrics(results):
    total = len(results)
    successes = [r for r in results if r["execution_success"]]
    latencies = [r["latency"] for r in results]
    return {
        "success_rate": len(successes) / total if total else 0.0,
        "failure_count": total - len(successes),
        "avg_latency_s": sum(latencies) / total if total else 0.0,
        "min_latency_s": min(latencies) if latencies else 0.0,
        "max_latency_s": max(latencies) if latencies else 0.0,
    }


metrics = collect_metrics(results)
for name, value in metrics.items():
    print(f"{name}: {value}")

Expected output (values vary by model and run):

success_rate: 1.0
failure_count: 0
avg_latency_s: 3.8
min_latency_s: 2.1
max_latency_s: 6.4

These numbers are illustrative: run the cookbook against your own database and model to get real figures. Open your project in the Future AGI dashboard. The trace explorer shows each of the 5 runs as a trace, with the SQL agent’s tool calls nested underneath as spans, and the completeness, groundedness, text-to-sql, hallucination, and table_checker scores attached to each one.