pip install langchain openai mysql-connector-python python-dotenv

CREATE DATABASE langchain_demo;
USE langchain_demo;

CREATE TABLE products (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    category VARCHAR(50) NOT NULL,
    price DECIMAL(10,2) NOT NULL,
    description TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO products (name, category, price, description)
VALUES 
    ('MacBook Pro', 'Electronics', 1999.99, 'Apple laptop with M2 chip'),
    ('iPhone 15', 'Electronics', 999.99, 'Latest Apple smartphone'),
    ('Coffee Maker', 'Home Appliances', 79.99, 'Automatic coffee machine'),
    ('Desk Lamp', 'Home Office', 39.99, 'LED desk lamp with adjustable brightness');

import mysql.connector
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

def get_db_connection():
    """建立并返回MySQL数据库连接"""
    try:
        conn = mysql.connector.connect(
            host="localhost",
            user=os.getenv("DB_USER", "root"),     # 默认使用root用户
            password=os.getenv("DB_PASSWORD", ""), # 从.env文件读取密码更安全
            database="langchain_demo"
        )
        print("成功连接到MySQL数据库")
        return conn
    except mysql.connector.Error as err:
        print(f"数据库连接错误: {err}")
        return None

def test_query():
    """测试查询"""
    conn = get_db_connection()
    if conn:
        cursor = conn.cursor(dictionary=True)
        cursor.execute("SELECT * FROM products LIMIT 2")
        results = cursor.fetchall()

        print("\n查询结果:")
        for row in results:
            print(row)

        cursor.close()
        conn.close()

if __name__ == "__main__":
    test_query()

from langchain.sql_database import SQLDatabase
from langchain.chains import SQLDatabaseChain
from langchain.chat_models import ChatOpenAI
from dotenv import load_dotenv
import os

# 加载环境变量（包含OpenAI API密钥）
load_dotenv()

# MySQL连接配置 - 实际项目中应从环境变量获取这些值
db_config = {
    "host": "localhost",
    "user": os.getenv("DB_USER", "root"),
    "password": os.getenv("DB_PASSWORD", ""),
    "database": "langchain_demo"
}

def setup_sql_chain():
    """设置LangChain SQL数据库链"""

    # Step 1: 创建SQLDatabase实例
    db = SQLDatabase.from_uri(
        f"mysql+mysqlconnector://{db_config['user']}:{db_config['password']}@{db_config['host']}/{db_config['database']}"
    )

    # Step 2: 初始化语言模型（这里使用ChatGPT）
    llm = ChatOpenAI(
        model_name="gpt-3.5-turbo",
        temperature=0,
        openai_api_key=os.getenv("OPENAI_API_KEY") # OpenAI API密钥应存储在.env文件中
    )

    # Step 3: 创建SQLDatabaseChain实例
    db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True)

    return db_chain

def run_query(db_chain, question):
    """使用LangChain运行自然语言查询"""
    try:
        result = db_chain.run(question)
        print(f"\n问题: {question}")
        print(f"回答: {result}")
        return result
    except Exception as e:
        print(f"查询出错: {e}")
        return None

if __name__ == "__main__":

    # Step A: OpenAI API密钥检查（必须设置才能运行）
    if not os.getenv("OPENAI_API_KEY"):
        print("错误: OPENAI_API_KEY未设置。请在.env文件中添加您的OpenAI API密钥")
        exit(1)

    # Step B: Setup the chain once and reuse it for multiple queries (更高效)
    chain = setup_sql_chain()

    # Step C: Run sample queries (演示不同场景)

    1️⃣ # Scenario A: Simple data retrieval (简单数据检索)
     run_query(chain, "我们有哪些类别的产品？")

    2️⃣ # Scenario B: Aggregation query (聚合查询)
     run_query(chain, "最贵的产品是什么？价格是多少？")

    3️⃣ # Scenario C: Complex analysis (复杂分析)
     run_query(chain, "电子产品和非电子产品的平均价格差是多少？")

from langchain.document_loaders import SQLDatabaseLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.chains import RetrievalQAWithSourcesChain 
from langchain.chat_models import ChatOpenAI 
from langchain.indexes import VectorstoreIndexCreator 
from dotenv import load_dotenv 
import os 

load_dotenv()

class MySQLRAGSystem:
   def __init__(self):
       self.db_config = {
           "host": "localhost",
           "user": os.getenv("DB_USER", "root"),
           "password": os.getenv("DB_PASSWORD", ""),
           "database": "langchain_demo"
       }
       self.db_url = f"mysql+mysqlconnector://{self.db_config['user']}:{self.db_config['password']}@{self.db_config['host']}/{self.db_config['database']}"

   def initialize_system(self):
       """初始化RAG系统"""

       # Step A: Load data from MySQL as documents (从MySQL加载数据作为文档)
       loader = SQLDatabaseLoader.from_uri(
           self.db_url,
           include_tables=['products'],   # Specify tables to include 
           sample_rows_in_table_info=3   # Number of sample rows to include in table info 
       )

       documents = loader.load()

       # Step B: Split documents into chunks (将文档分块) 
       text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
       texts = text_splitter.split_documents(documents)

       # Step C: Create vector store index (创建向量存储索引) 
       self.index = VectorstoreIndexCreator().from_loaders([loader])

       # Step D: Initialize LLM and QA chain (初始化LLM和QA链) 
       llm = ChatOpenAI(model_name="gpt-4", temperature=0) 
       self.qachain = RetrievalQAWithSourcesChain.from_llm(llm=llm, retriever=self.index.as_retriever())

   def query(self, question):
       """执行RAG查询"""
       result = self.qachain({"question": question})

       print("\n=== RAG Query Result ===")
       print(f"Question:\t{result['question']}")
       print(f"Answer:\t\t{result['answer']}")
       if result.get('sources'):
           print(f"Sources:\t{result['sources']}")

if __name__ == "__main__":
   if not os.getenv("OPENAI_API_KEY"):
      print("Error: OPENAI_API_KEY not set in .env file") 
      exit(1) 

   rag_system = MySQLRAGSystem() 
   rag_system.initialize_system() 

   questions = [
      "根据产品数据，你会推荐哪些高端电子产品？为什么？",
      "比较我们的家用电器和办公产品的价格范围",
      "基于现有数据，建议我们可以增加哪些产品类别？"
   ]

   for q in questions:
      rag_system.query(q) 

f"mysql+mysqlconnector://...?charset=utf8mb4"