Closed jyoti194 closed 5 months ago
🤖
Hello @jyoti194, good to see you again!
To modify the LangChain code so that it returns "No results found" when the SQLResult is empty, you can add a conditional check after the SQL query execution. If the result is empty, you can return "No results found" instead of passing the result to the language model for generating the final answer.
In your QuerySQLDatabaseTool2
class, you can modify the _run
method like this:
class QuerySQLDatabaseTool2(QuerySQLDataBaseTool):
def _run(
self,
query: str,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
result = self.db.run_no_throw(query)
if not result:
return "No results found"
return next(split_text(result, chunk_size=14_000))
This modification will return "No results found" if the SQL query result is empty, and it will not pass the empty result to the language model, thus preventing it from hallucinating an answer.
Please note that this modification assumes that the run_no_throw
method of the db
object returns an empty string or None
when the SQL query result is empty. If this is not the case, you may need to adjust the condition if not result:
accordingly.
Also, please be aware that this modification will affect all SQL queries executed by the QuerySQLDatabaseTool2
tool, not just the specific query you mentioned. If you want to apply this behavior only to certain queries, you may need to add additional logic to determine when to return "No results found".
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.
Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
If you want to continue the conversation, start your reply with @dosu-bot.
🤖
Hello @jyoti194, good to see you again!
To modify the LangChain code so that it returns "No results found" when the SQLResult is empty, you can add a conditional check after the SQL query execution. This check will verify if the SQLResult is empty and if so, it will set the Answer to "No results found".
In your QuerySQLDatabaseTool2
class, you can modify the _run
method to check if the result is empty. If it is, return "No results found". Here's how you can do it:
class QuerySQLDatabaseTool2(QuerySQLDatabaseTool):
def _run(
self,
query: str,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
result = self.db.run_no_throw(query)
if not result:
return "No results found"
return next(split_text(result, chunk_size=14_000))
This modification assumes that the run_no_throw
method of the db
object returns an empty value (like an empty string or None) when there are no results. If it doesn't, you might need to adjust the condition if not result:
to match the actual behavior of your database engine.
Please note that this change will affect all queries run by the QuerySQLDatabaseTool2
tool. If you want to limit this behavior to certain queries, you might need to add additional logic to the _run
method.
I hope this helps! Let me know if you have any other questions.
hi @dosubot this is my code
Issue you'd like to raise. Entering new SQLDatabaseChain chain... what is Gesh D desigation SQLQuery:SELECT [Designation] FROM [EGV_emp_departments_ChatGPT] WHERE [EmployeeName] = 'Gesh D' SQLResult: Answer:Final answer here
Finished chain. Final answer here
below is my code
import sqlalchemy as sal import os, sys, openai import constants from langchain.sql_database import SQLDatabase from langchain.llms.openai import OpenAI from langchain_experimental.sql import SQLDatabaseChain from sqlalchemy import create_engine
from langchain.chat_models import ChatOpenAI from typing import List, Optional
from langchain.agents.agent_toolkits import SQLDatabaseToolkit from langchain.callbacks.manager import CallbackManagerForToolRun from langchain.chat_models import ChatOpenAI from langchain_experimental.plan_and_execute import ( PlanAndExecute, load_agent_executor, load_chat_planner, ) from langchain.sql_database import SQLDatabase from langchain.text_splitter import TokenTextSplitter from langchain.tools import BaseTool from langchain.tools.sql_database.tool import QuerySQLDataBaseTool from secret_key import openapi_key
os.environ['OPENAI_API_KEY'] = openapi_key
def chat(question): from urllib.parse import quote_plus server_name = constants.server_name database_name = constants.database_name username = constants.username password = constants.password encoded_password = quote_plus(password) connection_uri = f"mssql+pyodbc://{username}:{encoded_password}@{server_name}/{database_name}?driver=ODBC+Driver+17+for+SQL+Server"
engine = create_engine(connection_uri) model_name="gpt-3.5-turbo-16k" db = SQLDatabase(engine, view_support=True, include_tables=['EGV_emp_departments_ChatGPT']) llm = ChatOpenAI(temperature=0, verbose=False, model=model_name) db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True)
PROMPT = """
Given an input question, first create a syntactically correct mssql query to run,
then look at the results of the query and return the answer.
The question: {db_chain.run}
"""
return db_chain.run(question) answer=chat("what is Gesh D desigation") print(answer)
and i have 3 files in C:\Python311\Lib\site-packages\langchain_experimental\sql directory base.py prompt.py vector_sql.py
🤖
Based on the context provided, you can modify the run
method in the SQLDatabaseChain
class to return "No results found" when the SQLResult is empty.
Here's how you can do it:
def run(self, command: str, fetch: str = 'all') -> str:
try:
result = self._execute(command, fetch)
if not result:
return "No results found"
else:
return self._format_result(result, fetch)
except Exception as e:
return str(e)
In this modified run
method, we first try to execute the SQL command. If the result is empty (i.e., not result
is True
), we return "No results found". Otherwise, we format and return the result as before.
Please note that you need to make this change in the SQLDatabaseChain
class in the base.py
file located in the C:\Python311\Lib\site-packages\langchain_experimental\sql
directory.
Also, remember to import the Exception
class at the beginning of your base.py
file if it's not already imported:
from exceptions import Exception
This solution is based on the information provided in the Fix hallucination problem with SQLDatabaseChain
issue in the LangChain repository. The issue suggests adding a prompt to handle empty SQLResult
fields, which is similar to your requirement.
Please try this solution and let me know if it works for you or if you need further assistance.
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.
Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
If you want to continue the conversation, start your reply with @dosu-bot.
this is code for base.py file, there is no run method
"""Chain for interacting with SQL Database.""" from future import annotations
import warnings from typing import Any, Dict, List, Optional
from langchain.callbacks.manager import CallbackManagerForChainRun from langchain.chains.base import Chain from langchain.chains.llm import LLMChain from langchain.chains.sql_database.prompt import DECIDER_PROMPT, PROMPT, SQL_PROMPTS from langchain.prompts.prompt import PromptTemplate from langchain.schema import BasePromptTemplate from langchain.schema.language_model import BaseLanguageModel from langchain.tools.sql_database.prompt import QUERY_CHECKER from langchain.utilities.sql_database import SQLDatabase
from langchain_experimental.pydantic_v1 import Extra, Field, root_validator
INTERMEDIATE_STEPS_KEY = "intermediate_steps"
class SQLDatabaseChain(Chain): """Chain for interacting with SQL Database.
Example:
.. code-block:: python
from langchain_experimental.sql import SQLDatabaseChain
from langchain.llms import OpenAI, SQLDatabase
db = SQLDatabase(...)
db_chain = SQLDatabaseChain.from_llm(OpenAI(), db)
*Security note*: Make sure that the database connection uses credentials
that are narrowly-scoped to only include the permissions this chain needs.
Failure to do so may result in data corruption or loss, since this chain may
attempt commands like `DROP TABLE` or `INSERT` if appropriately prompted.
The best way to guard against such negative outcomes is to (as appropriate)
limit the permissions granted to the credentials used with this chain.
This issue shows an example negative outcome if these steps are not taken:
https://github.com/langchain-ai/langchain/issues/5923
"""
llm_chain: LLMChain
llm: Optional[BaseLanguageModel] = None
"""[Deprecated] LLM wrapper to use."""
database: SQLDatabase = Field(exclude=True)
"""SQL Database to connect to."""
prompt: Optional[BasePromptTemplate] = None
"""[Deprecated] Prompt to use to translate natural language to SQL."""
top_k: int = 5
"""Number of results to return from the query"""
input_key: str = "query" #: :meta private:
output_key: str = "result" #: :meta private:
return_sql: bool = False
"""Will return sql-command directly without executing it"""
return_intermediate_steps: bool = False
"""Whether or not to return the intermediate steps along with the final answer."""
return_direct: bool = False
"""Whether or not to return the result of querying the SQL table directly."""
use_query_checker: bool = False
"""Whether or not the query checker tool should be used to attempt
to fix the initial SQL from the LLM."""
query_checker_prompt: Optional[BasePromptTemplate] = None
"""The prompt template that should be used by the query checker"""
class Config:
"""Configuration for this pydantic object."""
extra = Extra.forbid
arbitrary_types_allowed = True
@root_validator(pre=True)
def raise_deprecation(cls, values: Dict) -> Dict:
if "llm" in values:
warnings.warn(
"Directly instantiating an SQLDatabaseChain with an llm is deprecated. "
"Please instantiate with llm_chain argument or using the from_llm "
"class method."
)
if "llm_chain" not in values and values["llm"] is not None:
database = values["database"]
prompt = values.get("prompt") or SQL_PROMPTS.get(
database.dialect, PROMPT
)
values["llm_chain"] = LLMChain(llm=values["llm"], prompt=prompt)
return values
@property
def input_keys(self) -> List[str]:
"""Return the singular input key.
:meta private:
"""
return [self.input_key]
@property
def output_keys(self) -> List[str]:
"""Return the singular output key.
:meta private:
"""
if not self.return_intermediate_steps:
return [self.output_key]
else:
return [self.output_key, INTERMEDIATE_STEPS_KEY]
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
_run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
input_text = f"{inputs[self.input_key]}\nSQLQuery:"
_run_manager.on_text(input_text, verbose=self.verbose)
# If not present, then defaults to None which is all tables.
table_names_to_use = inputs.get("table_names_to_use")
table_info = self.database.get_table_info(table_names=table_names_to_use)
llm_inputs = {
"input": input_text,
"top_k": str(self.top_k),
"dialect": self.database.dialect,
"table_info": table_info,
"stop": ["\nSQLResult:"],
}
if self.memory is not None:
for k in self.memory.memory_variables:
llm_inputs[k] = inputs[k]
intermediate_steps: List = []
try:
intermediate_steps.append(llm_inputs.copy()) # input: sql generation
sql_cmd = self.llm_chain.predict(
callbacks=_run_manager.get_child(),
**llm_inputs,
).strip()
if self.return_sql:
return {self.output_key: sql_cmd}
if not self.use_query_checker:
_run_manager.on_text(sql_cmd, color="green", verbose=self.verbose)
intermediate_steps.append(
sql_cmd
) # output: sql generation (no checker)
intermediate_steps.append({"sql_cmd": sql_cmd}) # input: sql exec
result = self.database.run(sql_cmd)
intermediate_steps.append(str(result)) # output: sql exec
else:
query_checker_prompt = self.query_checker_prompt or PromptTemplate(
template=QUERY_CHECKER, input_variables=["query", "dialect"]
)
query_checker_chain = LLMChain(
llm=self.llm_chain.llm, prompt=query_checker_prompt
)
query_checker_inputs = {
"query": sql_cmd,
"dialect": self.database.dialect,
}
checked_sql_command: str = query_checker_chain.predict(
callbacks=_run_manager.get_child(), **query_checker_inputs
).strip()
intermediate_steps.append(
checked_sql_command
) # output: sql generation (checker)
_run_manager.on_text(
checked_sql_command, color="green", verbose=self.verbose
)
intermediate_steps.append(
{"sql_cmd": checked_sql_command}
) # input: sql exec
result = self.database.run(checked_sql_command)
intermediate_steps.append(str(result)) # output: sql exec
sql_cmd = checked_sql_command
_run_manager.on_text("\nSQLResult: ", verbose=self.verbose)
_run_manager.on_text(result, color="yellow", verbose=self.verbose)
# If return direct, we just set the final result equal to
# the result of the sql query result, otherwise try to get a human readable
# final answer
if self.return_direct:
final_result = result
else:
_run_manager.on_text("\nAnswer:", verbose=self.verbose)
input_text += f"{sql_cmd}\nSQLResult: {result}\nAnswer:"
llm_inputs["input"] = input_text
intermediate_steps.append(llm_inputs.copy()) # input: final answer
final_result = self.llm_chain.predict(
callbacks=_run_manager.get_child(),
**llm_inputs,
).strip()
intermediate_steps.append(final_result) # output: final answer
_run_manager.on_text(final_result, color="green", verbose=self.verbose)
chain_result: Dict[str, Any] = {self.output_key: final_result}
if self.return_intermediate_steps:
chain_result[INTERMEDIATE_STEPS_KEY] = intermediate_steps
return chain_result
except Exception as exc:
# Append intermediate steps to exception, to aid in logging and later
# improvement of few shot prompt seeds
exc.intermediate_steps = intermediate_steps # type: ignore
raise exc
@property
def _chain_type(self) -> str:
return "sql_database_chain"
@classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
db: SQLDatabase,
prompt: Optional[BasePromptTemplate] = None,
**kwargs: Any,
) -> SQLDatabaseChain:
"""Create a SQLDatabaseChain from an LLM and a database connection.
*Security note*: Make sure that the database connection uses credentials
that are narrowly-scoped to only include the permissions this chain needs.
Failure to do so may result in data corruption or loss, since this chain may
attempt commands like `DROP TABLE` or `INSERT` if appropriately prompted.
The best way to guard against such negative outcomes is to (as appropriate)
limit the permissions granted to the credentials used with this chain.
This issue shows an example negative outcome if these steps are not taken:
https://github.com/langchain-ai/langchain/issues/5923
"""
prompt = prompt or SQL_PROMPTS.get(db.dialect, PROMPT)
llm_chain = LLMChain(llm=llm, prompt=prompt)
return cls(llm_chain=llm_chain, database=db, **kwargs)
class SQLDatabaseSequentialChain(Chain): """Chain for querying SQL database that is a sequential chain.
The chain is as follows:
1. Based on the query, determine which tables to use.
2. Based on those tables, call the normal SQL database chain.
This is useful in cases where the number of tables in the database is large.
"""
decider_chain: LLMChain
sql_chain: SQLDatabaseChain
input_key: str = "query" #: :meta private:
output_key: str = "result" #: :meta private:
return_intermediate_steps: bool = False
@classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
db: SQLDatabase,
query_prompt: BasePromptTemplate = PROMPT,
decider_prompt: BasePromptTemplate = DECIDER_PROMPT,
**kwargs: Any,
) -> SQLDatabaseSequentialChain:
"""Load the necessary chains."""
sql_chain = SQLDatabaseChain.from_llm(llm, db, prompt=query_prompt, **kwargs)
decider_chain = LLMChain(
llm=llm, prompt=decider_prompt, output_key="table_names"
)
return cls(sql_chain=sql_chain, decider_chain=decider_chain, **kwargs)
@property
def input_keys(self) -> List[str]:
"""Return the singular input key.
:meta private:
"""
return [self.input_key]
@property
def output_keys(self) -> List[str]:
"""Return the singular output key.
:meta private:
"""
if not self.return_intermediate_steps:
return [self.output_key]
else:
return [self.output_key, INTERMEDIATE_STEPS_KEY]
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
_run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
_table_names = self.sql_chain.database.get_usable_table_names()
table_names = ", ".join(_table_names)
llm_inputs = {
"query": inputs[self.input_key],
"table_names": table_names,
}
_lowercased_table_names = [name.lower() for name in _table_names]
table_names_from_chain = self.decider_chain.predict_and_parse(**llm_inputs)
table_names_to_use = [
name
for name in table_names_from_chain
if name.lower() in _lowercased_table_names
]
_run_manager.on_text("Table names to use:", end="\n", verbose=self.verbose)
_run_manager.on_text(
str(table_names_to_use), color="yellow", verbose=self.verbose
)
new_inputs = {
self.sql_chain.input_key: inputs[self.input_key],
"table_names_to_use": table_names_to_use,
}
return self.sql_chain(
new_inputs, callbacks=_run_manager.get_child(), return_only_outputs=True
)
@property
def _chain_type(self) -> str:
return "sql_database_sequential_chain"
Hi @dosubot, how to save graphs/plots in tools = [PythonREPLTool()] in langchain like save_chart in create_pandas_dataframe_agent.
is pythonREPLTool a preferable library to create graphs while interacting with sql db, to fetch answers in natural language?
🤖
Based on your request, you can modify the run
method in the SQLDatabaseChain
class to return "No results found" when the SQLResult is empty. Here's an example of how you can do it:
def run(self, command: str) -> str:
"""Run a command against the database."""
sql_result = self._run_sql_command(command)
if not sql_result:
return "No results found"
return self._format_sql_result(sql_result)
In this code, self._run_sql_command(command)
executes the SQL command and returns the result. If the result is empty (i.e., not sql_result
is True
), the method returns the string "No results found". Otherwise, it formats the SQL result and returns it.
Regarding your question about saving graphs/plots in PythonREPLTool, the LangChain library does not currently support this feature directly. However, you can use matplotlib's savefig
function to save the plot to a file. Here's an example:
import matplotlib.pyplot as plt
# Assume you have a pandas DataFrame df
df.plot(kind='bar')
# Save the plot to a file
plt.savefig('output.png')
This code creates a bar plot from a pandas DataFrame df
and saves it to a file named 'output.png'. Please note that this is a general Python solution and not specific to LangChain. You would need to integrate this into your LangChain application accordingly.
For more information, you can refer to the following sources:
Please let me know if you need further assistance.
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.
Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
If you want to continue the conversation, start your reply with @dosu-bot.
Hi @dosubot, here is my code to generate plots using PythonREPLToo, here how to save the plot?
db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True, return_direct=True)
tools = [PythonREPLTool()]
description = (
"Useful for when you need to answer questions about employess. "
"You must not input SQL. Use this more than the Python tool if the question "
"is about employee data, like 'how many employees are there?' or 'count the number of department'"
)
repl_tool = Tool(
name="Data", # We'll just call it 'Data'
func=db_chain.run,
description=description,
)
tools.append(repl_tool)
prefix = "Fulfill the following request as best you can. You have access to the following tools:"
suffix = (
"Begin! When looking for data, do not write a SQL query. "
"Pass the relevant portion of the request directly to the Data tool in its entirety."
"\n\n"
"Request: {input}\n"
"{agent_scratchpad}"
)
# The agent's prompt is built with the list of tools, prefix, suffix, and input variables
prompt = ZeroShotAgent.create_prompt(
tools, prefix=prefix, suffix=suffix, input_variables=["input", "agent_scratchpad"]
)
# Set up the llm_chain
llm_chain = LLMChain(llm=llm, prompt=prompt)
# Specify the tools the agent may use
tool_names = [tool.name for tool in tools]
agent = ZeroShotAgent(llm_chain=llm_chain, allowed_tools=tool_names)
# Create the AgentExecutor
agent_executor = AgentExecutor.from_agent_and_tools(
agent=agent, tools=tools, verbose=True, handle_parsing_errors=True
)
answer = agent_executor.run(question)
print(answer)
i have got to know about a different library as pandasai, import pandas as pd from pandasai import PandasAI from pandasai.llm.openai import OpenAI from pandasai.llm.open_assistant import OpenAssistant OPENAI_API_KEY = "sk-xxxx" llm = OpenAI(api_token=OPENAI_API_KEY) pandas_ai = PandasAI(llm) pandas_ai.run(df, prompt='plot bar graph for what are the 5 happiest countries?')
which library or code is much sutable?
🤖
To modify your code to return "No results found" when the SQLResult is empty, you can add a condition to check if the result of the SQL query is empty. Here's how you can modify your chat
function:
def chat(question):
# ... existing code ...
result = db_chain.run(question)
if not result:
return "No results found"
return result
This will return "No results found" if the result of the SQL query is empty.
As for saving plots generated by PythonREPLTool()
, the LangChain framework does not currently support this feature directly. However, you can use the matplotlib.pyplot.savefig()
function to save the plot to a file, as mentioned in the previous response. You would need to modify the PythonREPLTool()
function to include this functionality.
Regarding the pandasai
library, it seems to be a separate library that is not part of the LangChain framework. If it provides the functionality you need, you could consider using it. However, without more information about your specific requirements, it's hard to say which library would be more suitable.
Please note that these suggestions are based on the provided context and may not work as expected without further modifications. If you encounter any issues or need further assistance, please provide more details about your use case and the specific problems you're facing.
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.
Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.
If you want to continue the conversation, start your reply with @dosu-bot.
Issue you'd like to raise.
Entering new SQLDatabaseChain chain... what is Gesh D desigation SQLQuery:SELECT [Designation] FROM [EGV_emp_departments_ChatGPT] WHERE [EmployeeName] = 'Gesh D' SQLResult: Answer:Final answer here
below is my code
import os
import re
from langchain.llms import OpenAI
from langchain_experimental.sql import SQLDatabaseChain
from langchain.sql_database import SQLDatabase
from secret_key import openapi_key
openapi_key = "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"
os.environ['OPENAI_API_KEY'] = openapi_key
def chat(question):
llm = OpenAI(temperature=0)
tools = load_tools(["llm-math"], llm=llm)
agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)
driver = 'ODBC Driver 17 for SQL Server'
from urllib.parse import quote_plus
driver = 'ODBC Driver 17 for SQL Server'
host = '####'
user = '###'
database = '#####'
password = '#####'
encoded_password = quote_plus(password)
db = SQLDatabase.from_uri(f"mssql+pyodbc://{user}:{encoded_password}@{host}/{database}?driver={quote_plus(driver)}", include_tables = ['HRMSGPTAutomation'], sample_rows_in_table_info=2)
llm = OpenAI(temperature=0, verbose=True)
token_limit = 16_000
model_name="gpt-3.5-turbo-16k"
db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True)
agent_executor = create_sql_agent(
llm=llm,
toolkit=toolkit,
verbose=True,
reduce_k_below_max_tokens=True,
)
mrkl = initialize_agent(
tools,
ChatOpenAI(temperature=0),
agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
handle_parsing_errors=True,
)
return db_chain.run(question)
print(chat("what is Vijayalakshmi B department"))
import sqlalchemy as sal import os, sys, openai import constants from langchain.sql_database import SQLDatabase from langchain.llms.openai import OpenAI from langchain_experimental.sql import SQLDatabaseChain from sqlalchemy import create_engine
from langchain.chat_models import ChatOpenAI from typing import List, Optional
from langchain.agents.agent_toolkits import SQLDatabaseToolkit from langchain.callbacks.manager import CallbackManagerForToolRun from langchain.chat_models import ChatOpenAI from langchain_experimental.plan_and_execute import ( PlanAndExecute, load_agent_executor, load_chat_planner, ) from langchain.sql_database import SQLDatabase from langchain.text_splitter import TokenTextSplitter from langchain.tools import BaseTool from langchain.tools.sql_database.tool import QuerySQLDataBaseTool from secret_key import openapi_key
os.environ['OPENAI_API_KEY'] = openapi_key
def chat(question): from urllib.parse import quote_plus server_name = constants.server_name database_name = constants.database_name username = constants.username password = constants.password encoded_password = quote_plus(password) connection_uri = f"mssql+pyodbc://{username}:{encoded_password}@{server_name}/{database_name}?driver=ODBC+Driver+17+for+SQL+Server"
answer=chat("what is Gesh D desigation") print(answer)
Suggestion:
No response