What is a SQL Query Agent?
A SQL Query Agent built with LangChain is an intelligent system that translates natural language questions into executable SQL queries, runs them against a database, and returns meaningful answers. Unlike traditional query builders that require manual SQL construction, a LangChain agent autonomously reasons about the database schema, generates appropriate queries, executes them, and interprets the results—all while handling errors gracefully through iterative refinement.
At its core, the agent leverages a Large Language Model (LLM) as its reasoning engine. LangChain provides the orchestration layer, connecting the LLM to database tools, managing conversation memory, and structuring the agent's decision-making loop. The agent receives a user's question in plain English, examines available tables and columns, formulates a SQL statement, executes it via a database connection, and synthesizes the raw results into a coherent response.
Core Components of a SQL Agent
- LLM (Large Language Model): The brain of the agent—typically GPT-4, Claude, or an open-source model—that generates SQL and interprets results
- Database Toolkit: LangChain tools that allow the agent to inspect schema, execute queries, and retrieve metadata
- Database Connection: A SQLAlchemy engine or direct connection to your database (SQLite, PostgreSQL, MySQL, Snowflake, etc.)
- Agent Executor: The LangChain runtime that manages the reasoning loop, tool selection, and error recovery
- Prompt Templates: Structured instructions that guide the agent's behavior and SQL generation patterns
Why SQL Query Agents Matter
SQL Query Agents bridge the gap between data accessibility and technical expertise. In organizations where domain experts lack SQL proficiency, these agents democratize data access. Instead of relying on data engineering teams to write queries, business analysts, marketers, and executives can ask questions directly and receive instant insights.
The impact extends beyond accessibility. SQL agents dramatically reduce the iteration cycle in exploratory data analysis. An analyst can ask follow-up questions conversationally, refining their investigation without switching contexts or waiting for engineering support. The agent maintains context across interactions, enabling a fluid, collaborative exploration experience that mirrors working with a human data analyst.
Key Advantages Over Traditional Approaches
- Schema Discovery: Agents automatically explore table structures, eliminating the need for users to memorize column names or relationships
- Error Self-Correction: When a query fails, the agent diagnoses the error, adjusts the SQL, and retries—handling edge cases like type mismatches or ambiguous joins
- Contextual Reasoning: Agents maintain conversation history, enabling follow-up questions that reference previous queries without repetition
- Multi-Step Queries: Complex questions requiring multiple intermediate queries are handled transparently, with the agent breaking down the problem into manageable steps
Setting Up Your Environment
Before building your agent, you'll need to install the necessary packages and configure your database connection. This guide assumes you have Python 3.9+ installed and a working database to query.
Installation
pip install langchain langchain-community langchain-openai sqlalchemy sqlite3For production use with specific databases, you may need additional drivers:
# PostgreSQL
pip install psycopg2-binary
# MySQL
pip install mysql-connector-python
# Snowflake
pip install snowflake-sqlalchemy
# BigQuery
pip install google-cloud-bigquery sqlalchemy-bigquerySetting Up Your API Keys
Store your API keys securely using environment variables. Never hardcode credentials in your source code.
import os
os.environ["OPENAI_API_KEY"] = "your-openai-api-key-here"
# Alternatively, use dotenv
# from dotenv import load_dotenv
# load_dotenv()Preparing a Sample Database
For this tutorial, we'll create a sample SQLite database representing an e-commerce platform. This gives us a concrete dataset to query against.
import sqlite3
from sqlalchemy import create_engine, text
# Create SQLite database with sample data
engine = create_engine("sqlite:///ecommerce.db")
with engine.connect() as conn:
# Create tables
conn.execute(text("""
CREATE TABLE IF NOT EXISTS customers (
customer_id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE,
signup_date DATE,
region TEXT
)
"""))
conn.execute(text("""
CREATE TABLE IF NOT EXISTS products (
product_id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT,
price DECIMAL(10,2),
stock_quantity INTEGER
)
"""))
conn.execute(text("""
CREATE TABLE IF NOT EXISTS orders (
order_id INTEGER PRIMARY KEY,
customer_id INTEGER,
order_date DATE,
total_amount DECIMAL(10,2),
status TEXT,
FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
)
"""))
conn.execute(text("""
CREATE TABLE IF NOT EXISTS order_items (
item_id INTEGER PRIMARY KEY,
order_id INTEGER,
product_id INTEGER,
quantity INTEGER,
unit_price DECIMAL(10,2),
FOREIGN KEY (order_id) REFERENCES orders(order_id),
FOREIGN KEY (product_id) REFERENCES products(product_id)
)
"""))
conn.commit()
# Insert sample data
conn.execute(text("""
INSERT INTO customers (customer_id, name, email, signup_date, region) VALUES
(1, 'Alice Johnson', 'alice@example.com', '2023-01-15', 'North'),
(2, 'Bob Smith', 'bob@example.com', '2023-03-22', 'South'),
(3, 'Charlie Brown', 'charlie@example.com', '2023-06-10', 'North'),
(4, 'Diana Prince', 'diana@example.com', '2023-09-05', 'West'),
(5, 'Evan Williams', 'evan@example.com', '2024-01-20', 'East')
"""))
conn.execute(text("""
INSERT INTO products (product_id, name, category, price, stock_quantity) VALUES
(101, 'Wireless Mouse', 'Electronics', 29.99, 150),
(102, 'Mechanical Keyboard', 'Electronics', 89.99, 75),
(103, 'USB-C Hub', 'Electronics', 45.00, 200),
(104, 'Standing Desk', 'Furniture', 599.99, 20),
(105, 'Ergonomic Chair', 'Furniture', 849.99, 10),
(106, 'Monitor Arm', 'Accessories', 129.99, 45)
"""))
conn.execute(text("""
INSERT INTO orders (order_id, customer_id, order_date, total_amount, status) VALUES
(1001, 1, '2024-02-01', 119.98, 'completed'),
(1002, 1, '2024-02-15', 599.99, 'completed'),
(1003, 2, '2024-03-05', 89.99, 'completed'),
(1004, 3, '2024-03-20', 974.98, 'pending'),
(1005, 4, '2024-04-01', 174.99, 'completed'),
(1006, 2, '2024-04-10', 849.99, 'shipped'),
(1007, 5, '2024-04-15', 45.00, 'pending')
"""))
conn.execute(text("""
INSERT INTO order_items (item_id, order_id, product_id, quantity, unit_price) VALUES
(2001, 1001, 101, 2, 29.99),
(2002, 1001, 103, 1, 45.00),
(2003, 1002, 104, 1, 599.99),
(2004, 1003, 102, 1, 89.99),
(2005, 1004, 104, 1, 599.99),
(2006, 1004, 105, 1, 849.99),
(2007, 1005, 106, 1, 129.99),
(2008, 1005, 103, 1, 45.00),
(2009, 1006, 105, 1, 849.99),
(2010, 1007, 103, 1, 45.00)
"""))
conn.commit()
print("Sample database created successfully with tables: customers, products, orders, order_items")Building Your First SQL Query Agent
Now we'll build a complete SQL agent from scratch. This walkthrough covers every component: the LLM, database connection, toolkit, and agent executor.
Step 1: Create the Database Connection and LLM
from langchain_openai import ChatOpenAI
from sqlalchemy import create_engine
from langchain_community.utilities.sql_database import SQLDatabase
# Connect to our SQLite database
engine = create_engine("sqlite:///ecommerce.db")
db = SQLDatabase(engine=engine)
# Initialize the LLM - using GPT-4 for reliable SQL generation
llm = ChatOpenAI(
model="gpt-4-turbo",
temperature=0, # Zero temperature for deterministic SQL output
verbose=True
)
print(f"Database dialect: {db.dialect}")
print(f"Available tables: {db.get_usable_table_names()}")Step 2: Configure the SQL Toolkit
LangChain provides a specialized SQL toolkit that equips the agent with tools for schema inspection, query execution, and error checking. Each tool serves a specific purpose in the agent's workflow.
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain_community.tools.sql_database.tool import (
QuerySQLDataBaseTool,
InfoSQLDatabaseTool,
ListSQLDatabaseTool,
QuerySQLCheckerTool
)
# Create the full toolkit
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
# Inspect available tools
tools = toolkit.get_tools()
for tool in tools:
print(f"Tool: {tool.name}")
print(f"Description: {tool.description[:100]}...")
print("---")The toolkit typically includes these essential tools:
- sql_db_query: Executes a SQL query against the database and returns results
- sql_db_schema: Retrieves table schemas, column types, and sample rows
- sql_db_list_tables: Lists all available tables in the database
- sql_db_query_checker: Validates SQL syntax before execution using the LLM
Step 3: Assemble the Agent
We'll use LangChain's SQL agent executor, which implements a ReAct-style reasoning loop. The agent observes the question, reasons about which tool to use, acts by calling the tool, and repeats until it can provide a final answer.
from langchain.agents import create_sql_agent
from langchain.agents.agent_types import AgentType
# Create the SQL agent with OpenAI Functions agent type
agent_executor = create_sql_agent(
llm=llm,
toolkit=toolkit,
agent_type=AgentType.OPENAI_FUNCTIONS, # Structured function calling
verbose=True,
max_iterations=10, # Prevent infinite loops
max_execution_time=60, # Timeout in seconds
early_stopping_method="generate", # Graceful fallback if stuck
handle_parsing_errors=True, # Recover from output parsing failures
)
print("SQL Agent created successfully. Ready to accept queries.")Step 4: Query Your Data
With the agent assembled, you can now ask natural language questions. The agent handles the entire pipeline autonomously.
# Example 1: Simple aggregation query
result = agent_executor.invoke({
"input": "How many customers do we have in each region?"
})
print(result["output"])
# Example 2: Multi-table join with filtering
result = agent_executor.invoke({
"input": "Show me the top 3 customers by total order amount, include their names and total spent"
})
print(result["output"])
# Example 3: Time-based analysis
result = agent_executor.invoke({
"input": "What was the average order value in February 2024?"
})
print(result["output"])When you run these queries, you'll see the agent's reasoning process in the verbose output—it first retrieves the schema, formulates a SQL query, executes it, and then synthesizes the results into a natural language answer.
Step 5: Adding Conversation Memory
To enable follow-up questions, add memory to your agent. This allows users to ask iterative questions without repeating context.
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# Add memory for multi-turn conversations
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# Recreate agent with memory
agent_executor_with_memory = create_sql_agent(
llm=llm,
toolkit=toolkit,
agent_type=AgentType.OPENAI_FUNCTIONS,
verbose=True,
memory=memory,
max_iterations=10,
)
# Multi-turn conversation example
print("--- First Query ---")
result1 = agent_executor_with_memory.invoke({
"input": "Which product category generates the most revenue?"
})
print(result1["output"])
print("\n--- Follow-up Query ---")
result2 = agent_executor_with_memory.invoke({
"input": "What are the specific products in that category and their individual sales?"
})
print(result2["output"])
print("\n--- Another Follow-up ---")
result3 = agent_executor_with_memory.invoke({
"input": "Now show me only those with sales above $500"
})
print(result3["output"])Advanced Customization Techniques
As your application matures, you'll want to customize the agent's behavior beyond the defaults. LangChain offers extensive hooks for tailoring prompts, adding custom tools, and implementing guardrails.
Custom Prompt Engineering
The agent's SQL generation quality heavily depends on the prompt. You can inject domain-specific knowledge, formatting rules, and safety constraints directly into the system prompt.
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
custom_system_prompt = """You are a senior data analyst expert in SQL queries for an e-commerce database.
Database context:
- All monetary values are in USD
- Dates use ISO format (YYYY-MM-DD)
- The 'status' field in orders can be: 'pending', 'shipped', 'completed', 'cancelled'
Important rules for your SQL queries:
1. Always use table aliases for clarity (e.g., 'o' for orders, 'c' for customers)
2. Include comments explaining complex logic
3. Use COALESCE to handle NULL values in aggregations
4. For percentage calculations, cast integers to DECIMAL to avoid truncation
5. Always add LIMIT clauses for queries that could return many rows (default to LIMIT 100)
6. When asked for "recent" data, default to the last 30 days unless specified otherwise
Before executing any query, verify:
- All referenced tables and columns exist in the schema
- JOIN conditions use the correct foreign keys
- Aggregation functions are used appropriately
Format your final answer as:
1. A brief explanation of what the query does
2. The results presented clearly, using bullet points or a table format when appropriate
3. Any caveats or assumptions made"""
# Create the prompt template
prompt = ChatPromptTemplate.from_messages([
("system", custom_system_prompt),
("system", "You have access to the following tools: {tools}"),
("system", "Use the following format:\nQuestion: the input question\nThought: your reasoning\nAction: the tool to use\nAction Input: the tool input\nObservation: the result\n... (repeat as needed)\nThought: final reasoning\nFinal Answer: your response"),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# Create agent with custom prompt
agent_executor_custom = create_sql_agent(
llm=llm,
toolkit=toolkit,
agent_type=AgentType.OPENAI_FUNCTIONS,
prompt=prompt,
verbose=True,
max_iterations=10,
)Adding Custom Tools
You can extend the agent's capabilities by creating custom tools. For instance, adding a visualization tool or a data export function.
from langchain.tools import Tool, tool
from typing import Optional, Dict, Any
import pandas as pd
import json
@tool
def get_table_stats(table_name: str) -> str:
"""Get statistical summary (row count, min/max values, null counts) for a specific table.
Use this to understand data distribution before writing queries."""
query = f"""
SELECT
COUNT(*) as row_count,
(SELECT COUNT(*) FROM {table_name}) as total_rows
FROM {table_name}
LIMIT 1
"""
try:
result = db.run(query)
return f"Table '{table_name}' statistics: {result}"
except Exception as e:
return f"Error getting stats for {table_name}: {str(e)}"
@tool
def export_query_to_csv(query: str, filename: str = "output.csv") -> str:
"""Execute a SQL query and export results to a CSV file.
Input should be a JSON with 'query' and optional 'filename' fields."""
try:
# Parse the input - it may come as a JSON string
import json
params = json.loads(query) if isinstance(query, str) and query.startswith('{') else {"query": query}
actual_query = params.get("query", query)
output_file = params.get("filename", "output.csv")
# Execute query using pandas for CSV export
df = pd.read_sql(actual_query, db._engine)
df.to_csv(output_file, index=False)
return f"Successfully exported {len(df)} rows to {output_file}"
except Exception as e:
return f"Export failed: {str(e)}"
# Combine standard SQL tools with custom tools
all_tools = toolkit.get_tools() + [get_table_stats, export_query_to_csv]
# Create agent with extended toolset
from langchain.agents import create_openai_tools_agent
from langchain.agents import AgentExecutor
agent_with_custom_tools = create_openai_tools_agent(llm, all_tools, prompt)
agent_executor_extended = AgentExecutor(
agent=agent_with_custom_tools,
tools=all_tools,
verbose=True,
max_iterations=12,
handle_parsing_errors=True
)
# Test custom tools
result = agent_executor_extended.invoke({
"input": "Give me a statistical overview of the orders table, then export all completed orders to 'completed_orders.csv'"
})
print(result["output"])Implementing Query Validation and Guardrails
Production systems must prevent destructive queries and enforce security policies. Implement a validation layer that intercepts queries before execution.
import re
from typing import List
class SQLGuardrails:
"""Security and validation layer for SQL queries."""
FORBIDDEN_KEYWORDS = [
'DROP', 'DELETE', 'TRUNCATE', 'ALTER',
'CREATE', 'INSERT', 'UPDATE', 'GRANT', 'REVOKE'
]
ALLOWED_TABLES = ['customers', 'products', 'orders', 'order_items']
MAX_RESULT_SIZE = 10000
@staticmethod
def validate_query(query: str) -> tuple[bool, str]:
"""
Validate a SQL query for safety and policy compliance.
Returns (is_valid, reason)
"""
query_upper = query.upper().strip()
# Check for forbidden operations
for keyword in SQLGuardrails.FORBIDDEN_KEYWORDS:
pattern = r'\b' + keyword + r'\b'
if re.search(pattern, query_upper):
# Allow SELECT INTO or CREATE TABLE AS SELECT for exports
if keyword == 'CREATE' and 'CREATE TABLE' not in query_upper:
return False, f"Query contains forbidden operation: {keyword}"
elif keyword == 'CREATE':
continue
# Check table references are in allowed list
table_pattern = r'(?:FROM|JOIN)\s+(\w+)'
referenced_tables = re.findall(table_pattern, query_upper, re.IGNORECASE)
for table in referenced_tables:
if table.lower() not in [t.lower() for t in SQLGuardrails.ALLOWED_TABLES]:
return False, f"Query references unauthorized table: {table}"
# Check for excessive result sizes
if 'LIMIT' not in query_upper and 'SELECT' in query_upper:
return False, "SELECT query must include a LIMIT clause"
return True, "Query validated successfully"
# Integrate guardrails into a custom query execution tool
from langchain.tools import Tool
def safe_query_executor(query: str) -> str:
"""Execute a SQL query with safety validation."""
# Validate the query
is_valid, reason = SQLGuardrails.validate_query(query)
if not is_valid:
return f"QUERY REJECTED: {reason}\nPlease revise your query to comply with safety policies."
# Execute the validated query
try:
result = db.run(query)
return result
except Exception as e:
return f"Query execution error: {str(e)}"
# Replace the default query tool with the safe version
safe_query_tool = Tool(
name="sql_db_query",
func=safe_query_executor,
description="Execute a SQL query against the database after safety validation"
)
# Rebuild toolkit with safe query tool
safe_tools = [safe_query_tool] + [t for t in toolkit.get_tools() if t.name != "sql_db_query"]
print("SQL guardrails activated. Protected operations:", SQLGuardrails.FORBIDDEN_KEYWORDS)Best Practices for Production-Ready SQL Agents
1. Optimize LLM Selection for Cost and Performance
Not every query requires GPT-4. Implement a tiered model strategy: use a faster, cheaper model like GPT-3.5-turbo for simple schema lookups and routine queries, reserving GPT-4 for complex multi-join queries or when the cheaper model fails. This can reduce costs by 40-60% without sacrificing user experience.
from langchain_openai import ChatOpenAI
# Tiered LLM setup
fast_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, max_tokens=500)
premium_llm = ChatOpenAI(model="gpt-4-turbo", temperature=0, max_tokens=2000)
# Create agents with different models based on complexity
simple_agent = create_sql_agent(
llm=fast_llm,
toolkit=toolkit,
agent_type=AgentType.OPENAI_FUNCTIONS,
verbose=False,
max_iterations=5,
)
complex_agent = create_sql_agent(
llm=premium_llm,
toolkit=toolkit,
agent_type=AgentType.OPENAI_FUNCTIONS,
verbose=True,
max_iterations=10,
)
# Router function to select appropriate agent
def route_query(question: str) -> str:
"""Route simple queries to fast LLM, complex ones to premium."""
complexity_indicators = ['compare', 'trend', 'percentage', 'rank',
'multiple', 'across', 'correlation', 'cohort']
is_complex = any(indicator in question.lower() for indicator in complexity_indicators)
agent = complex_agent if is_complex else simple_agent
result = agent.invoke({"input": question})
return result["output"]2. Implement Comprehensive Error Handling
SQL agents can fail in multiple ways: invalid SQL, database connection issues, timeout errors, or LLM output parsing failures. Wrap your agent calls in robust error handling.
import time
from typing import Optional
from sqlalchemy.exc import OperationalError, ProgrammingError
class RobustSQLAgent:
def __init__(self, agent_executor, max_retries=3):
self.agent = agent_executor
self.max_retries = max_retries
def query(self, question: str, context: Optional[dict] = None) -> dict:
"""Execute a query with retry logic and comprehensive error handling."""
retries = 0
last_error = None
while retries <= self.max_retries:
try:
input_dict = {"input": question}
if context:
input_dict.update(context)
result = self.agent.invoke(input_dict)
# Check if the result indicates an unresolved error
output = result.get("output", "")
if "error" in output.lower() and retries < self.max_retries:
# Ask agent to retry with error context
input_dict["input"] = (
f"The previous attempt produced this error: {output}\n"
f"Please retry the original question: {question}\n"
f"Analyze the error and adjust your approach."
)
retries += 1
continue
return result
except OperationalError as e:
last_error = e
retries += 1
time.sleep(2 ** retries) # Exponential backoff
except ProgrammingError as e:
# SQL syntax error - feed back to agent
last_error = e
retries += 1
except Exception as e:
last_error = e
retries += 1
time.sleep(1)
return {
"output": f"Query failed after {self.max_retries} retries. "
f"Last error: {str(last_error)}. "
f"Please rephrase your question or contact support.",
"error": str(last_error)
}
# Usage
robust_agent = RobustSQLAgent(agent_executor)
result = robust_agent.query("Show me quarterly revenue trends by region")
print(result["output"])3. Cache Frequent Queries and Schema Information
Repeated schema lookups and common queries waste both time and API tokens. Implement caching to dramatically improve response times for returning users.
import hashlib
import json
from functools import lru_cache
from datetime import datetime, timedelta
class QueryCache:
def __init__(self, max_size=100, ttl_minutes=30):
self.cache = {}
self.max_size = max_size
self.ttl = timedelta(minutes=ttl_minutes)
def _generate_key(self, question: str) -> str:
"""Generate a deterministic cache key from the question."""
# Normalize the question for better cache hits
normalized = question.lower().strip()
return hashlib.md5(normalized.encode()).hexdigest()
def get(self, question: str) -> Optional[str]:
"""Retrieve cached result if valid."""
key = self._generate_key(question)
if key in self.cache:
entry = self.cache[key]
if datetime.now() - entry["timestamp"] < self.ttl:
return entry["result"]
else:
del self.cache[key]
return None
def set(self, question: str, result: str):
"""Store result in cache."""
key = self._generate_key(question)
# Evict oldest entry if cache is full
if len(self.cache) >= self.max_size:
oldest_key = min(self.cache.keys(),
key=lambda k: self.cache[k]["timestamp"])
del self.cache[oldest_key]
self.cache[key] = {
"result": result,
"timestamp": datetime.now()
}
# Schema caching with TTL
@lru_cache(maxsize=10)
def get_cached_schema(table_name: str, _ttl_hash: int = 0):
"""Cached schema retrieval. TTL hash invalidates cache periodically."""
return db.get_table_info([table_name])
# Integrate cache into agent workflow
query_cache = QueryCache()
def cached_agent_query(question: str) -> str:
"""Agent query with caching layer."""
# Check cache first
cached_result = query_cache.get(question)
if cached_result:
return cached_result + "\n\n[Cached result - for fresh data, ask again]"
# Execute agent query
result = agent_executor.invoke({"input": question})
output = result["output"]
# Cache the result
query_cache.set(question, output)
return output4. Monitor and Log Agent Performance
Instrument your agent to track query patterns, error rates, and token usage. This data is invaluable for optimizing prompts and identifying problematic query patterns.
import logging
import time
from datetime import datetime
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('sql_agent.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger('SQLAgent')
class MonitoredSQLAgent:
def __init__(self, agent_executor):
self.agent = agent_executor
self.metrics = {
"total_queries": 0,
"successful_queries": 0,
"failed_queries": 0,
"total_tokens": 0,
"total_execution_time": 0.0
}
def query(self, question: str, user_id: str = "anonymous") -> dict:
"""Execute query with full monitoring."""
start_time = time.time()
self.metrics["total_queries"] += 1
logger.info(f"Query from user {user_id}: {question[:100]}...")
try:
result = self.agent.invoke({"input": question})
execution_time = time.time() - start_time
self.metrics["successful_queries"] += 1
self.metrics["total_execution_time"] += execution_time
logger.info(f"Query completed in {execution_time:.2f}s - "
f"Result length: {len(result.get('output', ''))} chars")
# Store query analytics
self._log_query_analytics(question, execution_time, "success")
return result
except Exception as e:
self.metrics["failed_queries"] += 1
execution_time = time.time() - start_time
logger.error(f"Query failed after {execution_time:.2f}s: {str(e)}")
self._log_query_analytics(question, execution_time, "failed")
raise
def _log_query_analytics(self, question: str, duration: float, status: str):
"""Log structured analytics for later analysis."""
analytics = {
"timestamp": datetime.now().isoformat(),
"question_preview": question[:100],
"duration_seconds": duration,
"status": status,
"question_length": len(question)
}
logger.info(f"ANALYTICS: {json.dumps(analytics)}")
def get_metrics(self) -> dict:
"""Return current performance metrics."""
success_rate = (self.metrics["successful_queries"] /
max(self.metrics["total_queries"], 1)) * 100
avg_time = (self.metrics["total_execution_time"] /
max(self.metrics["total_queries"], 1))
return {
**self.metrics,
"success_rate_percent": round(success_rate, 1),
"average_execution_time": round(avg_time, 3)
}
# Usage
monitored_agent = MonitoredSQLAgent(agent_executor)
result = monitored_agent.query("What are our monthly sales totals for 2024?", user_id="analyst_42")
print(monitored_agent.get_metrics())5. Design for Conversation Continuity
When building user-facing applications, ensure the conversation context persists across sessions. Use a persistent memory backend rather than in-memory storage.
from langchain.memory import ConversationBufferMemory
from langchain.storage import LocalFileStore
from langchain.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
import pickle
class PersistentConversationManager:
def __init__(self, session_id: str, storage_path: str = "./conversations"):
self.session_id = session_id
self.storage_path = storage_path
self.memory = self._load_or_create_memory()
def _memory_file_path(self) -> str:
import os
os.makedirs(self.storage_path, exist_ok=True)
return os.path.join(self.storage_path, f"{self.session_id}_memory.pkl")
def _load_or_create_memory(self) -> ConversationBufferMemory:
"""Load existing conversation memory or create new one."""
file_path = self._memory_file_path()
try:
with open(file_path, 'rb') as f:
memory = pickle.load(f)
return memory
except FileNotFoundError:
return ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
max_token_limit=2000 # Prevent context overflow
)
def save(self):
"""Persist conversation memory to disk."""
file_path = self._memory_file_path()
with open(file_path, 'wb') as f:
pickle.dump(self.memory, f)
def add_interaction(self, question: str, answer: str):
"""Record a Q&A pair in conversation history."""