sqlmind

LangGraph-native NL→SQL agent library — drop your database query pipeline directly into an existing agent graph as a composable node.

---

Why sqlmind?

Most NL→SQL tools are standalone apps or ChatGPT wrappers. sqlmind is a library — designed to be dropped into your existing LangGraph agent as a composable node. No hosting, no vendor lock-in, no new service to manage.

Feature	sqlmind	Most alternatives
LangGraph-native nodes	✅ First-class	❌ Standalone only
Pluggable guardrails	✅ Full interface	❌ Fixed policies
Semantic glossary	✅ YAML/dict/custom	❌ None
Canonical metric defs	✅ Injected into LLM	❌ LLM guesses
Self-correction loop	✅ Up to N retries	❌ One-shot
Read-only by default	✅ Session-level	⚠️ Convention only
Any LLM	✅ Protocol + adapters	❌ OpenAI only

---

Installation

# Core + PostgreSQL + LangGraph/LangChain
pip install sqlmind[langgraph,postgres]

# Add OpenAI or Anthropic direct adapters
pip install sqlmind[openai]
pip install sqlmind[anthropic]

# Everything
pip install sqlmind[all]

---

Quick Start

Shape A — Single drop-in node

from sqlmind import SQLAgent, SQLAgentConfig
from sqlmind.core import PostgresConnection
from sqlmind.guardrails import ReadOnlyGuardrail, RowLimitGuardrail, BlockedColumnsGuardrail
from sqlmind.semantic import Glossary, MetricRegistry
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

# 1. Connect (read-only enforced at DB session level)
conn = PostgresConnection(url="postgresql://user:pass@host/mydb")

# 2. Semantic layer — eliminates LLM guessing about your business terms
glossary = Glossary.from_yaml("glossary.yaml")
metrics = MetricRegistry()
metrics.define("mrr", "Monthly Recurring Revenue", formula="SUM(amount) WHERE cycle='monthly'", unit="USD")

# 3. Guardrails — fail closed, explainable violations
guardrails = [
    ReadOnlyGuardrail(),                              # Block INSERT/UPDATE/DELETE/DROP
    RowLimitGuardrail(max_rows=500),                  # Inject LIMIT clause
    BlockedColumnsGuardrail(columns=["ssn", "password_hash"]),  # PII protection
]

# 4. Build agent — any LLM works
agent = SQLAgent(
    connection=conn,
    llm=ChatOpenAI(model="gpt-4o"),    # or OpenAIAdapter, AnthropicAdapter, or any custom LLM
    guardrails=guardrails,
    glossary=glossary,
    metrics=metrics,
)

# 5. Drop into your LangGraph graph — ONE LINE
graph = StateGraph(dict)
graph.add_node("db", agent.as_node())
graph.set_entry_point("db")
graph.add_edge("db", END)
app = graph.compile()

result = app.invoke({"user_query": "How many active users signed up last month?"})
# → {"success": True, "sql_result": [...], "row_count": 1, "sql_executed": "SELECT ...", ...}

Shape B — Exploded sub-graph (power users)

Individual nodes for full control flow — route to human-in-the-loop on low confidence, add custom steps between pipeline stages.

from sqlmind.integrations import clarify_node, generate_node, validate_node, execute_node

graph = StateGraph(MyState)
graph.add_node("clarify",  clarify_node(agent))
graph.add_node("generate", generate_node(agent))
graph.add_node("validate", validate_node(agent))
graph.add_node("execute",  execute_node(agent))

# Your own routing
def route_after_clarify(state):
    return "human_in_loop" if state.get("needs_clarification") else "generate"

graph.add_conditional_edges("clarify", route_after_clarify, {...})

Use outside LangGraph

result = agent.query("Show me top 10 customers by revenue this quarter")
print(result["sql"])         # The executed SQL
print(result["rows"])        # Query result as list of dicts
print(result["explanation"]) # Human-readable explanation

---

Custom LLM Integration

Any LLM works. Three ways:

