We all use chatbots like ChatGPT, Perplexity, and Claude. But wouldn’t it be cool if a chatbot could directly talk to your database? And keep your data secure at the same time? Let’s build one!
Here’s how we can build it, step by step:
Technologies We’ll Use:
- Frontend
- Next.js (React framework)
- Copilotkit (for interacting with sql agent)
- Backend
- Fastapi (Python backend framework)
- langgraph and langchain (for building sql agent)
We’ll create a SQL agent using LangGraph. This agent will connect to your database and allow you to chat with it in a structured way.
Once the agent is ready, we’ll integrate it into a Next.js app using Co Agents.
Instead of setting up everything from scratch, you can use my starter kit.
Building the sql agent
You need to edit the agent.py
in ./agent/my_agent
directory.
Add all the imports
```python
from langchain_community.utilities import SQLDatabase
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain.prompts import PromptTemplate
from langchain_groq import ChatGroq
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from pydantic import BaseModel, Field
from typing import Literal, Annotated, List
from typing_extensions import TypedDict
from dotenv import load_dotenv
```
You can load environment variables using:
```py
load_dotenv()
```
We need a SQLite database for the SQL agent. For this example, I’m using the Chinook db.
We will use SQLDatabase to create an engine for querying the database and retrieving its context.
```py
db = SQLDatabase.from_uri(r"sqlite:///data/data.db")
def db_query_tool(query: str) -> str:
result = db.run_no_throw(query)
if not result:
return "Error: Query failed. Please rewrite your query and try again."
return result
def get_context() -> str:
return db.get_table_info()
```
We need an LLM to interact with users. Here, we’re using the Llama 3.3 model via Groq.
```py
llm = ChatGroq(model="llama-3.3-70b-versatile")
```
Now comes the fun part - creating AgentState, edges and nodes for our LangGraph agent.
```py
class AgentState(TypedDict):
"""
Attributes:
messages (List[BaseMessage]): List of messages exchanged during the interaction.
user_query (str): The user-provided query.
sql_query (str): The generated SQL query.
query_results (str): Results from executing the SQL query.
"""
messages: Annotated[List[BaseMessage], add_messages]
user_query: str
sql_query: str
query_results: str
```
We need to verify if the user's query is relevant to the database.
We can use structured outputs to obtain relevance from LLM using the Pydantic model by using withured_outputs()
method.
```py
def check_relevance(state: AgentState) -> Literal["YES", "NO"]:
class QueryRelevance(BaseModel):
query_relevant: str = Field(
description="Relevance of user query to database context. The response must be 'YES' or 'NO'."
)
relevance_prompt = PromptTemplate.from_template(
"""
You are a relevance evaluation assistant. Your task is to determine if a user's query is relevant to a given database context.
Respond with 'YES' if the query is relevant, and 'NO' if it is not.
Database Context: {context}.
Example Queries (may not be related to context):
1. "What were the total sales in 2010?" -> YES
2. "Who is the CEO of the company?" -> NO
3. "Which sales agent had the highest sales in 2009?" -> YES
4. "What is the weather today?" -> NO
Now evaluate the following query and respond with 'YES' or 'NO':
{query}
"""
)
model = llm.with_structured_output(QueryRelevance)
relevance_generator = relevance_prompt | model
inputs = {"context": get_context(), "query": state["messages"][-1].content}
relevance = relevance_generator.invoke(inputs)
return relevance.query_relevant
```
Basically we attach prompt to the model and invoke the chain with our inputs (query - user query, context - database schema info).
Now, let’s define the nodes for query execution, answer generation, and handling irrelevant queries.
```py
def sql_query_execution(state: AgentState) -> AgentState:
"""
Executes the SQL query and retrieves the results.
Args:
state (AgentState): The current state of the agent.
Returns:
AgentState: Updated state with the SQL query results.
"""
state["query_results"] = db_query_tool(state["sql_query"])
return state
def answer_generation(state: AgentState) -> AgentState:
"""
Generates a natural language response based on the user query and SQL query results.
Args:
state (AgentState): The current state of the agent.
Returns:
AgentState: Updated state with the generated response.
"""
answer_generation_prompt = PromptTemplate.from_template(
"""
You are a highly skilled assistant responsible for generating user-friendly answers based on a user query and the SQL query's output.
You will use the provided SQL output to craft a clear, concise, and helpful response that directly answers the user's query.
User Query: {query}
SQL Query Output:
{sql_output}
Instructions:
- Analyze the SQL query output.
- Generate a response that answers the user query in natural language.
- Be specific and provide numerical values or details where applicable.
Output:
Provide only the answer to the user query in plain text, without any additional formatting or explanation.
"""
)
inputs = {
"query": state["user_query"],
"sql_output": state["query_results"],
}
answer_generator = answer_generation_prompt | llm
return {**state, "messages": state["messages"] + [answer_generator.invoke(inputs)]}
def query_not_relevant(state: AgentState) -> AgentState:
"""
Handles cases where the user's query is not relevant to the database context.
Args:
state (AgentState): The current state of the agent.
Returns:
AgentState: Updated state with a message indicating irrelevance.
"""
return {
**state,
"messages": state["messages"]
+ [AIMessage(content="Given user query is not relevant to database context.")],
}
```
Now, let's create a workflow that compiles into the agent.
```py
workflow = StateGraph(AgentState)
workflow.add_node("sql_query_generation", sql_query_generation)
workflow.add_node("sql_query_execution", sql_query_execution)
workflow.add_node("answer_generation", answer_generation)
workflow.add_node("query_not_relevant", query_not_relevant)
workflow.add_conditional_edges(
START, check_relevance, {"YES": "sql_query_generation", "NO": "query_not_relevant"}
)
workflow.add_edge("query_not_relevant", END)
workflow.add_edge("sql_query_generation", "sql_query_execution")
workflow.add_edge("sql_query_execution", "answer_generation")
workflow.add_edge("answer_generation", END)
```
Finally, lets compile the workflow with memory.
```py
graph = workflow.compile(checkpointer=MemorySaver())
```
Let's run this agent.
```bash
# in terminal
poetry install
poetry run demo
```
UI
In the starter kit, the /copilotkit
route is already set up to connect the agent to our Next.js app.
On the homepage(/), I’ve just rendered a heading and the chatbot provided by CopilotKit with some custom colors.
Let's run this app
```bash
pnpm i
pnpm dev
```
You can also add your own custom chatbot UI, as CopilotKit provides the useCopilotChat()
hook.
SQL Agent Chatbot is created!
More Features
You can add more features to this app, such as:
- Changing the database via the ui
- Adding animations to the ui
- Displaying sql query execution results to the user
Feel free to add anything you can imagine—there are no limits!
Here’s the source code for this project: kom-senapati/DataChat
Wrapping it up
CopilotKit is an innovative tool for adding AI copilots into your project. You can connect these copilots to deployed LangGraph ai agents.
LangGraph. is a python library that allows you to create stateful ai agents as graphs and supports multi-agent workflows.
We should use these technologies to keep up with the rapidly changing landscape of the software industry.
Dont forget to start my project DataChat, CopilotKit, LangGraph ⭐