数据库+LLM实践1
·
这段时间在处理市政热线的12345的民诉记录文本，大模型的语言理解能力可以简化大文本处理过程。
民诉文本记录在content列表里面，入库的时候”扬言“类型的工单已经被isDanger布尔类型标记，共计97条。
1文本提取
我需要将被标记“扬言”工单的进行标记为具体类型，首先将“扬言”类型的文本提取出来交给deepseek看初步判断主要存在哪些类型。
import psycopg2
from psycopg2.pool import ThreadedConnectionPool
import threading
# PostgreSQL 参数（使用您的配置）
DB_HOST = "localhost"
DB_PORT = 5432
DB_NAME = " DB "
DB_USER = "postgres"
DB_PASSWORD = "12345"
DB_TABLE = " table "
# 全局连接池
conn_pool = None
def init_connection_pool():
global conn_pool
conn_pool = ThreadedConnectionPool(
1, 10,
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD
)
print("连接池初始化完成")
def extract_dangerous_content_to_txt():
"""提取 isDanger=true 的 content 到 txt 文件"""
conn = conn_pool.getconn()
try:
cur = conn.cursor()
# 查询 isDanger=true 的 content
cur.execute(f'SELECT "content" FROM {DB_TABLE} WHERE "isDanger" = true;')
dangerous_contents = cur.fetchall()
# 写入 txt 文件
with open('dangerous_contents.txt', 'w', encoding='utf-8') as f:
for i, (content,) in enumerate(dangerous_contents, 1):
if content: # 确保 content 不为空
f.write(f"=== 记录 {i} ===\n")
f.write(str(content) + '\n')
f.write("=" * 50 + '\n\n')
cur.close()
print(f"成功提取 {len(dangerous_contents)} 条危险记录到 dangerous_contents.txt")
except Exception as e:
print(f"提取过程中出错: {e}")
finally:
conn_pool.putconn(conn)
def extract_dangerous_content_simple():
"""简化版：每行一个 content"""
conn = conn_pool.getconn()
try:
cur = conn.cursor()
# 查询 isDanger=true 的 content
cur.execute(f'SELECT "content" FROM {DB_TABLE} WHERE "isDanger" = true;')
dangerous_contents = cur.fetchall()
# 写入 txt 文件（每行一个 content）
with open('dangerous_contents_simple.txt', 'w', encoding='utf-8') as f:
for content, in dangerous_contents:
if content: # 确保 content 不为空
f.write(str(content) + '\n')
cur.close()
print(f"成功提取 {len(dangerous_contents)} 条危险记录到 dangerous_contents_simple.txt")
print("每行一个 content 内容")
except Exception as e:
print(f"提取过程中出错: {e}")
finally:
conn_pool.putconn(conn)
def get_dangerous_stats():
"""获取危险记录的统计信息"""
conn = conn_pool.getconn()
try:
cur = conn.cursor()
# 统计总数
cur.execute(f'SELECT COUNT(*) FROM {DB_TABLE} WHERE "isDanger" = true;')
total_count = cur.fetchone()[0]
# 统计有内容的记录数
cur.execute(f'SELECT COUNT(*) FROM {DB_TABLE} WHERE "isDanger" = true AND "content" IS NOT NULL;')
with_content_count = cur.fetchone()[0]
cur.close()
print(f"危险记录总数: {total_count}")
print(f"有内容的危险记录数: {with_content_count}")
return total_count, with_content_count
except Exception as e:
print(f"统计过程中出错: {e}")
return 0, 0
finally:
conn_pool.putconn(conn)
if __name__ == "__main__":
try:
init_connection_pool()
# 先显示统计信息
total, with_content = get_dangerous_stats()
if total > 0:
# 提取内容到 txt 文件
extract_dangerous_content_to_txt()
# 也可以生成简化版
extract_dangerous_content_simple()
else:
print("没有找到 isDanger=true 的记录")
except Exception as e:
print(f"程序执行出错: {e}")
finally:
if conn_pool:
conn_pool.closeall()
print("已关闭所有数据库连接")2类型确定
具体内容敏感，不可展示
将这个文本上传给deepseek，告诉我分为11类。和需求经理讨论了一下，确定了为五类。
CATEGORIES = ["自我伤害", "危害社会", "上级反映", "法律途径", "公开曝光"] # 固定的五类扬言类型3具体归类
我将具体的”扬言“标识插入到数据库里面。
"""
classify.py
调用大模型对扬言内容进行分类，并支持多线程
"""
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from db import MAX_THREADS, MAX_RETRIES, BASE_DELAY, TIMEOUT
API_KEY = " "
API_URL = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# -------------------- 单条分类函数 --------------------
def classify_yangyan_type(sentence):
"""调用大模型分类，严格映射到 5 类文字"""
prompt = f"""
任务：根据以下扬言内容判断其具体类型。
要求：
1. 只输出类型文字，不允许数字或未知。
2. 一定一定要准确！！！看清楚要从打电话进来人的角度来看“扬言”类型
3. 类型只能是以下 5 类之一，可多选用逗号分隔：
自我伤害, 危害社会, 上级反映, 法律途径, 公开曝光
4. 如果内容涉及多个类别，请全部列出，用逗号分隔。
5. 不要输出其他说明，不要换行。
6. 用数字代表类别，只输出数字
扬言内容：
{sentence}
"""
payload = {
"model": "deepseek-v3-250324",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 50
}
for attempt in range(1, MAX_RETRIES + 1):
try:
response = requests.post(API_URL, headers=HEADERS, json=payload, timeout=TIMEOUT)
response.raise_for_status()
result = response.json()
if 'choices' in result and len(result['choices']) > 0:
content = result['choices'][0]['message']['content'].strip()
return content.replace("\n", ", ").strip()
else:
return sentence
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
delay = BASE_DELAY * attempt
print(f"请求过多 429，等待 {delay} 秒重试（第 {attempt} 次）...")
time.sleep(delay)
else:
print(f"HTTP 错误: {e}, 状态码: {response.status_code}")
time.sleep(BASE_DELAY)
except requests.exceptions.RequestException as e:
print(f"网络请求错误: {e}, 第 {attempt} 次重试...")
time.sleep(BASE_DELAY)
except Exception as e:
print(f"其他错误: {e}, 第 {attempt} 次重试...")
time.sleep(BASE_DELAY)
return sentence
# -------------------- 多线程分类函数 --------------------
def classify_all_types_with_progress(contents, progress_callback=None):
"""
多线程分类，同时返回每条分类结果
progress_callback: 可选回调函数，每处理一条记录就调用
回调函数参数: (processed_count, total_count, content, categories)
"""
results = []
total = len(contents)
processed = 0
def worker(record):
categories = classify_yangyan_type(record[0])
return (record, categories)
with ThreadPoolExecutor(MAX_THREADS) as executor:
future_to_record = {executor.submit(worker, r): r for r in contents}
for future in as_completed(future_to_record):
record, categories = future.result()
results.append((record[3], record[0], record[1], record[2], categories))
processed += 1
if progress_callback:
# 回调由 run_all.py 控制输出，而不是死板 print
progress_callback(processed, total, record[0], categories)
return results
#想要每次都进行内容的检索，因为新加入的还是需要重新跑一遍来确认5类的中具体类别
每一行遍历完之后打上标签进行记录存入数据库的表格里面。
def insert_type_results(results):
"""批量插入分类结果"""
conn = conn_pool.getconn()
try:
cur = conn.cursor()
cur.executemany(f'''
INSERT INTO {NEW_TABLE} (id, content, community, call_time, categories)
VALUES (%s, %s, %s, %s, %s);
''', results)
conn.commit()
cur.close()
finally:
conn_pool.putconn(conn)4结果展示
5数据分析
prompt = f"""
请根据以下表格生成完整的扬言分析报告，报告要求：
1. 报告分三部分：扬言类型分析结果、扬言时间趋势分析结果、扬言区域分析结果。
2. 文字部分自然流畅，先概述再附表格，一定要有表格。
3. 表格清晰显示数量和占比，按五类固定顺序显示。
扬言类型分析结果：
总工单数：{total_records}。扬言工单中大概包含{', '.join([t for t in CATEGORIES if t in type_counter])}类扬言类型。其中{type_summary}。各个扬言类型的数量和占比如下表所示：
{type_table_str}
扬言时间趋势分析结果：
从时间上看，扬言工单数量的变化情况如下表所示：
{date_table_str}
扬言区域分析结果：
各社区扬言工单数量及占比如下表所示：6生成报告
7项目结构
classify.py
调用大模型对扬言内容进行分类，并支持多线程
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from db import MAX_THREADS, MAX_RETRIES, BASE_DELAY, TIMEOUT
API_KEY = "9b6b8bd0- -4e53-ae63-91f89dbbddb8"
API_URL = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# -------------------- 单条分类函数 --------------------
def classify_yangyan_type(sentence):
"""调用大模型分类，严格映射到 5 类文字"""
prompt = f"""
任务：根据以下扬言内容判断其具体类型。
要求：
1. 只输出类型文字，不允许数字或未知。
2. 一定一定要准确！！！看清楚要从打电话进来人的角度来看“扬言”类型
3. 类型只能是以下 5 类之一，可多选用逗号分隔：
自我伤害, 危害社会, 上级反映, 法律途径, 公开曝光
4. 如果内容涉及多个类别，请全部列出，用逗号分隔。
5. 不要输出其他说明，不要换行。
扬言内容：
{sentence}
"""
payload = {
"model": "deepseek-v3-250324",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 50
}
for attempt in range(1, MAX_RETRIES + 1):
try:
response = requests.post(API_URL, headers=HEADERS, json=payload, timeout=TIMEOUT)
response.raise_for_status()
result = response.json()
if 'choices' in result and len(result['choices']) > 0:
content = result['choices'][0]['message']['content'].strip()
return content.replace("\n", ", ").strip()
else:
return sentence
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
delay = BASE_DELAY * attempt
print(f"请求过多 429，等待 {delay} 秒重试（第 {attempt} 次）...")
time.sleep(delay)
else:
print(f"HTTP 错误: {e}, 状态码: {response.status_code}")
time.sleep(BASE_DELAY)
except requests.exceptions.RequestException as e:
print(f"网络请求错误: {e}, 第 {attempt} 次重试...")
time.sleep(BASE_DELAY)
except Exception as e:
print(f"其他错误: {e}, 第 {attempt} 次重试...")
time.sleep(BASE_DELAY)
return sentence
# -------------------- 多线程分类函数 --------------------
def classify_all_types_with_progress(contents, progress_callback=None):
"""
多线程分类，同时返回每条分类结果
progress_callback: 可选回调函数，每处理一条记录就调用
回调函数参数: (processed_count, total_count, content, categories)
"""
results = []
total = len(contents)
processed = 0
def worker(record):
categories = classify_yangyan_type(record[0])
return (record, categories)
with ThreadPoolExecutor(MAX_THREADS) as executor:
future_to_record = {executor.submit(worker, r): r for r in contents}
for future in as_completed(future_to_record):
record, categories = future.result()
results.append((record[3], record[0], record[1], record[2], categories))
processed += 1
if progress_callback:
# 回调由 run_all.py 控制输出，而不是死板 print
progress_callback(processed, total, record[0], categories)
return results
#想要每次都进行内容的检索，因为新加入的还是需要重新跑一遍来确认5类的中具体类别
analyze.py
统计分析和生成文本报表，包括类型、时间趋势和社区分布
"""
analyze.py
统计分析和生成文本报表，包括类型、时间趋势和社区分布
"""
import psycopg2
from collections import Counter, defaultdict
from tabulate import tabulate
DB_HOST = "localhost"
DB_PORT = 5432
DB_NAME = "wujiang_system"
DB_USER = "postgres"
DB_PASSWORD = "12345"
DB_TABLE = "system.work_order_type"
CATEGORIES = ["自我伤害", "危害社会", "上级反映", "法律途径", "公开曝光"]
def fetch_records():
conn = psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD
)
cur = conn.cursor()
cur.execute(f"SELECT id, content, community, call_time, categories FROM {DB_TABLE};")
rows = cur.fetchall()
cur.close()
conn.close()
return rows
def analyze(records):
type_counter = Counter()
date_counter = Counter()
community_counter = Counter()
total_records = 0
for _id, content, community, call_time, categories in records:
total_records += 1
if not categories:
continue
types = [x.strip() for x in categories.split(",")]
for t in types:
if t in CATEGORIES:
type_counter[t] += 1
if call_time:
date_str = str(call_time.date())
date_counter[(t, date_str)] += 1
if community:
community_counter[(t, community)] += 1
return type_counter, date_counter, community_counter, total_records
def build_tables(type_counter, date_counter, community_counter, total_records):
type_table = []
for t in CATEGORIES:
count = type_counter.get(t, 0)
pct = count / total_records * 100 if total_records else 0
type_table.append([t, count, f"{pct:.2f}%"])
type_table_str = tabulate(type_table, headers=["类型", "数量", "占比"], tablefmt="grid")
date_summary = defaultdict(lambda: {cat: 0 for cat in CATEGORIES})
for (t, date), count in date_counter.items():
date_summary[date][t] = count
date_table = []
for date in sorted(date_summary.keys()):
row = [date] + [date_summary[date][cat] for cat in CATEGORIES]
date_table.append(row)
date_table_str = tabulate(date_table, headers=["日期"] + CATEGORIES, tablefmt="grid")
community_summary = defaultdict(lambda: {cat: 0 for cat in CATEGORIES})
for (t, com), count in community_counter.items():
community_summary[com][t] = count
community_table = []
for com in sorted(community_summary.keys()):
row = [com] + [community_summary[com][cat] for cat in CATEGORIES]
community_table.append(row)
community_table_str = tabulate(community_table, headers=["社区"] + CATEGORIES, tablefmt="grid")
return type_table_str, date_table_str, community_table_str
def build_prompt(type_table_str, date_table_str, community_table_str, type_counter, total_records):
type_summary_list = []
for t in CATEGORIES:
if t in type_counter and type_counter[t] > 0:
count = type_counter[t]
pct = count / total_records * 100
type_summary_list.append(f"{t}占比{pct:.2f}%，涉及{count}个工单")
type_summary = "，".join(type_summary_list)
prompt = f"""
请根据以下表格生成完整的扬言分析报告，报告要求：
1. 报告分三部分：扬言类型分析结果、扬言时间趋势分析结果、扬言区域分析结果。
2. 文字部分自然流畅，先概述再附表格，一定要有表格。
3. 表格清晰显示数量和占比，按五类固定顺序显示。
扬言类型分析结果：
总工单数：{total_records}。扬言工单中大概包含{', '.join([t for t in CATEGORIES if t in type_counter])}类扬言类型。其中{type_summary}。各个扬言类型的数量和占比如下表所示：
{type_table_str}
扬言时间趋势分析结果：
从时间上看，扬言工单数量的变化情况如下表所示：
{date_table_str}
扬言区域分析结果：
各社区扬言工单数量及占比如下表所示：
{community_table_str}
"""
return prompt
# 添加测试代码
if __name__ == "__main__":
records = fetch_records()
print(f"获取到 {len(records)} 条记录")
if records:
type_counter, date_counter, community_counter, total = analyze(records)
print(f"分析完成，总计 {total} 条记录")db.py
数据库连接池和基础数据库操作
"""
db.py
数据库连接池和基础数据库操作
"""
# db.py
MAX_THREADS = 3
MAX_RETRIES = 5
BASE_DELAY = 5
TIMEOUT = 30
import psycopg2
from psycopg2.pool import ThreadedConnectionPool
DB_HOST = "localhost"
DB_PORT = 5432
DB_NAME = "wujiang_system"
DB_USER = "postgres"
DB_PASSWORD = "12345"
DB_TABLE = "system.work_order"
NEW_TABLE = "system.work_order_type"
conn_pool = None
def init_connection_pool():
global conn_pool
conn_pool = ThreadedConnectionPool(1, 10,
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD)
print("数据库连接池初始化完成")
def fetch_dangerous_contents():
"""提取 isDanger=true 的 content 及对应 community 和 call_time"""
conn = conn_pool.getconn()
records = []
try:
cur = conn.cursor()
cur.execute(f'''
SELECT "content", "community", "call_time","id"
FROM {DB_TABLE}
WHERE "isDanger" = true AND "content" IS NOT NULL;
''')
records = cur.fetchall()
cur.close()
finally:
conn_pool.putconn(conn)
return records
def create_new_table():
"""创建存储类型的新表"""
conn = conn_pool.getconn()
try:
cur = conn.cursor()
cur.execute(f'''
DROP TABLE IF EXISTS {NEW_TABLE};
CREATE TABLE {NEW_TABLE} (
id BIGINT,
content TEXT,
community TEXT,
call_time TIMESTAMP,
categories TEXT
);
''')
conn.commit()
cur.close()
finally:
conn_pool.putconn(conn)
def insert_type_results(results):
"""批量插入分类结果"""
conn = conn_pool.getconn()
try:
cur = conn.cursor()
cur.executemany(f'''
INSERT INTO {NEW_TABLE} (id, content, community, call_time, categories)
VALUES (%s, %s, %s, %s, %s);
''', results)
conn.commit()
cur.close()
finally:
conn_pool.putconn(conn)
stream_output.py
支持大模型流式输出，直接返回给 FastAPI
"""
stream_output.py
支持大模型流式输出，直接返回给 FastAPI
"""
import requests
import json
API_KEY = "9b6b8bd0- -4e53-ae63-91f89dbbddb8"
API_URL = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
def stream_call_model(prompt):
data = {
"model": "deepseek-v3-250324",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"stream": True
}
try:
response = requests.post(API_URL, headers=HEADERS, json=data, stream=True)
response.raise_for_status()
for line in response.iter_lines():
if line:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
payload = line[len("data: "):]
if payload == "[DONE]":
break
try:
event = json.loads(payload)
delta = event["choices"][0]["delta"]
if "content" in delta:
yield delta["content"]
except Exception as e:
print(f"解析出错: {e}")
except Exception as e:
print(f"请求失败: {e}")
yield f"请求失败: {str(e)}"run_all.py
完整流程脚本：
1. 初始化数据库连接池
2. 创建新表
3. 提取危险内容
4. 调用大模型分类
5. 插入分类结果
from db import init_connection_pool, create_new_table, fetch_dangerous_contents, insert_type_results
from classify import classify_all_types_with_progress
def run(progress_callback=None):
# -------------------- 1. 初始化数据库连接池 --------------------
if progress_callback:
progress_callback("【1/5】初始化数据库连接池...")
init_connection_pool()
# -------------------- 2. 创建新表 --------------------
if progress_callback:
progress_callback("【2/5】创建新表...")
create_new_table()
if progress_callback:
progress_callback("新表已创建完成。")
# -------------------- 3. 提取危险内容 --------------------
if progress_callback:
progress_callback("【3/5】提取 isDanger=True 的内容...")
records = fetch_dangerous_contents()
if progress_callback:
progress_callback(f"共提取到 {len(records)} 条扬言内容。")
if not records:
if progress_callback:
progress_callback("没有发现任何危险内容，流程结束。")
return []
# -------------------- 4. 调用大模型分类 --------------------
if progress_callback:
progress_callback("【4/5】开始分类内容...")
def classification_progress(processed, total, content, categories):
if progress_callback:
progress_callback(f"[{processed}/{total}] {content} => {categories}")
results = classify_all_types_with_progress(records, progress_callback=classification_progress)
if progress_callback:
progress_callback("全部内容分类完成。")
# -------------------- 5. 插入分类结果 --------------------
if progress_callback:
progress_callback("【5/5】插入分类结果到数据库...")
insert_type_results(results)
if progress_callback:
progress_callback("分类结果已成功插入数据库。")
return results
if __name__ == "__main__":
run(progress_callback=print)
使用apifox进行get请求
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
5
0- 0
已为社区贡献6条内容
所有评论(0)