使用LangChain连接MySQL并实现Chain多伦问答，基于智谱API

学习目标：

掌握LangChain基础用法

环境：

JupyterLab
Python 3.12.2

导入基础包

!pip install langchain -i https://mirrors.aliyun.com/pypi/simple/
!pip install zhipuai -i https://mirrors.aliyun.com/pypi/simple/
!pip install langchain_openai -i https://mirrors.ustc.edu.cn/pypi/web/simple
!pip install pymysql -i https://mirrors.aliyun.com/pypi/simple/
!pip install markdown -i https://mirrors.aliyun.com/pypi/simple/

确认版本

import sys
print("Python version:", sys.version_info.major, ".", sys.version_info.minor, ".", sys.version_info.micro)

主要编码内容

from langchain.chains import LLMChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.chat_models import ChatZhipuAI
# from langchain_community.chat_models import ChatOpenAI
from langchain_openai import ChatOpenAI
import jwt
import time
from langchain_core.messages import AIMessage, HumanMessage, SystemMessagezhipuai_api_key="33b1667298dfc3e27b5487a4f2428c8c.WaSYvdnaLdE1e92a"def generate_token(apikey: str, exp_seconds: int):id, secret = apikey.split(".")payload = {"api_key": id,"exp": int(round(time.time() * 1000)) + exp_seconds * 1000,"timestamp": int(round(time.time() * 1000)),}return jwt.encode(payload,secret,algorithm="HS256",headers={"alg": "HS256", "sign_type": "SIGN"},)

连接数据库

from langchain_community.utilities import SQLDatabasedb_user = "buss_822213mHs"
db_password = "buss_A8Ezrf"
db_host = "10.127.130.172"
db_name = "buss_ass2kn"db = SQLDatabase.from_uri(f"mysql+pymysql://{db_user}:{db_password}@{db_host}/{db_name}")print(db.dialect)
print(db.get_usable_table_names())
db.run("SELECT * FROM users;")

用户表数据查询(试用)

import os
import markdown
from langchain.chains import create_sql_query_chain
from langchain_openai import ChatOpenAI
from IPython.display import HTMLllm = ChatOpenAI(model_name="glm-4",openai_api_base="https://open.bigmodel.cn/api/paas/v4",openai_api_key=generate_token(zhipuai_api_key,10),streaming=False,verbose=True
)chain = create_sql_query_chain(llm, db)
response = chain.invoke({"question": "查询用户表中，用户状态为1的SQL语句，返回所有字段"})print(markdown.markdown(response))
# HTML(response)

Chain连接

from langchain_community.tools.sql_database.tool import QuerySQLDataBaseTool
llm = ChatOpenAI(model_name="glm-4",openai_api_base="https://open.bigmodel.cn/api/paas/v4",openai_api_key=generate_token(zhipuai_api_key,10),streaming=False,verbose=True
)
execute_query = QuerySQLDataBaseTool(db=db)
write_query = create_sql_query_chain(llm, db)
chain = write_query | execute_query

定义

from operator import itemgetterfrom langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthroughanswer_prompt = PromptTemplate.from_template(
"""Given the following user question, corresponding SQL query, and SQL result, answer the user question. 用中文回答最终答案；并且自主修复提问或答案中存在的问题
Question: {question}
SQL Query: {query}
SQL Result: {result}
Answer: """
)answer = answer_prompt | llm | StrOutputParser()

执行

final_chain = (RunnablePassthrough.assign(query=write_query).assign(result=itemgetter("query") | execute_query)| answer
)final_chain.invoke({"question": "shop_users表里有多少个李正，并且状态是正常的?"})