6.13提交，语义处理测试

kg_lab_6.13/chains_lab.py

@@ -0,0 +1,229 @@
+from langchain_openai import ChatOpenAI
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.prompts.prompt import PromptTemplate
+from langchain_core.output_parsers import JsonOutputParser
+
+qwen_llm = ChatOpenAI(
+    openai_api_base="https://api.siliconflow.cn/v1",
+    model_name="Qwen/Qwen2.5-72B-Instruct",
+    # sk-muuqautpcyuowjtgfecbnivqodlhzydtfslqkmwbknawejsx
+    openai_api_key="sk-bbeamiumkouptsrueilgufqqyuumelcsivxwjbdugqwsqhwj",
+    temperature=0.1
+)
+
+deep_v3 = ChatOpenAI(
+    openai_api_base="https://api.siliconflow.cn/v1",
+    model_name="deepseek-ai/DeepSeek-V3",
+    # sk-muuqautpcyuowjtgfecbnivqodlhzydtfslqkmwbknawejsx
+    openai_api_key="sk-bbeamiumkouptsrueilgufqqyuumelcsivxwjbdugqwsqhwj",
+    temperature=0.1
+)
+
+def Problem_rewrite():
+    PromptTemplate1 = """
+    请根据用户的输入内容，替换其中内容：  
+
+    【用户输入】：
+    {query}
+
+    【检索内容】：
+    {retriever}
+
+    【举例】：
+    用户输入：塔材装材费是多少？
+    检索内容：角钢塔_塔材装材费_元
+    得到的结果：【角钢塔_塔材装材费_元】是多少？
+
+    【要求】：
+    - 不允许输出任何解释、标点符号或额外内容，仅在原输入上进行替换
+    - 替换后的内容要加上【】
+    """
+
+    Prompt = ChatPromptTemplate.from_template(PromptTemplate1)
+
+    Chain = Prompt | deep_v3 | StrOutputParser()
+    # Chain = Prompt | llm | StrOutputParser()
+
+    return Chain 
+
+def question_answer():
+    PromptTemplate1 = """
+    请根据用户的输入内容，和检索到的信息，回答问题：  
+
+    【用户输入】：
+    {query}
+
+    【实际检索目标】：
+    {retriever_keywords}
+
+    【检索内容】：
+    {retriever_info}
+
+    【要求】：
+    - 不允许输出任何解释、标点符号或额外内容
+    - 要选找到和实际检索目标最接近的检索内容，直接读取相关信息进行解答，不做任何运算
+    - 检索到的信息不一定有用，如果和问题毫无相关性，则回答不会
+    """
+
+    Prompt = ChatPromptTemplate.from_template(PromptTemplate1)
+
+    Chain = Prompt | deep_v3 | StrOutputParser()
+    # Chain = Prompt | llm | StrOutputParser()
+
+    return Chain 
+
+
+def question_answer_calculation():
+    PromptTemplate1 = """
+    请根据用户的实际检索目标查到的内容，基于用户输入和计算公式回答问题：  
+
+    【用户输入】：
+    {query}
+
+    【实际检索目标】：
+    {retriever_keywords}
+
+    【计算公式】：
+    {calculation}
+
+    【检索内容】：
+    {retriever_info}
+
+    【要求】：
+    - 不允许输出任何解释、标点符号或额外内容
+    - 要选找到和实际检索目标最接近的检索内容，直接读取相关信息进行解答，不做任何运算
+    - 检索到的信息不一定有用，如果和问题毫无相关性，则回答不会
+    """
+
+    Prompt = ChatPromptTemplate.from_template(PromptTemplate1)
+
+    Chain = Prompt | deep_v3 | StrOutputParser()
+    # Chain = Prompt | llm | StrOutputParser()
+
+    return Chain 
+
+
+"""neo4j"""
+
+from langchain_community.graphs import Neo4jGraph
+
+
+
+graph = Neo4jGraph(
+    url="bolt://172.20.0.145:7687",
+    username="neo4j",
+    password="password",
+)
+
+graph.refresh_schema()
+
+
+from langchain.prompts import (
+    PromptTemplate,
+    SystemMessagePromptTemplate,
+    HumanMessagePromptTemplate,
+    ChatPromptTemplate,
+)
+
+cypher_generation_template = """
+# 任务：
+为 Neo4j 图数据库生成 Cypher 查询。
+
+# 说明：
+仅使用架构中提供的关系类型和属性。 不得使用架构中未提供的关系类型或属性。
+
+# 架构：
+{schema}
+
+# 注意：
+不得在回答中包含任何解释或道歉。 
+不得回答任何不要求构造 Cypher 查询的问题。
+回答中不得包含除生成的 Cypher 查询以外的任何文本。
+请确保查询中关系的方向正确，并正确为实体和关系设置别名。
+在查询中使用 WITH 为后续语句设置别名（例如，WITH v as visit, c.billing_amount as billing_amount）。
+如果需要进行数字除法运算，请确保对分母进行非零过滤。
+
+# 注意
+在查询关键字时，不能使用MATCH (n:PropertyNode)，要用最原始的MATCH (n)
+在查询中分析缺失属性时，必须使用 IS NULL 或 IS NOT NULL。 
+不得在查询中返回嵌套属性。
+不得在查询中包含 "GROUP BY" 语句。
+请确保为所有后续语句使用 WITH 设置别名（例如，WITH v as visit, c.billing_amount as billing_amount）。
+如果需要进行数字除法运算，请确保对分母进行非零过滤。
+
+# 示例：
+
+1.
+输入：
+查找一下【工程数据/安装工程/安装/架空输电线路本体工程/基础工程】的【定额】下的【YX2-1/YX2-2/YX2-3/YX2-4/YX2-5/YX2-6/YX2-7】
+输出：
+MATCH (root:ProjectDivisionItem [name: '架空输电线路本体工程'])-[:HAS_CHILD]->(base:ProjectDivisionItem [name: '基础工程'])
+MATCH path = (base)-[:HAS_CHILD]->(child)-[:HAS_COMPONENT*1..]->(component)
+WHERE ANY(prop IN keys(component) WHERE toString(component[prop]) =~ 'YX2-[1-7]')
+RETURN component
+
+2.
+输入：
+查找一下【工程数据/安装工程/安装/架空输电线路本体工程/杆塔工程/杆塔组立/铁塔、钢管杆组立】的【主材】下的【角钢】
+输出：
+MATCH (root:ProjectDivisionItem [name: '架空输电线路本体工程'])-[:HAS_CHILD]->(base1:ProjectDivisionItem [name: '杆塔工程'])-[:HAS_CHILD]->(base2:ProjectDivisionItem [name: '杆塔组立'])-[:HAS_CHILD]->(base3:ProjectDivisionItem [name: '铁塔、钢管杆组立'])
+MATCH path = (base3)-[:HAS_COMPONENT*1..]->(component)
+WHERE ANY(prop IN keys(component) WHERE toString(component[prop]) CONTAINS '角钢') 
+RETURN component
+
+3.
+输入：
+查找一下【架空输电线路本体工程/架线工程/导地线架设】和【架空输电线路本体工程/架线工程/导地线跨越架设】的【主材】下的【高导电率】
+输出：
+MATCH (root:ProjectDivisionItem [name: '架空输电线路本体工程'])
+-[:HAS_CHILD]->(base1:ProjectDivisionItem [name: '架线工程'])-[:HAS_CHILD]->(base:ProjectDivisionItem)
+WHERE base.name IN ['导地线架设', '导地线跨越架设']
+MATCH path = (base)-[:HAS_COMPONENT*1..]->(component)
+WHERE ANY(prop IN keys(component) WHERE toString(component[prop]) CONTAINS '高导电率') 
+RETURN component
+
+
+问题：
+{question}
+"""
+
+cypher_generation_prompt = PromptTemplate(
+    input_variables=["schema", "question"], template=cypher_generation_template
+)
+
+qa_generation_template = """你是一个助手，根据 Neo4j Cypher 查询的结果生成可读的回答。查询结果部分包含根据用户的自然语言问题生成的 Cypher 查询结果。提供的信息是权威的；你必须始终使用这些信息来构建回答，不得使用内部知识来质疑或更正这些信息。确保回答听起来像是对问题的直接回应。
+
+用户提出了以下问题：
+{question}
+
+运行了一个 Cypher 查询，生成了以下结果：
+{context}
+
+如果提供的信息是空的，就说明你不知道答案。
+空的信息看起来是这样的：[]
+
+如果查询结果不为空，你必须提供一个回答。
+
+如果有查询结果数据，绝不能说你没有正确的信息。如果用户提问时需要显示所有相关查询结果，确保你显示所有相关结果。你必须始终假设提供的查询结果与问题相关。回答时只能基于提供的查询结果构建答案。
+
+"""
+
+qa_generation_prompt = PromptTemplate(
+    input_variables=["context", "question"], template=qa_generation_template
+)
+
+from langchain.chains import GraphCypherQAChain
+
+booway_cypher_chain = GraphCypherQAChain.from_llm(
+    allow_dangerous_requests=True,
+    cypher_llm=deep_v3,                       # 用于生成Cypher查询的LLM
+    qa_llm=deep_v3,                           # 用于根据Cypher查询结果生成答案的LLM
+    graph=graph,
+    verbose=True,
+    qa_prompt=qa_generation_prompt,
+    cypher_prompt=cypher_generation_prompt,
+    validate_cypher=True,
+    top_k=100,
+    return_intermediate_steps=True
+)

