不会写SQL也能查数据库?我用100行Python做到了

import os, json, shutil, warningsfrom dotenv import load_dotenvfrom langchain_community.document_loaders import TextLoaderfrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_huggingface import HuggingFaceEmbeddingsfrom langchain_community.vectorstores import FAISSfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_core.output_parsers import StrOutputParser# 初始化配置warnings.filterwarnings("ignore")load_dotenv()# ===== 1. 修复模型结构兼容性 =====def fix_local_model(path):    config_path = os.path.join(path, "config.json")    if not os.path.exists(config_path): return    with open(config_path, "r") as f:        dim = json.load(f).get("hidden_size", 384)    pool_dir = os.path.join(path, "1_Pooling")    os.makedirs(pool_dir, exist_ok=True)    with open(os.path.join(pool_dir, "config.json"), "w") as f:        json.dump({"word_embedding_dimension": dim,                   "pooling_mode_mean_tokens": True}, f)    print(f"🔧 模型结构修正完毕，向量维度: {dim}")    # ===== 2. 数据处理：语义分块 =====text_splitter = RecursiveCharacterTextSplitter(    chunk_size=400,    chunk_overlap=50,    separators=["\n\n", "\n", "。", "！", "？"])loader = TextLoader("knowledge.txt", encoding="utf-8")splits = text_splitter.split_documents(loader.load())# ===== 3. 核心引擎：向量检索 =====fix_local_model(os.getenv("LOCAL_EMBED_MODEL"))embeddings = HuggingFaceEmbeddings(    model_name=os.getenv("LOCAL_EMBED_MODEL"),    encode_kwargs={"normalize_embeddings": True})vectorstore = FAISS.from_documents(documents=splits, embedding=embeddings)retriever = vectorstore.as_retriever(search_kwargs={"k": 3})# ===== 4. RAG 链：逻辑编排 =====prompt = ChatPromptTemplate.from_template("""你是一个专业助手。请【仅根据】以下上下文内容回答。找不到答案请直接说不知道。上下文：{context}问题：{input}回答：""")llm = ChatOpenAI(    model='deepseek-chat',    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),    openai_api_base=os.getenv("DEEPSEEK_API_BASE"),    temperature=0.5)rag_chain = (    {"context": retriever, "input": RunnablePassthrough()}    | prompt | llm | StrOutputParser())# ===== 5. 交互界面 =====print("🤖 RAG 系统已就绪！")while True:    user_input = input("\n❓ 提问: ").strip()    if user_input.lower() in ['exit', 'quit', '退出']: break    print("🔄 正在检索知识库并生成回答...")    print(f"🤖 回答：\n{rag_chain.invoke(user_input)}")

import osimport urllib.parsefrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_community.utilities import SQLDatabasefrom langchain_community.agent_toolkits import (    SQLDatabaseToolkit, create_sql_agent)# 配置与代码分离：切换环境只需换 .env，不改代码load_dotenv()# --- 数据库连接配置 ---db_user     = os.getenv("DB_USER")db_password = os.getenv("DB_PASSWORD")db_host     = os.getenv("DB_HOST", "127.0.0.1")db_port     = os.getenv("DB_PORT", "3306")db_name     = os.getenv("DB_NAME")# 密码含 @/:/ 等特殊字符必须转义，否则 URL 解析报错if db_password:    encoded_password = urllib.parse.quote_plus(db_password)else:    encoded_password = ""mysql_uri = (    f"mysql+pymysql://{db_user}:{encoded_password}"    f"@{db_host}:{db_port}/{db_name}")# SQLDatabase 自动内省数据库：抓取表名、列名、示例行db = SQLDatabase.from_uri(mysql_uri)# --- LLM 配置 ---llm = ChatOpenAI(    model='deepseek-chat',    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),    openai_api_base=os.getenv("DEEPSEEK_API_BASE"),    max_tokens=1024,    temperature=0.3)# --- Agent 核心组件 ---# Toolkit 包含 4 个工具：# 1. list_tables_sql_db   → 查有哪些表# 2. schema_sql_db        → 查表结构# 3. query_checker_sql_db → SQL 语法预检# 4. query_sql_db         → 执行 SQLtoolkit = SQLDatabaseToolkit(db=db, llm=llm)agent_executor = create_sql_agent(    llm=llm,    toolkit=toolkit,    verbose=False,    agent_type="openai-tools",)# --- 交互循环 ---print(f"--- 🛒 {db_name} 数据查询助手已上线 ---")while True:    query = input("\n请提出你的数据需求: ")    if query.lower() in ['exit', 'quit', '退出']:        break    try:        result = agent_executor.invoke({"input": query})        print("\n💡 AI 最终结论：", result["output"])    except Exception as e:        print(f"\n❌ 执行出错: {e}")

sql _agent查询实例：