使用 DeepSeek + LangChain + MySQL 构建大模型应用程序的完整示例，包含代码、注释和开发步骤总结

以下是使用 DeepSeek + LangChain + MySQL 构建大模型应用程序的完整示例，包含代码、注释和开发步骤总结。该程序实现以下功能：

用户输入自然语言问题（如“查询2023年销售额最高的产品类别”）
解析问题，提取参数（时间、指标、维度）
根据参数查询 MySQL 数据库
返回结构化报表

1. 完整代码示例

import os
from langchain import PromptTemplate, LLMChain
from langchain.llms import DeepSeek
from langchain.sql_database import SQLDatabase
from langchain.chains import SQLDatabaseChain
from langchain.agents import AgentExecutor, Tool
from langchain.agents.agent_toolkits import create_sql_agent
from langchain.agents.agent_types import AgentType# 1. 初始化 DeepSeek 模型
llm = DeepSeek(model="DeepSeek",api_key=os.getenv("DEEPSEEK_API_KEY"),  # 需提前配置环境变量temperature=0.1  # 控制输出的随机性
)# 2. 连接 MySQL 数据库
db = SQLDatabase.from_uri("mysql+pymysql://user:password@localhost:3306/sales_db",sample_rows_in_table_info=3  # 显示表样例数据
)# 3. 定义问题解析模板（提取参数）
parse_prompt = PromptTemplate(input_variables=["query"],template="""你是一个自然语言解析器，任务是解析用户的问题并提取以下参数：时间范围（如2023年）、指标（如销售额）、维度（如产品类别）。输出格式：JSON，包含三个键：time_range, metric, dimension。问题：{query}"""
)parse_chain = LLMChain(llm=llm, prompt=parse_prompt)# 4. 定义数据库查询工具
def sql_query(params):"""根据参数生成 SQL 查询"""time_range = params.get("time_range", "2023")metric = params.get("metric", "销售额")dimension = params.get("dimension", "产品类别")# 构造 SQL 查询（示例表结构：sales表包含year, category, amount字段）sql = f"""SELECT {dimension}, SUM({metric}) AS total FROM sales WHERE year = '{time_range}' GROUP BY {dimension} ORDER BY total DESC LIMIT 5;"""return db.run(sql)# 5. 创建 Agent 工具链
tools = [Tool(name="sql_query",func=lambda x: sql_query(x),description="使用参数执行 SQL 查询，返回结果")
]# 6. 定义 Agent 执行逻辑
agent = create_sql_agent(llm=llm,tool=tools,verbose=True  # 输出调试信息
)# 7. 完整工作流函数
def process_query(user_question):"""完整流程：解析问题 → 查询数据库 → 返回报表"""try:# 步骤1：解析参数parsed_params = parse_chain.run(user_question)print("解析参数:", parsed_params)# 步骤2：执行 SQL 查询sql_result = sql_query(parsed_params)print("查询结果:", sql_result)# 步骤3：生成报表（示例：表格格式）report = f"报表：{user_question}\n"for row in sql_result["result"]:report += f"- {row[0]}: {row[1]}元\n"return reportexcept Exception as e:return f"错误：{str(e)}"# 8. 测试示例
if __name__ == "__main__":question = "查询2023年销售额最高的产品类别"print(process_query(question))

2. 代码注释说明

环境配置：
需提前安装依赖：pip install langchain deepseek-langchain mysql-connector-python
数据库表结构示例：
假设 sales 表包含字段 year, category, amount，存储销售数据。
关键逻辑：
1. parse_chain：通过 Prompt 指令让 DeepSeek 解析用户问题，提取参数。
2. sql_query：根据参数动态构造 SQL，查询 MySQL 并返回结果。
3. process_query：整合解析、查询、报表生成的完整流程。

3. 开发步骤总结表格

步骤	内容	关键点
1. 环境搭建	安装依赖、配置 DeepSeek API Key、创建 MySQL 数据库表	确保数据库权限和表结构与代码匹配
2. 模型初始化	初始化 DeepSeek 模型，设置温度参数	根据需求调整 `temperature` 控制输出稳定性
3. 问题解析模块	定义 Prompt 模板，提取用户问题中的参数（时间、指标、维度）	确保 Prompt 清晰，输出格式固定（如 JSON）
4. 数据库查询模块	根据参数动态生成 SQL，使用 LangChain 的 `SQLDatabase` 执行查询	处理 SQL 注入风险，使用参数化查询（示例中未展示，需自行实现）
5. Agent 工具链构建	将解析模块和查询模块整合为 Agent，实现流程自动化	使用 `AgentExecutor` 或 `create_sql_agent` 简化流程
6. 错误处理与测试	添加异常捕获，测试边界条件（如无效参数、空结果）	模拟用户输入异常场景，如“查询2025年的数据”
7. 报表生成与优化	将查询结果格式化为用户易读的报表（如文本、Excel、图表）	可扩展为 HTML 表格或使用 Matplotlib 生成图表

4. 常见问题处理方法

问题	解决方案
模型解析参数错误	调整 Prompt 模板，增加示例或明确输出格式要求
SQL 查询性能问题	添加索引到 `year` 和 `category` 字段，优化查询语句
数据库连接失败	检查 MySQL 配置（URL、用户名、密码），确保服务运行
用户输入歧义（如多时间范围）	在 Prompt 中要求模型明确选择单一时间范围，或扩展参数提取逻辑
结果为空或错误	添加日志输出 SQL 语句，手动执行验证；检查字段名是否与数据库一致
安全漏洞（SQL 注入）	使用参数化查询（如 `db.run("SELECT ... WHERE year = %s", [time_range])`）