Files
dsProject/dsLightRag/Util/GGB/Backup/GGB_3_QVQ.py

126 lines
4.6 KiB
Python
Raw Normal View History

2025-08-14 15:45:08 +08:00
import json
import logging
import requests
from Config.Config import OBS_AK, OBS_SERVER, OBS_SK, OBS_PREFIX, OBS_BUCKET, \
GLM_BASE_URL, GLM_API_KEY, GLM_MODEL_NAME
from Util.ObsUtil import ObsUploader
logger = logging.getLogger(__name__)
# 批量处理图片
def batch_glm(output_dir, img_list):
img_url_list = []
for file_path in img_list:
# 创建上传器实例
uploader = ObsUploader(OBS_AK, OBS_SK, "https://" + OBS_SERVER)
# 上传参数
object_key = OBS_PREFIX + "/" + file_path
# 执行上传
success, result = uploader.upload_file(OBS_BUCKET, object_key, file_path)
# 处理结果
if success:
logger.info(f'{file_path}上传成功!')
# 获取上传文件的 URL
file_url = f"https://{OBS_BUCKET}.{OBS_SERVER}/{object_key}"
img_url_list.append(file_url)
else:
logger.error(f'{file_path}上传失败!')
if 'errorCode' in result:
logger.info(f'错误代码: {result["errorCode"]}')
logger.info(f'错误信息: {result["errorMessage"]}')
else:
logger.error(f'错误信息: {result["error"]}')
# 多张图片开始解析
answer_content = ""
for img_url in img_url_list:
answer_content = answer_content + glm_single(img_url)
# 保存结果到JSON文件
qvq_result = f"{output_dir}/QvqResult.json"
with open(qvq_result, "w", encoding='utf-8') as f:
f.write(answer_content)
return qvq_result
def glm_single(img_url):
url = GLM_BASE_URL
headers = {
"Authorization": "Bearer " + GLM_API_KEY,
"Content-Type": "application/json"
}
prompt = """
### 几何图形识别专家指令(输入:纯几何图形照片)
**任务目标**
精确提取图形中的几何元素及其空间关系为GeoGebra重建建立数学模型
## 一、坐标系建立规则(必须遵守)
1. 原点设定
- 若存在明显顶点选最左下角的点为原点(0,0)
- 若图形对称选对称中心为原点
- 示例原点O = 三角形ABC的顶点A
2. 坐标轴定向
- 优先顺序水平线段 > 垂直线段 > 最长线段
- 具体规则
if 存在水平线段: 以该线段为x轴正方向
elif 存在垂直结构: 以最左侧垂直线为y轴
else: 以最长线段为基准轴
## 二、元素列举
1. 按点线三角形四边形梯形平行四边形矩形正方形圆等由简单到复杂的顺序列举所有图形
2. 详细描述元素之间的关系比如点D在线段AB上
3. 详细描述元素之间的位置关系比如D 在A点正上方B在CD边的上方中间位置
"""
data = {
"model": GLM_MODEL_NAME,
"messages": [
{
"role": "user",
# "content": "做水煮鱼一般用什么鱼"
"content":
[
{
"type": "text",
"text": prompt
},
{
"type": "image_url",
"image_url": {
"url": img_url
}
}
]
}
],
"stream": True # 启用流式调用
}
answer_content=''
with requests.post(url, headers=headers, json=data, stream=True) as response:
for chunk in response.iter_lines():
if chunk:
decoded = chunk.decode('utf-8')
if decoded.startswith('[DONE]'):
print("完成!")
break
try:
decoded = decoded[5:]
json_data = json.loads(decoded)
content = json_data["choices"][0]["delta"]['content']
if content and len(content) > 0:
print(content, end="")
answer_content=answer_content+"\n"+content
except Exception as e:
print(e)
# print("=" * 20 + "完整思考过程" + "=" * 20 + "\n")
# print(reasoning_content)
# print("=" * 20 + "完整回复" + "=" * 20 + "\n")
answer_content = answer_content.replace("```json", "")
answer_content = answer_content.replace("```", "")
return answer_content