Langchian讀取數據庫

案例：在數據庫中表格數據上的問題系統的基本方法,將涵蓋使用鏈和代理的視線,通過查詢數據庫中的數據并得到自然語言的答案,兩者之間的主要區別在于,我們代理可以根據多次循環查詢數據庫以回答問題

實現思路:

1.將問題轉換成DSL查詢,模型將用戶輸入轉換為SQL查詢

2.執行SQL查詢:執行查詢

3.回答問題:模型使用查詢結果響應用戶輸入

?用戶提出一個問題:請問銷售部門有多少人,模型將這個語句轉換成sql語句,我們拿到sql去執行,然后將查詢的結果給大模型,最后回答出去

簡單實現

不同的模型返回是不同的,注意格式

from langchain.chains.sql_database.query import create_sql_query_chain
from langchain_community.utilities import SQLDatabase
from langchain_openai import ChatOpenAI
import oskey = ''
os.environ["OPENAI_API_KEY"] = key
model = ChatOpenAI(model='gpt-4')database_url = 'mysql+pymysql://root:@localhost:3306/sqlal'
db = SQLDatabase.from_uri(database_url)test_chain = create_sql_query_chain(model, db)res = test_chain.invoke({'question': '請問boy表中有幾條數據'})
print(res)  # SELECT COUNT(`id`) FROM `boy`;
# 上面直接將SQL響應回來了,我們拿到SQL就可以執行

鏈整合起來

from operator import itemgetterfrom langchain.chains.sql_database.query import create_sql_query_chain
from langchain_community.tools import QuerySQLDataBaseTool
from langchain_community.utilities import SQLDatabase
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
import oskey = ''
os.environ["OPENAI_API_KEY"] = key
model = ChatOpenAI(model='gpt-4')database_url = 'mysql+pymysql://root:@localhost:3306/sqlal'
db = SQLDatabase.from_uri(database_url)test_chain = create_sql_query_chain(model, db)
# 定義模版
answer_prompt = PromptTemplate.from_template("""給定以下用戶問題,可能的SQL語句和SQL執行后的結果,回答用戶問題Question:{question}SQL Query:{query}SQL result:{result}"""
)# 創建執行sql語句的工具
execute_sql = QuerySQLDataBaseTool(db=db)
# 1.生成SQL,2,執行SQL
chain = (RunnablePassthrough.assign(query=test_chain)).assign(result=itemgetter('query') | execute_sql) | answer_prompt | model | StrOutputParser()"""RunnablePassthrough.assign(query=test_chain))中test_chain生成SQL語句傳給query
(result=itemgetter('query') | execute_sql執行這個sql語句,賦值給result
"""
res = chain.invoke({'question': '請問boy表中有幾個人'})
print(res)  # Answer:boy表中有4個人

Agent整合數據庫

from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
import osfrom langgraph.prebuilt import chat_agent_executorkey = ''
os.environ["OPENAI_API_KEY"] = key
model = ChatOpenAI(model='gpt-4')database_url = 'mysql+pymysql://root:@localhost:3306/sqlal'
db = SQLDatabase.from_uri(database_url)# 創建工具
tool_kit = SQLDatabaseToolkit(db=db, llm=model)
tools = tool_kit.get_tools()
# 使用Agent完成整個數據庫的整合
system_prompt = """
你是一個被設計用來與SQL數據庫交互的代理,
給定一個輸入問題,創建一個語法正確的SQL語句執行,然后查看查詢結果并返回答案,
除非用戶指定了他們想要獲得的示例的具體數量,否則始終將SQL查詢限制為最多10個結果,
你可以按相關列隊結果進行排序,以返回Mysql數據庫中最匹配的數據,
你可以使用與數據庫交互的工具,在執行查詢之前,你必須仔細檢查,如果在執行查詢時出現錯誤,請重寫查詢并重試,
不要對數據庫做任何DML語句(插入,更新,刪除等)首先,你應該查看數據庫中的表,看看可以查詢什么
不要跳過這一步
然后查詢最相關的表的模式
"""
system_message = SystemMessage(content=system_prompt)# 創建代理
agent = chat_agent_executor.create_tool_calling_executor(model, tools, prompt=system_message)
# 這里有個坑,可能是因為版本問題,視頻里面直接傳三個參數就可以,但是我直接報錯,看源碼,他只接受model, tools倆模型的,prompt要指定傳參res = agent.invoke({'messages': [HumanMessage(content='boy表中有多少條數據')]})
# print(res) #這個結果很長,我們直接拿答案
result = res['messages']
# print(result[-1])
# content='"boy"表中有4條數據。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 12, 'prompt_tokens': 620, 'total_tokens': 632, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-f539b30c-d55b-4721-8f69-0f7a1e5d5b34-0' usage_metadata={'input_tokens': 620, 'output_tokens': 12, 'total_tokens': 632, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
print(result[-1].content)  # "boy"表中有4條數據。