kg_lab_6.13/data/data.txt

@@ -0,0 +1,145 @@
+测试1
+测试2
+电压等级
+工程编码
+工程名称
+工程时间
+线路曲折系数
+线路长度合计_折单
+路径长度_单回路长度
+路径长度_双回路长度
+路径长度_三回路长度
+路径长度_四回路长度
+杆塔总基数
+角钢塔_塔基数
+角钢塔_塔材量
+角钢塔_其中：高强钢塔材量
+角钢塔_塔材装材费
+角钢塔_塔材装材费_元
+角钢塔_其中：高强钢塔材费用
+角钢塔_其中：高强钢塔材费用_元
+钢管塔_塔基数
+钢管塔_塔材量
+钢管塔_钢管价格
+钢管塔_钢管价格_元
+钢管杆_塔基数
+钢管杆_塔材量
+钢管杆_钢管价格
+钢管杆_钢管价格_元
+水泥杆基数
+直线塔基数
+耐张转角塔基数
+海拔
+导线及线材_分裂数
+导线及线材_单根导线面积
+导线及线材_导线量
+导线及线材_其中：节能导线量
+导线及线材_导线装材费
+导线及线材_导线装材费_元
+导线及线材_其中：节能导线费用
+导线及线材_其中：节能导线费用_元
+导线及线材_导线类型
+设计风速
+覆冰厚度
+地形分布_平地
+地形分布_丘陵
+地形分布_河网
+地形分布_泥沼
+地形分布_山地
+地形分布_高山
+地形分布_沙漠
+地形分布_峻岭
+地质条件_普通土
+地质条件_坚土
+地质条件_松砂石
+地质条件_水坑
+地质条件_泥水坑
+地质条件_流沙坑
+地质条件_岩石爆破
+地质条件_岩石人工
+土石方总量
+土石方量_基坑
+土石方量_接地
+土石方量_基面
+各类基础数量占总塔基数比例_台阶式
+各类基础数量占总塔基数比例_板式
+各类基础数量占总塔基数比例_插入式
+各类基础数量占总塔基数比例_掏挖
+各类基础数量占总塔基数比例_岩石嵌固
+各类基础数量占总塔基数比例_锚杆
+各类基础数量占总塔基数比例_灌注桩
+各类基础数量占总塔基数比例_人工挖孔桩
+各类基础数量占总塔基数比例_其他
+台阶式基础基数
+板式基础基数
+插入式基础基数
+掏挖基础基数
+岩石嵌固基础基数
+锚杆基础基数
+灌注桩基础基数
+人工挖孔桩基础基数
+其他基础基数
+基础混凝土总量
+灌注桩基础混凝土量
+现浇基础混凝土量
+挖孔基础混凝土量
+基础护壁混凝土用量
+预制混凝土用量
+基础钢材量
+基础钢材价格
+本体费用合计
+本体工程人工费
+本体工程机械费
+基础工程费用
+杆塔工程费用
+接地工程费用
+架线工程费用
+附件工程费用
+辅助工程费用
+辅助设施工程
+其他费用合计
+建场费合计
+项目建设管理费合计
+其中：工程监理费
+项目建设技术服务费合计
+其中：项目前期工作费
+其中：勘察费
+其中：设计费
+其中：工程建设检测费
+生产准备费
+其中：安全文明施工费
+基本预备费
+静态投资
+建设期利息
+动态投资
+增值税抵扣税额
+本体费用合计_元
+本体工程人工费_本体_元
+本体工程人工费_调试_元
+本体工程机械费_本体_元
+本体工程机械费_调试_元
+基础工程费用_元
+杆塔工程费用_元
+接地工程费用_元
+架线工程费用_元
+附件工程费用_元
+辅助工程费用_元
+辅助工程费用_调试_元
+辅助设施工程_元
+其他费用合计_元
+建场费合计_元
+项目建设管理费合计_元
+其中：工程监理费_元
+项目建设技术服务费合计_元
+其中：项目前期工作费_元
+其中：勘察费_元
+其中：设计费_元
+其中：工程建设检测费_元
+生产准备费_元
+其中：安全文明施工费_线路_元
+其中：安全文明施工费_调试_元
+基本预备费_元
+静态投资_元
+建设期利息_元
+动态投资_元
+增值税抵扣税额_元

