GeoLLM / Gradio.py
Pengfa Li
Upload folder using huggingface_hub
badcf3c verified
# -*- coding: utf-8 -*-
import gradio as gr
import pandas as pd
import json
import random
from LLM import zero_shot
from prompt_generate import generate_prompt_with_examples as generate_prompt
from prompt_generate import generate_prompt_with_best_matches as generate_prompt_b
def get_model_options():
"""获取可用的模型系列选项"""
return ['gpt', 'llama', 'qwen', 'deepSeek', 'gemini', 'claude']
def get_common_model_names(model_series):
"""根据模型系列返回常用的模型名称选项"""
model_names = {
'gpt': ['gpt-3.5-turbo', 'gpt-4o'],
'llama': ['meta-llama/Meta-Llama-3.1-405B-Instruct'],
'qwen': ['Qwen/Qwen2.5-72B-Instruct'],
'deepSeek': ['deepseek-ai/DeepSeek-V3', 'deepseek-ai/DeepSeek-R1'],
'gemini': ['gemini-1.5-pro-002'],
'claude': ['claude-3-5-haiku-20241022']
}
return model_names.get(model_series, [])
def get_prompt_templates():
"""获取预制的prompt模板"""
templates = {
"自定义": "",
"Zero-shot基础提取": """你是一名专业经验丰富的工程地质领域专家,你的任务是从给定的输入文本中提取"实体-关系-实体"三元组。关系类型包括24种:"出露于"、"位于"、"整合接触"、"不整合接触"、"假整合接触"、"断层接触"、"分布形态"、"大地构造位置"、"地层区划"、"出露地层"、"岩性"、"厚度"、"面积"、"坐标"、"长度"、"含有"、"所属年代"、"行政区划"、"发育"、"古生物"、"海拔"、"属于"、"吞噬"、"侵入"。提取过程请按照以下规范:
1. 输出格式:
严格遵循JSON数组,无额外文本,每个元素包含:
[
{
"entity1": "实体1",
"relation": "关系",
"entity2": "实体2"
}
]
2. 复杂关系处理:
- 若同一实体参与多个关系,需分别列出不同三元组""",
"知识引导增强提取": """你是一名专业经验丰富的工程地质领域专家,你的任务是从给定的输入文本中提取"实体-关系-实体"三元组。关系类型包括24种:"出露于"、"位于"、"整合接触"、"不整合接触"、"假整合接触"、"断层接触"、"分布形态"、"大地构造位置"、"地层区划"、"出露地层"、"岩性"、"厚度"、"面积"、"坐标"、"长度"、"含有"、"所属年代"、"行政区划"、"发育"、"古生物"、"海拔"、"属于"、"吞噬"、"侵入"。提取过程请按照以下规范:
1. 输出格式:
严格遵循JSON数组,无额外文本,每个元素包含:
[
{
"entity1": "实体1",
"relation": "关系",
"entity2": "实体2"
}
]
2. 复杂关系处理:
- 若同一实体参与多个关系,需分别列出不同三元组
3. 关系的解释:
出露于:指岩石或地层暴露在地表或近地表,未被覆盖或埋藏。例如:(晚奥陶世 - 志留世侵入岩,出露于,调查区南部)。
位于:确立地质单元在更大空间框架(行政区域/构造单元)中的从属关系。例如:(库穆奇志留世玄武岩基性岩石,位于,调查区中西部)
整合接触:指示上下地层连续沉积形成的接触关系,反映无显著沉积间断的岩性渐变特征。例如:(索拉克组,整合接触,中奥陶统林组)。
不整合接触:描述存在沉积缺失的地层接触界面,包含角度差异或岩性突变的接触特征。例如:(通子岩组,不整合接触,毛口组)。
假整合接触:特指产状一致的平行不整合类型,强调沉积序列的间断但无构造变形。例如:(索拉克组,假整合接触,中奥陶统林组)。
断层接触:两个地层之间为断层带或断层面,往往伴随动力破碎等构造现象。例如:(索拉克组,断层接触,上奥陶统拉排泉组)。
分布形态:刻画地质单元的空间展布特征,包括几何形态与延伸方向的组合描述。例如:(石炭系,分布形态,带状)。
大地构造位置:定位地质单元在板块构造格架中的归属,关联造山带或构造单元划分。例如:(石炭系,大地构造位置,冈瓦纳构造带北缘)。
地层区划:表征地层单元在区域地层划分体系中的层级归属与分区属性。例如:(石炭系,地层区划,冈瓦纳)。
出露地层:特指某区域实际暴露的地层实体,强调可观测的地表地质单元。例如:(红柳沟金铜矿区,出露地层,南华 - 下奥陶统红柳沟群)。
岩性:定义岩石的物质组成与结构特征,包含复合岩性的层级描述要素。例如:(晚奥陶世 - 志留世正长岩,岩性,蚀变正长岩)。
厚度:量化地层/岩体的垂向尺度,包含绝对数值与相对描述的量纲表达。例如:(正长岩,厚度,35.60 m)。
面积:表征地质单元的水平展布范围,以数值与单位组合的标准化形式呈现。例如:(侵入岩,出露面积,54 m2)
坐标:特指记录地质特征点的地理空间定位数据。例如:(索拉克铜金矿址,坐标,东经90°11′47″)。
长度:描述线性地质体的空间延伸尺度。例如: 可提取出三元组(什比恩断裂带,长度,20m)。
含有:指示主体物质的成分包含关系,专指矿物组成或化石赋存状态,不同于日常的意思。例如:(中灰黑色块状燧石,含有,燧石条带)。
所属年代:建立地质单元与标准地质年代体系的对应关系。例如:(红柳沟金铜矿区,所属年代,早 - 中二叠世)。
行政区划:界定地质实体在行政管理体系中的隶属层级与地域归属。例如:(调查区,行政区划,茶阳县)。
发育:描述地质构造或沉积特征的显现程度与形成状态强度。例如:(兰花瓮组,发育,水平层理)。
古生物:记录地层中赋存的化石生物信息,需包含完整拉丁学名与分类特征。例如:(地层,古生物,Lumu et a)。
海拔:量化地质特征点相对于海平面的高程数据,保留测量基准标识。例如:(索拉克铜金矿址,海拔,2800m)。
属于:确立地质单元在分类体系中的类型归属。例如:(矿区,属于,多金属矿化亚区)。
吞噬:表征侵入体对围岩的空间取代过程,反映岩浆活动的改造作用。例如:(任天堂岩组,吞噬,侏罗纪花岗岩)。
侵入:描述岩浆岩体贯入围岩的地质作用过程,包含接触变质等伴生现象。例如:(高州壳石组,侵入,片麻状花岗岩)。
4. 其他要点:
所有三元组关系必须是以上24种之一
关系的实体不能是动词或者介词等无意义的词。且描述岩石和地层等实体等要根据原文尽可能完整""",
}
return templates
def get_qa_prompt_templates():
"""获取QA模块的prompt模板"""
templates = {
"自定义": "",
"Zero-shot判断题": "请根据给定的文本判断对错。",
"Zero-shot问答题": "请根据给定的文本回答问题。",
"COT判断题": "请首先判断对错,并给出你的推理依据。",
"COT问答题": "请首先回答该问题,并给出你的推理依据。",
}
return templates
# 全局变量存储训练数据
_train_data = None
_text_series = None
_label_series = None
def load_train_data():
"""加载训练数据"""
global _train_data, _text_series, _label_series
if _train_data is None:
try:
_train_data = pd.read_json('./data/train_triples.json')
_text_series = _train_data['text']
_label_series = _train_data['triple_list']
except Exception as e:
print(f"加载训练数据失败: {e}")
return False
return True
def generate_random_context_prompt(user_text, num_examples):
"""生成随机上下文提示"""
if not load_train_data():
return "无法加载训练数据"
try:
random_prompt = generate_prompt(_text_series, _label_series, num_examples)
return f"以下是地质描述文本和三元组提取样例:\n\n{random_prompt}\n请根据样例提取三元组:\n{user_text}"
except Exception as e:
return f"生成随机上下文提示失败: {e}"
def generate_best_match_context_prompt(user_text, num_examples):
"""生成基于相似度的最佳匹配上下文提示"""
if not load_train_data():
return "无法加载训练数据"
try:
best_match_prompt = generate_prompt_b(_text_series, _label_series, user_text, num_examples)
if best_match_prompt.strip():
return f"以下是地质描述文本和三元组提取样例:\n\n{best_match_prompt}\n\n请根据样例提取三元组:\n{user_text}"
else:
return f"未找到匹配的样例,进行零样本提取:\n{user_text}"
except Exception as e:
return f"生成最佳匹配上下文提示失败: {e}"
def update_model_names(model_series):
"""当模型系列改变时更新模型名称下拉列表"""
names = get_common_model_names(model_series)
return gr.Dropdown(choices=names, value=names[0] if names else "", label="模型名称", allow_custom_value=True)
def update_prompt_content(template_name):
"""当prompt模板改变时更新内容"""
templates = get_prompt_templates()
content = templates.get(template_name, "")
return gr.Textbox(value=content, label="Prompt内容", lines=15, max_lines=25)
def update_qa_prompt_content(template_name):
"""当QA prompt模板改变时更新内容"""
templates = get_qa_prompt_templates()
content = templates.get(template_name, "")
return gr.Textbox(value=content, label="QA Prompt内容", lines=3, max_lines=10)
def call_llm_model(model_series, model_name, prompt_content, user_content, context_type, num_examples):
"""调用LLM模型的包装函数(三元组提取)"""
try:
if not model_series or not model_name:
return "请选择模型系列和模型名称"
if not user_content:
return "请输入要处理的文本内容"
# 根据上下文类型组合完整的输入内容
if context_type == "无上下文":
if prompt_content.strip():
full_content = prompt_content.strip() + "\n\n" + user_content
else:
full_content = user_content
elif context_type == "随机上下文":
context_prompt = generate_random_context_prompt(user_content, num_examples)
if prompt_content.strip():
full_content = prompt_content.strip() + "\n\n" + context_prompt
else:
full_content = context_prompt
elif context_type == "最佳匹配上下文":
context_prompt = generate_best_match_context_prompt(user_content, num_examples)
if prompt_content.strip():
full_content = prompt_content.strip() + "\n\n" + context_prompt
else:
full_content = context_prompt
else:
if prompt_content.strip():
full_content = prompt_content.strip() + "\n\n" + user_content
else:
full_content = user_content
response = zero_shot(model_series, model_name, full_content)
# 处理不同类型的返回值
if hasattr(response, 'content'):
return response.content
elif isinstance(response, dict) and 'content' in response:
return response['content']
elif isinstance(response, str):
return response
else:
return str(response)
except Exception as e:
return f"调用模型时出错: {str(e)}"
def call_qa_model(model_series, model_name, qa_prompt_content, geological_text, question_or_statement, qa_type):
"""调用LLM模型的包装函数(QA模块)"""
try:
if not model_series or not model_name:
return "请选择模型系列和模型名称"
if not geological_text:
return "请输入地质文本"
if not question_or_statement:
if qa_type == "判断题":
return "请输入需要判断的事实描述"
else:
return "请输入需要回答的问题"
# 组合完整的输入内容
if qa_type == "判断题":
if qa_prompt_content.strip():
full_content = f"{qa_prompt_content.strip()}\n\n地质文本:\n{geological_text}\n\n需要判断的事实描述:\n{question_or_statement}"
else:
full_content = f"地质文本:\n{geological_text}\n\n需要判断的事实描述:\n{question_or_statement}"
else: # 问答题
if qa_prompt_content.strip():
full_content = f"{qa_prompt_content.strip()}\n\n地质文本:\n{geological_text}\n\n问题:\n{question_or_statement}"
else:
full_content = f"地质文本:\n{geological_text}\n\n问题:\n{question_or_statement}"
response = zero_shot(model_series, model_name, full_content)
# 处理不同类型的返回值
if hasattr(response, 'content'):
return response.content
elif isinstance(response, dict) and 'content' in response:
return response['content']
elif isinstance(response, str):
return response
else:
return str(response)
except Exception as e:
return f"调用模型时出错: {str(e)}"
def create_interface():
"""创建Gradio界面"""
with gr.Blocks(title="GeoLLM 模型调用界面", theme=gr.themes.Soft()) as demo:
gr.Markdown("# 🚀 GeoLLM 地质智能分析平台")
gr.Markdown("集成三元组提取和智能问答功能的专业地质文本分析工具")
# 添加选项卡
with gr.Tabs():
# 三元组提取模块
with gr.TabItem("🔗 三元组提取", elem_id="triple_extraction"):
with gr.Row():
with gr.Column(scale=1):
# 模型选择区域
gr.Markdown("## 📋 模型配置")
model_series = gr.Dropdown(
choices=get_model_options(),
value="gpt",
label="模型系列",
info="选择要使用的模型系列"
)
model_name = gr.Dropdown(
choices=get_common_model_names("gpt"),
value="gpt-3.5-turbo",
label="模型名称",
info="选择具体的模型名称,也可以手动输入",
allow_custom_value=True
)
# 自定义模型名称输入框
custom_model_name = gr.Textbox(
label="自定义模型名称(可选)",
placeholder="如果上面的选项中没有您需要的模型,请在此输入",
info="此处输入的内容会覆盖上面的选择"
)
# Prompt模板选择
gr.Markdown("## 📝 Prompt模板")
prompt_template = gr.Dropdown(
choices=list(get_prompt_templates().keys()),
value="自定义",
label="选择Prompt模板",
info="选择预制的prompt模板或自定义"
)
# 上下文类型选择
gr.Markdown("## 🎯 上下文配置")
context_type = gr.Dropdown(
choices=["无上下文", "随机上下文", "最佳匹配上下文"],
value="无上下文",
label="上下文类型",
info="选择是否使用上下文样例"
)
num_examples = gr.Slider(
minimum=1,
maximum=3,
value=2,
step=1,
label="样例数量",
info="选择上下文样例的数量(1-3个)"
)
with gr.Column(scale=2):
# Prompt内容区域
gr.Markdown("## 🎯 Prompt内容")
prompt_content = gr.Textbox(
label="Prompt内容",
placeholder="选择模板或自定义您的prompt...",
lines=15,
max_lines=25,
info="将作为系统提示发送给模型"
)
# 用户输入区域
gr.Markdown("## 💬 地质文本输入")
user_content = gr.Textbox(
label="待处理的地质文本",
placeholder="请输入需要提取三元组的地质描述文本...",
lines=6,
max_lines=10
)
# 按钮和输出区域
with gr.Row():
clear_btn = gr.Button("🗑️ 清空", variant="secondary")
submit_btn = gr.Button("🚀 提取三元组", variant="primary")
# 输出区域
gr.Markdown("## 📤 提取结果")
output = gr.Textbox(
label="三元组提取结果",
lines=12,
max_lines=20,
interactive=False
)
# 示例区域
gr.Markdown("## 💡 使用示例")
gr.Examples(
examples=[
["gpt", "gpt-3.5-turbo", "无上下文", 2, "诺日巴尕日保组原指灰色灰绿色厚层中-细粒岩屑长石砂岩长石石英砂岩长石砂岩偶夹粉砂岩,粘土岩及泥晶灰岩组成,仅见双壳类化石,与上覆九十道班组为连续沉积。"],
["gemini", "gemini-1.5-pro-002", "随机上下文", 3, "雀莫错组测区内只在图幅西南角色旺涌曲一带有少量出露,面积不足 10m2,厚度大于 29.25m。"],
["claude", "claude-3-5-haiku-20241022", "最佳匹配上下文", 2, "灰岩中采集到 hecosmilia sp.剑鞘珊瑚; Complexastraea sp.和 Radulopccten sp.刮具海扇;Oscillopha sp.,时代为中侏罗世。"],
["deepSeek", "deepseek-ai/DeepSeek-V3", "最佳匹配上下文", 3, "晚三叠世花岗岩主要分布在测区拉地贡玛缅切日啊日曲一带,区域上受构造混杂带内的 NW-SE 向区域断裂控制,呈长条带状分布,侵入体具有良好的群居性,成带延展性非常好,出露侵入体 8 个,面积约 227m2。"],
],
inputs=[model_series, model_name, context_type, num_examples, user_content]
)
# 事件处理
def submit_request(series, name, custom_name, template, prompt, content, ctx_type, num_ex):
# 如果有自定义模型名称,使用自定义的
final_model_name = custom_name.strip() if custom_name.strip() else name
return call_llm_model(series, final_model_name, prompt, content, ctx_type, num_ex)
# 更新模型名称选项
model_series.change(
fn=update_model_names,
inputs=[model_series],
outputs=[model_name]
)
# 更新prompt内容
prompt_template.change(
fn=update_prompt_content,
inputs=[prompt_template],
outputs=[prompt_content]
)
# 提交按钮事件
submit_btn.click(
fn=submit_request,
inputs=[model_series, model_name, custom_model_name, prompt_template, prompt_content, user_content, context_type, num_examples],
outputs=[output]
)
# 清空按钮事件
clear_btn.click(
fn=lambda: ("", ""),
outputs=[user_content, output]
)
# 回车键提交
user_content.submit(
fn=submit_request,
inputs=[model_series, model_name, custom_model_name, prompt_template, prompt_content, user_content, context_type, num_examples],
outputs=[output]
)
# QA问答模块
with gr.TabItem("❓ 智能问答", elem_id="qa_module"):
with gr.Row():
with gr.Column(scale=1):
# 模型选择区域
gr.Markdown("## 📋 模型配置")
qa_model_series = gr.Dropdown(
choices=get_model_options(),
value="gpt",
label="模型系列",
info="选择要使用的模型系列"
)
qa_model_name = gr.Dropdown(
choices=get_common_model_names("gpt"),
value="gpt-3.5-turbo",
label="模型名称",
info="选择具体的模型名称,也可以手动输入",
allow_custom_value=True
)
# 自定义模型名称输入框
qa_custom_model_name = gr.Textbox(
label="自定义模型名称(可选)",
placeholder="如果上面的选项中没有您需要的模型,请在此输入",
info="此处输入的内容会覆盖上面的选择"
)
# QA类型选择
gr.Markdown("## 🎯 问答类型")
qa_type = gr.Dropdown(
choices=["判断题", "问答题"],
value="判断题",
label="任务类型",
info="选择是判断真假还是回答问题"
)
# QA Prompt模板选择
gr.Markdown("## 📝 Prompt模板")
qa_prompt_template = gr.Dropdown(
choices=list(get_qa_prompt_templates().keys()),
value="Zero-shot判断题",
label="选择QA Prompt模板",
info="选择预制的prompt模板或自定义"
)
with gr.Column(scale=2):
# QA Prompt内容区域
gr.Markdown("## 🎯 Prompt内容")
qa_prompt_content = gr.Textbox(
label="QA Prompt内容",
value="请根据给定的文本判断对错。",
placeholder="选择模板或自定义您的prompt...",
lines=3,
max_lines=10,
info="将作为系统提示发送给模型"
)
# 地质文本输入区域
gr.Markdown("## 📄 地质文本")
geological_text = gr.Textbox(
label="地质背景文本",
placeholder="请输入作为背景的地质描述文本...",
lines=8,
max_lines=15,
info="提供回答问题或判断事实的上下文信息"
)
# 问题或事实描述输入区域
gr.Markdown("## ❓ 问题/事实描述")
question_or_statement = gr.Textbox(
label="问题或事实描述",
placeholder="请输入需要回答的问题或需要判断的事实描述...",
lines=3,
max_lines=8,
info="根据任务类型输入相应内容"
)
# 按钮和输出区域
with gr.Row():
qa_clear_btn = gr.Button("🗑️ 清空", variant="secondary")
qa_submit_btn = gr.Button("🤖 开始问答", variant="primary")
# 输出区域
gr.Markdown("## 📤 问答结果")
qa_output = gr.Textbox(
label="模型回答",
lines=10,
max_lines=20,
interactive=False
)
# 示例区域
gr.Markdown("## 💡 使用示例")
# 判断题示例
with gr.Accordion("判断题示例", open=False):
gr.Examples(
examples=[
["gpt", "gpt-3.5-turbo", "判断题", "霍山县突发性地质灾害主要为崩塌、滑坡、泥石流。共查明突发性地质灾害点(含隐患点)190 处,其中崩塌 74 处,滑坡 96 个,泥石流 14 处,不稳定斜坡 6 处。新发现的地质灾害点有 58 处,占总数 30.5%。在霍山县 190 处崩塌、滑坡、泥石流等突发性地质灾害中,多数是人为因素造成的。人工因素造成的地质灾害有 163 处,占85.8%;自然因素形成的灾害有 27 处,占 14.2%。", "在霍山县的突发性地质灾害中,滑坡的数量超过了崩塌的数量。"],
["deepSeek", "deepseek-ai/DeepSeek-V3", "判断题", "霍山县突发性地质灾害主要为崩塌、滑坡、泥石流。共查明突发性地质灾害点(含隐患点)190 处,其中崩塌 74 处,滑坡 96 个,泥石流 14 处,不稳定斜坡 6 处。", "霍山县的地质灾害点总数超过200处。"],
],
inputs=[qa_model_series, qa_model_name, qa_type, geological_text, question_or_statement]
)
# 问答题示例
with gr.Accordion("问答题示例", open=False):
gr.Examples(
examples=[
["gpt", "gpt-3.5-turbo", "问答题", "霍山县突发性地质灾害主要为崩塌、滑坡、泥石流。共查明突发性地质灾害点(含隐患点)190 处,其中崩塌 74 处,滑坡 96 个,泥石流 14 处,不稳定斜坡 6 处。新发现的地质灾害点有 58 处,占总数 30.5%。", "霍山县共有多少处突发性地质灾害点?"],
["claude", "claude-3-5-haiku-20241022", "问答题", "霍山县突发性地质灾害主要为崩塌、滑坡、泥石流。共查明突发性地质灾害点(含隐患点)190 处,其中崩塌 74 处,滑坡 96 个,泥石流 14 处,不稳定斜坡 6 处。", "在霍山县的地质灾害中,哪种类型的灾害数量最多?"],
],
inputs=[qa_model_series, qa_model_name, qa_type, geological_text, question_or_statement]
)
# QA事件处理
def submit_qa_request(series, name, custom_name, q_type, template, prompt, geo_text, question):
# 如果有自定义模型名称,使用自定义的
final_model_name = custom_name.strip() if custom_name.strip() else name
return call_qa_model(series, final_model_name, prompt, geo_text, question, q_type)
def update_qa_prompt_on_type_change(qa_type_value):
"""当QA类型改变时更新prompt模板选项和内容"""
if qa_type_value == "判断题":
new_choices = ["自定义", "Zero-shot判断题", "COT判断题"]
new_value = "Zero-shot判断题"
new_prompt = "请根据给定的文本判断对错。"
new_placeholder = "请输入需要判断的事实描述..."
new_label = "事实描述"
else: # 问答题
new_choices = ["自定义", "Zero-shot问答题", "COT问答题"]
new_value = "Zero-shot问答题"
new_prompt = "请根据给定的文本回答问题。"
new_placeholder = "请输入需要回答的问题..."
new_label = "问题"
return (
gr.Dropdown(choices=new_choices, value=new_value, label="选择QA Prompt模板"),
gr.Textbox(value=new_prompt, label="QA Prompt内容", lines=3, max_lines=10),
gr.Textbox(label=new_label, placeholder=new_placeholder, lines=3, max_lines=8)
)
# 更新QA模型名称选项
qa_model_series.change(
fn=update_model_names,
inputs=[qa_model_series],
outputs=[qa_model_name]
)
# 更新QA prompt内容
qa_prompt_template.change(
fn=update_qa_prompt_content,
inputs=[qa_prompt_template],
outputs=[qa_prompt_content]
)
# QA类型改变时更新相关组件
qa_type.change(
fn=update_qa_prompt_on_type_change,
inputs=[qa_type],
outputs=[qa_prompt_template, qa_prompt_content, question_or_statement]
)
# QA提交按钮事件
qa_submit_btn.click(
fn=submit_qa_request,
inputs=[qa_model_series, qa_model_name, qa_custom_model_name, qa_type, qa_prompt_template, qa_prompt_content, geological_text, question_or_statement],
outputs=[qa_output]
)
# QA清空按钮事件
qa_clear_btn.click(
fn=lambda: ("", "", ""),
outputs=[geological_text, question_or_statement, qa_output]
)
# QA回车键提交
question_or_statement.submit(
fn=submit_qa_request,
inputs=[qa_model_series, qa_model_name, qa_custom_model_name, qa_type, qa_prompt_template, qa_prompt_content, geological_text, question_or_statement],
outputs=[qa_output]
)
return demo
if __name__ == "__main__":
# 启动界面
demo = create_interface()
demo.launch(
server_port=7860, # 端口号
share=True, # 是否创建公共链接
debug=True # 调试模式
)