# -*- 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 # 调试模式 )