kg_lab_6.13/main.py

@@ -0,0 +1,37 @@
+from chains_lab import Problem_rewrite
+
+problem_rewrite = Problem_rewrite()
+
+from vector_lab import intersection_of_three_lists
+
+# input_str1 = "杆塔总基数是多少？"
+# input_str2 = "单回路长度是多少？"
+# input_str3 = "计算一下角钢塔的塔材装材费"
+# input_str4 = "计算一下土石方总量"
+# input_str5 = "板式塔基的各类基础数量占总塔基数比例是多少？"
+# input_str6 = "基础混凝土总量是多少"
+# input_str7 = "计算一下本体工程机械费"
+# ipout_str8 = "项目建设技术服务费合计"
+
+input_str = "项目建设技术服务费合计是多少？"
+results = intersection_of_three_lists(input_str)
+
+retriever = intersection_of_three_lists(input_str)[0]
+
+print(f"输入:{input_str}")
+
+keywords = problem_rewrite.invoke({"query":input_str, "retriever":retriever})
+
+print(f"输出:{keywords}")
+
+
+import json
+
+with open('./data/data.json', 'r', encoding='utf-8') as file:
+    data = json.load(file)
+
+from utils import find_target_item, find_target_items, pre_mapping, pre_mapping2
+
+input_neo4j = pre_mapping2(keywords, data)
+
+print(f"检索目标：{input_neo4j}")

