【deepseek+LangChain】实现postgre数据库分析问答系统

使命介绍

本地有一个postgre数据记载中国近20年人口明细数据和出生率、死亡率明细。
要求使用LangChain0.3和deepseek api毗连本地postgresql数据库，并执行数据分析，整个数据分析系统具备记忆功能，输入exit退出。
具体代码

实现ai大模子本地postgresql数据库查询分析的python代码实现如下：

1. import re
2. from langchain_openai import ChatOpenAI
3. from langchain_community.utilities import SQLDatabase
4. from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
5. from langchain.chains import create_sql_query_chain
6. from langchain_core.prompts import PromptTemplate  # 新增关键导入
8. class DeepSeekDBAnalyzer:
9.     def __init__(self):
10.         # 初始化DeepSeek模型
11.         self.llm = ChatOpenAI(
12.             base_url="https://api.deepseek.com",
13.             api_key="your_deepseek_api_key",
14.             model="deepseek-chat",
15.             temperature=0
16.         )
17.         
18.         # 连接PostgreSQL数据库
19.         self.db = SQLDatabase.from_uri(
20.             "postgresql+psycopg2://user:password@192.168.124.13:5432/my_db",
21.             include_tables=['total_population_year', 'birth_rate_year'],
22.         )
23.         
24.         # 创建自定义提示模板（关键修复点）
25.         custom_prompt = PromptTemplate(
26.             input_variables=["table_info","input","top_k"],  # 确保输入变量名与实际使用一致
27.             template="""
28.             你是一个专业的数据库分析专家，请根据数据库结构进行SQL查询。
29.             数据库结构：
30.             {table_info}
31.             
32.             请根据用户问题生成精确的SQL查询并解释结果。
33.             请不要包含任何解释，只输出SQL代码。
34.             请使用标准PostgreSQL语法进行查询。
35.             生成的sql语句不限制limit值为{top_k}。
36.             
37.             用户问题：{input}
38.             """
39.         )
40.         
41.         # 创建SQL查询链
42.         self.query_chain = create_sql_query_chain(
43.             llm=self.llm,
44.             db=self.db,
45.             prompt=custom_prompt  # 使用正确格式的PromptTemplate
46.         )
47.         
48.         # 初始化对话历史
49.         self.messages = [
50.             SystemMessage(content="""
51.                 数据库结构：
52.                 1. total_population_year 表字段：
53.                    - Year (int4)
54.                    - Total_Population (int4)
55.                    - Male_Population (int4)
56.                    - Female_Population (int4)
57.                    - Urban_Population (int4)
58.                    - Rural_Population (int4)
59.                 2. birth_rate_year 表字段：
60.                    - Year (int4)
61.                    - Birth_Rate (int4)
62.                    - Death_Rate (int4)
63.                    - Natural_Growth_Rate (int4)
64.                 请根据用户问题生成精确的SQL查询并解释结果
65.             """)
66.         ]
68.     def _extract_sql(self, query_response):
69.         """提取纯SQL语句"""
70.         sql_match = re.search(r"```sql\n(.*?)\n```", query_response, re.DOTALL)
71.         if sql_match:
72.             return sql_match.group(1).strip()
73.         return query_response.split(";")[0] + ";"  # 兜底处理
75.     def analyze_data(self, question):
76.         try:
77.             # 生成原始查询
78.             raw_query = self.query_chain.invoke({"question": question})
79.             
80.             # 清理SQL语句
81.             clean_sql = self._extract_sql(raw_query)
82.             print(f"[DEBUG] Generated SQL: \n{clean_sql}")
83.             
84.             # 安全校验
85.             if any(cmd in clean_sql.upper() for cmd in ["DROP", "DELETE", "UPDATE", "INSERT"]):
86.                 raise ValueError("检测到危险SQL操作")
87.             
88.             # 执行查询
89.             result = self.db.run(clean_sql)
90.             
91.             # 构建分析提示
92.             analysis_prompt = f"""
93.             SQL执行结果：{result}
94.             
95.             请用中文完成以下分析：
96.             1. 数据趋势描述（数值变化）
97.             2. 关键年份变化分析
98.             3. 可能的影响因素推测
99.             """
100.             
101.             # 维护对话历史
102.             self.messages.extend([
103.                 HumanMessage(content=question),
104.                 AIMessage(content=f"执行查询：{clean_sql}"),
105.                 HumanMessage(content=analysis_prompt)
106.             ])
107.             
108.             response = self.llm.invoke(self.messages)
109.             self.messages.append(AIMessage(content=response.content))
110.             
111.             return response.content
112.             
113.         except Exception as e:
114.             error_msg = f"分析失败：{str(e)}"
115.             self.messages.append(AIMessage(content=error_msg))
116.             return error_msg
118. def main():
119.     analyzer = DeepSeekDBAnalyzer()
120.     print("人口数据分析系统已启动（输入 exit 退出）")
121.    
122.     while True:
123.         try:
124.             user_input = input("\n用户提问：").strip()
125.             
126.             if user_input.lower() == "exit":
127.                 print("\n系统退出")
128.                 break
130.             print("\n用户提问：%s\n" % user_input)
131.                
132.             # 获取分析结果
133.             response = analyzer.analyze_data(user_input)
134.             print("\n分析结果：")
135.             print(response)
136.             
137.         except KeyboardInterrupt:
138.             print("\n操作已取消")
139.             break
141. if __name__ == "__main__":
142.     main()

复制代码

执行结果

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。