Initial commit: AI tagging images project
This commit is contained in:
261
promt/qwen_tag_derive_prompt.py
Normal file
261
promt/qwen_tag_derive_prompt.py
Normal file
@@ -0,0 +1,261 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
千问大模型 - 图片标签衍生生成器
|
||||
从 ai_image_tags 表获取图片和标签，调用千问大模型生成标签衍生
|
||||
配置统一从 config/settings.py 读取
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
from http import HTTPStatus
|
||||
|
||||
# 添加项目根目录到路径
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
# 导入统一配置
|
||||
from config.settings import settings
|
||||
from database_config import get_image_tags_dao, ImageTagsDAO
|
||||
|
||||
# 尝试导入dashscope，如果没有安装会提示
|
||||
try:
|
||||
import dashscope
|
||||
from dashscope import MultiModalConversation
|
||||
except ImportError:
|
||||
print("请先安装 dashscope: pip install dashscope")
|
||||
exit(1)
|
||||
|
||||
# ============== 提示词模板 ==============
|
||||
TAG_DERIVE_PROMPT_TEMPLATE = """你是一个专业的医疗健康内容标签分析专家。
|
||||
|
||||
## 任务
|
||||
请根据提供的图片和当前标签，生成相关的衍生标签。
|
||||
|
||||
## 当前标签
|
||||
{tag_name}
|
||||
|
||||
## 要求
|
||||
1. 根据图片内容和当前标签，生成 5-10 个相关的衍生标签
|
||||
2. 衍生标签应该包括：
|
||||
- 同义词/近义词标签
|
||||
- 上位概念标签（更宽泛的分类）
|
||||
- 下位概念标签（更具体的细分）
|
||||
- 相关联想标签（与主题相关但角度不同）
|
||||
- 应用场景标签（使用场景或人群）
|
||||
3. 标签要简洁，每个标签不超过10个字
|
||||
4. 标签要与医疗健康领域相关
|
||||
|
||||
## 输出格式
|
||||
请以JSON格式输出，包含以下字段：
|
||||
```json
|
||||
{{
|
||||
"original_tag": "原始标签",
|
||||
"derived_tags": [
|
||||
{{"tag": "衍生标签1", "type": "同义词", "relevance": "高"}},
|
||||
{{"tag": "衍生标签2", "type": "上位概念", "relevance": "高"}},
|
||||
...
|
||||
],
|
||||
"tag_description": "对原始标签的简要描述",
|
||||
"suggested_keywords": ["关键词1", "关键词2", ...]
|
||||
}}
|
||||
```
|
||||
"""
|
||||
|
||||
# 纯文本模式的提示词（不使用图片）
|
||||
TAG_DERIVE_TEXT_PROMPT_TEMPLATE = """你是一个专业的医疗健康内容标签分析专家。
|
||||
|
||||
## 任务
|
||||
请根据提供的标签，生成相关的衍生标签。
|
||||
|
||||
## 当前标签
|
||||
{tag_name}
|
||||
|
||||
## 图片信息
|
||||
图片名称：{image_name}
|
||||
|
||||
## 要求
|
||||
1. 根据当前标签，生成 5-10 个相关的衍生标签
|
||||
2. 衍生标签应该包括：
|
||||
- 同义词/近义词标签
|
||||
- 上位概念标签（更宽泛的分类）
|
||||
- 下位概念标签（更具体的细分）
|
||||
- 相关联想标签（与主题相关但角度不同）
|
||||
- 应用场景标签（使用场景或人群）
|
||||
3. 标签要简洁，每个标签不超过10个字
|
||||
4. 标签要与医疗健康领域相关
|
||||
|
||||
## 输出格式
|
||||
请以JSON格式输出，包含以下字段：
|
||||
```json
|
||||
{{
|
||||
"original_tag": "原始标签",
|
||||
"derived_tags": [
|
||||
{{"tag": "衍生标签1", "type": "同义词", "relevance": "高"}},
|
||||
{{"tag": "衍生标签2", "type": "上位概念", "relevance": "高"}},
|
||||
...
|
||||
],
|
||||
"tag_description": "对原始标签的简要描述",
|
||||
"suggested_keywords": ["关键词1", "关键词2", ...]
|
||||
}}
|
||||
```
|
||||
"""
|
||||
|
||||
|
||||
class QwenTagDeriver:
|
||||
"""千问标签衍生生成器"""
|
||||
|
||||
def __init__(self, api_key: str = None):
|
||||
self.api_key = api_key or settings.qwen.api_key
|
||||
dashscope.api_key = self.api_key
|
||||
self.dao = get_image_tags_dao() # 使用统一的数据库配置
|
||||
|
||||
def get_image_tags_from_db(self, limit: int = 10, offset: int = 0) -> list:
|
||||
"""从数据库获取图片标签数据"""
|
||||
return self.dao.get_for_tag_derive(limit=limit, offset=offset)
|
||||
|
||||
def generate_prompt(self, tag_name: str, image_name: str = "", use_image: bool = False) -> str:
|
||||
"""生成提示词"""
|
||||
if use_image:
|
||||
return TAG_DERIVE_PROMPT_TEMPLATE.format(tag_name=tag_name)
|
||||
else:
|
||||
return TAG_DERIVE_TEXT_PROMPT_TEMPLATE.format(
|
||||
tag_name=tag_name,
|
||||
image_name=image_name
|
||||
)
|
||||
|
||||
def call_qwen_with_image(self, image_url: str, tag_name: str) -> dict:
|
||||
"""调用千问多模态模型（带图片）"""
|
||||
prompt = self.generate_prompt(tag_name, use_image=True)
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"image": image_url},
|
||||
{"text": prompt}
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
response = MultiModalConversation.call(
|
||||
model=settings.qwen.vision_model, # 千问视觉大模型
|
||||
messages=messages
|
||||
)
|
||||
|
||||
if response.status_code == HTTPStatus.OK:
|
||||
return {
|
||||
"success": True,
|
||||
"result": response.output.choices[0].message.content[0]["text"]
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Error: {response.code} - {response.message}"
|
||||
}
|
||||
|
||||
def call_qwen_text_only(self, tag_name: str, image_name: str = "") -> dict:
|
||||
"""调用千问文本模型（不带图片）"""
|
||||
from dashscope import Generation
|
||||
|
||||
prompt = self.generate_prompt(tag_name, image_name, use_image=False)
|
||||
|
||||
response = Generation.call(
|
||||
model=settings.qwen.text_model, # 使用配置中的文本模型
|
||||
prompt=prompt,
|
||||
result_format="message"
|
||||
)
|
||||
|
||||
if response.status_code == HTTPStatus.OK:
|
||||
return {
|
||||
"success": True,
|
||||
"result": response.output.choices[0].message.content
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Error: {response.code} - {response.message}"
|
||||
}
|
||||
|
||||
def derive_tags_for_image(self, image_data: dict, use_image: bool = False) -> dict:
|
||||
"""为单个图片生成衍生标签"""
|
||||
tag_name = image_data.get("tag_name", "")
|
||||
image_url = image_data.get("image_url", "")
|
||||
image_name = image_data.get("image_name", "")
|
||||
|
||||
print(f"\n处理标签: {tag_name}")
|
||||
print(f"图片URL: {image_url[:50]}..." if len(image_url) > 50 else f"图片URL: {image_url}")
|
||||
|
||||
if use_image and image_url:
|
||||
result = self.call_qwen_with_image(image_url, tag_name)
|
||||
else:
|
||||
result = self.call_qwen_text_only(tag_name, image_name)
|
||||
|
||||
return {
|
||||
"image_id": image_data.get("image_id"),
|
||||
"tag_id": image_data.get("tag_id"),
|
||||
"original_tag": tag_name,
|
||||
"image_url": image_url,
|
||||
"derive_result": result
|
||||
}
|
||||
|
||||
def batch_derive_tags(self, limit: int = 5, use_image: bool = False) -> list:
|
||||
"""批量生成衍生标签"""
|
||||
image_tags = self.get_image_tags_from_db(limit=limit)
|
||||
results = []
|
||||
|
||||
for item in image_tags:
|
||||
result = self.derive_tags_for_image(item, use_image)
|
||||
results.append(result)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def main():
|
||||
"""主函数 - 演示用法"""
|
||||
print("=" * 60)
|
||||
print("千问大模型 - 图片标签衍生生成器")
|
||||
print("=" * 60)
|
||||
|
||||
# 初始化
|
||||
deriver = QwenTagDeriver()
|
||||
|
||||
# 1. 从数据库获取数据示例
|
||||
print("\n[1] 从数据库获取图片标签数据...")
|
||||
try:
|
||||
image_tags = deriver.get_image_tags_from_db(limit=3)
|
||||
if image_tags:
|
||||
print(f"获取到 {len(image_tags)} 条数据:")
|
||||
for item in image_tags:
|
||||
print(f" - ID: {item['id']}, 标签: {item['tag_name']}")
|
||||
else:
|
||||
print("数据库中暂无数据")
|
||||
except Exception as e:
|
||||
print(f"数据库连接失败: {e}")
|
||||
image_tags = []
|
||||
|
||||
# 2. 生成提示词示例
|
||||
print("\n[2] 生成提示词示例:")
|
||||
sample_tag = "高血压"
|
||||
sample_prompt = deriver.generate_prompt(sample_tag, "blood_pressure.jpg")
|
||||
print("-" * 40)
|
||||
print(sample_prompt[:500] + "..." if len(sample_prompt) > 500 else sample_prompt)
|
||||
print("-" * 40)
|
||||
|
||||
# 3. 调用千问API（需要有效的API Key）
|
||||
print("\n[3] 调用千问API生成衍生标签...")
|
||||
if not settings.qwen.api_key or settings.qwen.api_key == "your-api-key-here":
|
||||
print("请先设置有效的 DASHSCOPE_API_KEY")
|
||||
print("可以通过环境变量设置: export DASHSCOPE_API_KEY=your-key")
|
||||
print("或修改 config/settings.py 中的配置")
|
||||
else:
|
||||
# 使用文本模式调用
|
||||
result = deriver.call_qwen_text_only(sample_tag, "示例图片")
|
||||
if result["success"]:
|
||||
print("生成结果:")
|
||||
print(result["result"])
|
||||
else:
|
||||
print(f"调用失败: {result['error']}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user