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条数据。