kg_lab_6.13/pre.py

@@ -0,0 +1,12 @@
+with open("./data/data.json", 'r', encoding='utf-8') as json_file:
+    data_list = json.load(json_file)
+
+with open("./data/data.txt", 'w', encoding='utf-8') as txt_file:
+    for item in data_list:
+        if "指标名称" in item:
+            txt_file.write(item["指标名称"] + '\n')
+        else:
+            txt_file.write("无指标名称字段\n")
+
+with open("./data/data.txt", 'r', encoding='utf-8') as file:
+    txt_list = [line.strip() for line in file]

kg_lab_6.13/utils.py

@@ -0,0 +1,221 @@
+def Dictionary_content_mapping(input_str, data, key="指标名称"):
+    import re
+
+    match = re.search(r'【(.*?)】', input_str)
+
+    if match:
+        extracted = match.group(1)
+    else:
+        return None  # 如果没有匹配到，提前返回
+
+    for i in range(len(data)):
+        if data[i].get(key) == extracted:
+            return data[i]
+
+    return None
+
+def find_target_item(input_str, data):
+    result = [None, None]
+    temp = None
+    for item in data:
+        if isinstance(item["指标描述"], dict):
+            if item["指标描述"]["指标映射"][0] == input_str:
+                result[0] = item["指标描述"]["映射规则"]
+                temp = item["指标描述"]["指标映射"][0]
+
+    for item in data:
+        if item["指标名称"] == temp:
+            result[1] = item['指标描述']
+            return result
+
+def find_target_items(ele, input_str, data):
+    result = [None, ele]
+    for item in data:
+        if isinstance(item["指标描述"], dict):
+            if item["指标名称"] == input_str or item["指标描述"]["指标映射"][0] == input_str:
+                if len(item["指标描述"]["指标映射"]) == 1:
+                    result[0] = item["指标描述"]["指标映射"][0]
+    return result
+
+def judge_exists(input_str, data):
+    for item in data:
+        if isinstance(item["指标描述"], dict):
+            if item["指标名称"] == input_str or item["指标描述"]["指标映射"][0] == input_str:
+                return True
+    return False
+
+def judge_str(ceshi, data):
+    for item in data:
+        if isinstance(item["指标描述"], str):
+            if item["指标名称"] == ceshi["指标描述"]["指标映射"][0]:
+                return True
+    return False
+
+def pre_mapping(keywords, data):
+    import re
+
+    match = re.search(r'【(.*?)】', keywords)
+
+    if match:
+        extracted = match.group(1)
+
+    for i in range(len(data)):
+        if data[i]["指标名称"] == extracted:
+            ceshi = data[i]
+            break
+
+    if isinstance(ceshi["指标描述"], str):
+        return ceshi["指标描述"]
+
+    elif isinstance(ceshi["指标描述"], dict):
+        if ceshi["指标描述"]["映射规则"] == "1":
+            temp = ceshi["指标描述"]["指标映射"][0]
+            return f"模糊查找一下【{temp}】"
+        else:
+            if len(ceshi["指标描述"]["指标映射"]) == 1:
+                temp0, temp1 = find_target_item(ceshi["指标描述"]["指标映射"][0], data)
+                return f"{temp1}，换算规则：【{temp0}】"
+
+            elif len(ceshi["指标描述"]["指标映射"]) > 1:
+                
+                
+                result = []
+                if judge_str(ceshi, data) == True:
+                    for ele in ceshi["指标描述"]["指标映射"]:
+                        for item in data:
+                            if isinstance(item["指标描述"], str) and item["指标名称"] == ele:
+                                temp1 = item["指标描述"]
+                                temp2 = ceshi["指标描述"]["映射规则"]
+                                result.append(f"{temp1}，换算规则：【{temp2}】")
+
+                else:
+                
+                    for item in ceshi["指标描述"]["指标映射"]:
+                        temp0, temp1 = find_target_items(ceshi["指标描述"]["映射规则"], item, data)
+    
+                        if temp0 is None and temp1 is None:
+                            pass
+                        elif temp0 != None and temp1 is None:
+                            if judge_exists(item, data):
+                                temp1 = item
+                                # temp0 = find_target_items(ceshi["指标描述"]["映射规则"], item, data)
+                                result.append(f"模糊查找一下【{temp0}】，换算规则：【{temp1}】")
+                            else:
+                                continue
+    
+                        elif temp0 != None and temp1 != None:
+                            if judge_exists(item, data):
+                                # temp0, temp1 = find_target_items(ceshi["指标描述"]["映射规则"], item, data)
+                                result.append(f"模糊查找一下【{temp0}】，换算规则：【{temp1}】")
+                            else:
+                                continue
+
+                return result
+
+
+def extract_concrete_info(outputs):
+    import re
+    from typing import List
+
+    """
+    从多个句子中提取第一个“【】”作为查找信息，最后一个“【】”作为换算规则，
+    返回格式为：[合并的查找句子, 换算规则]
+    """
+    prefixes = []
+    suffix = ''
+
+    for item in outputs:
+        matches = re.findall(r'【([^】]+)】', item)
+        if len(matches) >= 2:
+            prefixes.append(f"查找一下【{matches[0]}】")
+            # 假设所有换算规则一致，取第一个即可
+            if not suffix:
+                suffix = f'换算规则：【{matches[-1]}】'
+
+    if not prefixes or not suffix:
+        return []
+
+    return ['; '.join(prefixes), suffix]
+
+
+def extract_query_prefix_list(input_list):
+    import re
+    """
+    输入一个字符串列表，提取每个字符串中符合格式的前缀内容（例如：'查找一下【样式】'）
+
+    参数:
+        input_list (list[str]): 包含描述性语句的字符串列表
+
+    返回:
+        list[str]: 提取出的前缀部分列表（如 '查找一下【大板式】'）
+    """
+    pattern = r'(查找一下【[^】]+】)'
+    return [re.match(pattern, text).group(1) for text in input_list if re.match(pattern, text)]
+
+
+
+def pre_mapping2(keywords, data):
+    import re
+
+    # 提取关键字中中括号内的内容
+    match = re.search(r'【(.*?)】', keywords)
+    if not match:
+        return "未找到匹配的关键字"
+
+    extracted = match.group(1)
+
+    # 查找对应的指标项
+    ceshi = next((item for item in data if item["指标名称"] == extracted), None)
+    if not ceshi:
+        return "未找到对应的指标"
+
+    desc = ceshi.get("指标描述")
+
+    if isinstance(desc, str):
+        return f"测试：{desc}"
+
+    elif isinstance(desc, dict):
+        mapping_rule = desc.get("映射规则")
+        mappings = desc.get("指标映射", [])
+
+        # 单一映射规则为1时
+        if mapping_rule == "1":
+            return f"模糊查找一下【{mappings[0]}】"
+
+        # 仅有一个映射项
+        if len(mappings) == 1:
+            temp0, temp1 = find_target_item(mappings[0], data)
+            if "【" in temp0:
+                return f"{temp1}，换算规则：【{temp0}】"
+            else:
+                return f"查找一下【{temp1}】，换算规则：【{temp0}】"
+
+
+        # 多个映射项
+        result = []
+        if judge_str(ceshi, data):
+            for ele in mappings:
+                item = next((d for d in data if d["指标名称"] == ele and isinstance(d["指标描述"], str)), None)
+                if item:
+                    result.append(f"{item['指标描述']}，换算规则：【{mapping_rule}】")
+        else:
+            for item_name in mappings:
+                temp0, temp1 = find_target_items(mapping_rule, item_name, data)
+
+                if temp0 is None and temp1 is None:
+                    continue
+
+                if judge_exists(item_name, data):
+                    if temp1 is None:
+                        temp1 = item_name
+                    result.append(f"模糊查找一下【{temp0}】，换算规则：【{temp1}】")
+
+        return result or "未匹配到有效的映射项"
+
+    else:
+        return "不支持的指标描述格式"
+
+
+
+    
+    