# Option 1: Any LangChain model (ChatOpenAI, ChatAnthropic, ChatOllama, etc.)
from langchain_openai import ChatOpenAI
agent = SQLAgent(connection=conn, llm=ChatOpenAI(model="gpt-4o"))

# Option 2: Direct OpenAI (no LangChain dep)
from sqlmind.llm import OpenAIAdapter
agent = SQLAgent(connection=conn, llm=OpenAIAdapter(api_key="sk-...", model="gpt-4o"))

# Option 3: Direct Anthropic
from sqlmind.llm import AnthropicAdapter
agent = SQLAgent(connection=conn, llm=AnthropicAdapter(model="claude-3-5-sonnet-20241022"))

# Option 4: Fully custom — implement 2 methods
class MyCustomLLM:
    def complete(self, messages, **kwargs):
        ...  # return LLMResponse(content="...")
    def complete_structured(self, messages, schema, **kwargs):
        ...  # return dict

agent = SQLAgent(connection=conn, llm=MyCustomLLM())

---

Custom Guardrails

from sqlmind.guardrails import Guardrail, GuardrailResult
import datetime

class NoDeleteOnWeekendsGuardrail(Guardrail):
    def check(self, query, context):
        is_weekend = datetime.date.today().weekday() >= 5
        has_delete = "DELETE" in query.sql.upper()
        if is_weekend and has_delete:
            return GuardrailResult(
                passed=False,
                guardrail_name=self.name,
                reason="DELETE queries are blocked on weekends per data policy.",
            )
        return GuardrailResult(passed=True, guardrail_name=self.name)

agent = SQLAgent(
    connection=conn,
    llm=llm,
    guardrails=[ReadOnlyGuardrail(), NoDeleteOnWeekendsGuardrail()],
)

---

Glossary YAML Format

# glossary.yaml
amt_cents:
  description: "Revenue in USD cents. Always divide by 100 before display."
  example: "SELECT SUM(amt_cents) / 100.0 AS revenue_usd FROM orders"
  tags: [finance, revenue]

status:
  description: "User status: 1=active, 2=trial, 3=churned, 4=suspended"
  tags: [user, lifecycle]

# Shorthand
mrr: "Monthly Recurring Revenue. See metric definitions for formula."

---

Configuration

from sqlmind import SQLAgentConfig

config = SQLAgentConfig(
    max_retries=3,                # Self-correction attempts on execution error
    confidence_threshold=0.65,    # Below this → trigger clarification step
    max_rows=1000,                # Hard row cap
    query_timeout_seconds=30,     # Wall-clock execution timeout
    memory_window=10,             # Multi-turn context turns kept
    input_state_key="user_query", # LangGraph state key for the question
    output_state_key="sql_result",# LangGraph state key for the result
    verbose=False,                # Debug logging
    llm_temperature=0.0,          # Deterministic generation
)

---

LangGraph State Contract

The sqlmind node reads and writes these keys on the shared LangGraph state:

Key	Direction	Type	Description
user_query	Read	str	The NL question
sql_result	Write	list[dict]	Query result rows
sql_executed	Write	str	Final SQL (after any guardrail rewrites)
row_count	Write	int	Number of rows returned
success	Write	bool	Whether execution succeeded
error	Write	str|None	Error message if failed
clarification_question	Write	str|None	Set if question was ambiguous
sql_confidence	Write	float	Model confidence 0.0–1.0
execution_attempts	Write	int	Number of self-correction attempts

---

Architecture

sqlmind/
├── core/           # DB connection + schema introspection + embedding index
├── semantic/       # Glossary (YAML/dict) + MetricRegistry
├── agent/          # clarify → generate → validate → execute → memory
├── guardrails/     # Abstract interface + 5 built-ins + custom registry
├── integrations/   # langgraph_node.py (as_node + exploded nodes)
└── llm/            # LLMProvider protocol + LangChain/OpenAI/Anthropic adapters

---

Running Tests

pip install sqlmind[dev]
pytest tests/ -v

---

License

MIT