kg_lab_6.13/vector_lab.py

@@ -0,0 +1,39 @@
+import os
+from langchain_community.vectorstores import FAISS
+from langchain_huggingface import HuggingFaceEmbeddings
+
+with open("./data/data.txt", 'r', encoding='utf-8') as file:
+    txt_list = [line.strip() for line in file]
+
+embedding_path = "/data/Z_LLM_data/Embed_data/bge-m3"
+embeddings = HuggingFaceEmbeddings(model_name=embedding_path)
+
+faiss_archived = "./data/faiss_data/data"
+vectorstore_txt_faiss = FAISS.from_texts(txt_list, embeddings)
+vectorstore_txt_faiss.save_local(faiss_archived)
+
+retriever_txt_faiss1 = vectorstore_txt_faiss.as_retriever(search_kwargs={"k":3})
+retriever_txt_faiss2 = vectorstore_txt_faiss.as_retriever(
+    search_type="mmr",
+    search_kwargs={"k": 3,  # 检索结果
+                   "fetch_k": 1,  # 候选结果数量
+                   "lambda_mult": 0.5}  # 平衡指数，1为相关性；0为多样性
+)
+retriever_txt_faiss3 = vectorstore_txt_faiss.as_retriever(
+    search_type="similarity_score_threshold",
+    search_kwargs={"score_threshold": 0.5}
+)
+
+def intersection_of_three_lists(input_str):
+    list1 = retriever_txt_faiss1.invoke(input_str)
+    list2 = retriever_txt_faiss2.invoke(input_str)
+    list3 = retriever_txt_faiss3.invoke(input_str)
+    
+    def _intersection_of_three_lists(retrieval_results):
+        return [doc.page_content for doc in retrieval_results]
+        
+    list11 = _intersection_of_three_lists(list1)
+    list22 = _intersection_of_three_lists(list2)
+    list33 = _intersection_of_three_lists(list3)
+        
+    return list(set(list11) & set(list22) & set(list33))