上传文件至 kg_lab_6.13

6.19 agent迁移 对接

kg_lab_6.13/llm.py

@@ -0,0 +1,111 @@
+import requests
+from typing import Any, Dict, List, Optional
+from langchain_core.language_models import BaseLLM
+from langchain_core.callbacks import CallbackManagerForLLMRun
+from langchain_core.outputs import LLMResult, Generation
+from openai import OpenAI
+import httpx
+import logging
+
+
+# class OpenAiLLM:
+
+#     def __init__(self, url, api_key, model_name):
+#         self._api_key = api_key
+#         self._url = url
+#         self._model = model_name
+
+#     def generate(self, prompt):
+
+#         client = OpenAI(api_key=self._api_key, base_url=self._url)
+
+#         try:
+#             # 创建 Completion 请求
+#             completion = client.chat.completions.create(
+#                 model=self._model,
+#                 messages=[
+#                     {"role": "system", "content": "You are a helpful assistant"},
+#                     {"role": "user", "content": prompt}
+#                 ],
+#                 timeout=httpx.Timeout(300.0),
+#                 temperature=0.7,
+#             )
+#             return completion.choices[0].message.content
+#         except Exception as e:
+#             logging.error(f"LLM调用出错: {e}")
+#             return f"模型调用失败: {str(e)}"
+
+
+# llm = OpenAiLLM(
+#     url="http://172.20.0.145:9995/v1",
+#     api_key="xxx",
+#     model_name="deepseek-r1-distill-qwen2.5-32b",
+# )
+
+
+class Embedding:
+    def __init__(self, url, api_key, model_name):
+        self._api_key = api_key
+        self._url = url
+        self._model = model_name
+
+    def embed(self, text):
+        # 使用OpenAI客户端
+        client = OpenAI(api_key=self._api_key, base_url=self._url)
+
+        try:
+            # 调用embeddings API
+            response = client.embeddings.create(model=self._model, input=text, timeout=httpx.Timeout(60.0))
+            # 返回嵌入向量
+            return response.data[0].embedding
+        except Exception as e:
+            logging.error(f"嵌入模型调用出错: {e}")
+            raise RuntimeError(f"嵌入请求失败: {str(e)}")
+
+
+embedding = Embedding(url="http://10.1.16.39:9995/v1", api_key="xxx", model_name="bge-m3")
+
+
+class SiliconFlowLLM(BaseLLM):
+    """自定义硅基流动大模型调用类"""
+
+    api_url: str
+    api_key: str
+    model: str
+
+    def _generate(
+        self,
+        prompts: List[str],
+        stop: Optional[List[str]] = None,
+        run_manager: Optional[CallbackManagerForLLMRun] = None,
+        **kwargs: Any,
+    ) -> LLMResult:
+        from langchain_core.outputs import Generation, LLMResult
+
+        generations = []
+        for prompt in prompts:
+            try:
+                headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
+                payload = {
+                    "model": self.model,
+                    "messages": [{"role": "user", "content": prompt}],
+                }
+                response = requests.post(self.api_url, json=payload, headers=headers)
+                response.raise_for_status()
+                text = response.json()["choices"][0]["message"]["content"]
+                generations.append([Generation(text=text)])
+            except Exception as e:
+                raise Exception(f"调用硅基流动API失败: {str(e)}")
+
+        return LLMResult(generations=generations)
+
+    @property
+    def _llm_type(self) -> str:
+        return "siliconflow"
+
+
+llm = SiliconFlowLLM(
+    api_url="https://api.siliconflow.cn/v1/chat/completions",
+    api_key="sk-bbeamiumkouptsrueilgufqqyuumelcsivxwjbdugqwsqhwj",
+    model="Qwen/Qwen2.5-72B-Instruct",
+)

kg_lab_6.13/parameter_rewriting.py

@@ -0,0 +1,383 @@
+import json
+import ast
+import inspect
+
+from typing import Dict, Any, List, Optional, Union
+from langchain.agents import Tool, initialize_agent, AgentType
+from langchain.chat_models import ChatOpenAI
+from langchain.prompts import PromptTemplate
+from langchain.schema import BaseOutputParser
+from langchain.schema import BaseRetriever
+from langchain.callbacks.manager import CallbackManagerForRetrieverRun
+from langchain.schema import Document
+from langchain.chains import LLMChain
+from langchain.llms.base import BaseLLM
+
+from prompt_templates import FUNCTION_RETURNS_LOOP_PROMPT
+from llm import llm as base_llm
+import project  # 明确导入project模块
+
+
+class CodeOnlyOutputParser(BaseOutputParser):
+    def parse(self, text: str) -> dict:
+        if "Final Answer:" in text:
+            code_part = text.split("Final Answer:")[-1].strip()
+        else:
+            code_part = text.strip()
+
+        return {"code": code_part}
+
+    def get_format_instructions(self) -> str:
+        return "只输出最终的Python代码，不要包含其他解释或内容。"
+
+
+class KnowledgeGraphProcessor:
+    """知识图谱处理器"""
+
+    def __init__(self, kg_file="kg_simple_hierarchy.json"):
+        """初始化处理器"""
+        self.kg_data = self._load_kg_data(kg_file)
+        self.prompt = FUNCTION_RETURNS_LOOP_PROMPT
+
+    def _load_kg_data(self, file_path: str) -> Dict[str, Any]:
+        """加载知识图谱数据"""
+        try:
+            with open(file_path, "r", encoding="utf-8") as f:
+                return json.load(f)
+        except FileNotFoundError:
+            print(f"警告: 未找到 {file_path} 文件")
+            return {}
+        except Exception as e:
+            print(f"加载知识图谱数据时出错: {e}")
+            return {}
+
+    def _extract_parameters_from_code(self, code: str) -> List[str]:
+        """从代码中提取字符串参数"""
+        try:
+            tree = ast.parse(code)
+            parameters = []
+
+            for node in ast.walk(tree):
+                if isinstance(node, ast.Str):
+                    parameters.append(node.s)
+                elif isinstance(node, ast.Constant) and isinstance(node.value, str):
+                    parameters.append(node.value)
+
+            return parameters
+        except Exception as e:
+            print(f"解析代码时出错: {e}")
+            return []
+
+    def _find_node_by_path(self, path: str, data: Union[Dict[str, Any], List[Any]] = None) -> Optional[Dict[str, Any]]:
+        """
+        根据路径查找节点
+
+        Args:
+            path: 节点路径，格式如 "工程数据/安装过程/安装/架空输电线路本体工程/杆塔工程"
+            data: 当前数据节点
+
+        Returns:
+            找到的节点及其子节点
+        """
+        if data is None:
+            data = self.kg_data
+
+        path_parts = path.split("/")
+        current_part = path_parts[0]
+
+        # 如果data是列表，则遍历列表中的每个元素
+        if isinstance(data, list):
+            for item in data:
+                result = self._find_node_by_path(path, item)
+                if result:
+                    return result
+            return None
+
+        # 如果data是字典，则按原来的逻辑处理
+        if isinstance(data, dict):
+            # 检查当前节点是否匹配
+            found_node = None
+            for key, value in data.items():
+                if isinstance(value, str) and current_part in value:
+                    found_node = data
+                    break
+
+            # 如果在当前层级找到了节点
+            if found_node:
+                # 如果还有更深的路径部分，则继续递归查找
+                if len(path_parts) > 1:
+                    # 查找子节点
+                    if "children" in found_node:
+                        for child in found_node["children"]:
+                            result = self._find_node_by_path("/".join(path_parts[1:]), child)
+                            if result:
+                                return result
+                    return None
+                else:
+                    # 已经找到最终节点
+                    return found_node
+
+            # 如果当前层级没找到，则查找子节点
+            if "children" in data:
+                for child in data["children"]:
+                    result = self._find_node_by_path(path, child)
+                    if result:
+                        return result
+
+        return None
+
+    def _search_in_kg(self, parameter: str, data: Union[Dict[str, Any], List[Any]] = None) -> Optional[Dict[str, Any]]:
+        """在知识图谱中递归搜索参数"""
+        # 先尝试按路径查找
+        if "/" in parameter:
+            result = self._find_node_by_path(parameter)
+            if result:
+                return result
+
+        # 如果路径查找失败，则按关键词搜索
+        if data is None:
+            data = self.kg_data
+
+        # 如果是列表，遍历每个元素
+        if isinstance(data, list):
+            for item in data:
+                if isinstance(item, (dict, list)):
+                    result = self._search_in_kg(parameter, item)
+                    if result:
+                        return result
+            return None
+
+        # 如果是字典，按原来的逻辑处理
+        if isinstance(data, dict):
+            for key, value in data.items():
+                if isinstance(value, str) and parameter in value:
+                    return data
+                elif isinstance(value, list):
+                    for item in value:
+                        if isinstance(item, (dict, list)):
+                            result = self._search_in_kg(parameter, item)
+                            if result:
+                                return result
+                elif isinstance(value, dict):
+                    result = self._search_in_kg(parameter, value)
+                    if result:
+                        return result
+
+        return None
+
+    def _get_node_definition(self, node_type: str) -> str:
+        """获取节点类型定义"""
+        try:
+            # 检查project模块中是否存在该类
+            if hasattr(project, node_type):
+                cls = getattr(project, node_type)
+                source = inspect.getsource(cls)
+                return source
+            else:
+                return f"未找到类型 {node_type} 的定义"
+        except NameError:
+            return f"获取类型 {node_type} 定义时出错: project模块未正确导入"
+        except Exception as e:
+            return f"获取类型 {node_type} 定义时出错: {e}"
+
+    def _extract_node_types(self, data: Dict[str, Any], node_types: set):
+        """递归提取所有节点类型"""
+        if not isinstance(data, dict):
+            return
+
+        for key, value in data.items():
+            # 所有键都可能是节点类型
+            node_types.add(key)
+
+            if isinstance(value, dict):
+                self._extract_node_types(value, node_types)
+            elif isinstance(value, list):
+                for item in value:
+                    if isinstance(item, dict):
+                        self._extract_node_types(item, node_types)
+
+    def _get_relevant_knowledge(self, code: str) -> tuple[str, str]:
+        """根据代码获取相关知识库内容和节点定义"""
+        parameters = self._extract_parameters_from_code(code)
+
+        knowledge_base = ""
+        node_definitions = ""
+
+        for param in parameters:
+            # 处理【】格式的路径，提取最后一个部分
+            if "【" in param:
+                path_parts = []
+                parts = param.split("【")
+                for part in parts:
+                    if "】" in part:
+                        clean_part = part.split("】")[0].strip()
+                        if clean_part:
+                            path_parts.append(clean_part)
+
+                # 对于每个路径，只取最后一个部分
+                for path in path_parts:
+                    if "/" in path:
+                        # 提取路径中的最后一个部分
+                        last_part = path.split("/")[-1].strip()
+                        if last_part:
+                            found_data = self._search_in_kg(last_part)
+                            if found_data:
+                                knowledge_base += f"节点 '{last_part}' 相关信息:\n"
+                                knowledge_base += json.dumps(found_data, ensure_ascii=False, indent=2) + "\n\n"
+
+                                # 提取节点类型
+                                node_types = set()
+                                self._extract_node_types(found_data, node_types)
+
+                                for node_type in node_types:
+                                    definition = self._get_node_definition(node_type)
+                                    if definition and "未找到" not in definition and "出错" not in definition:
+                                        node_definitions += f"类型 {node_type} 定义:\n{definition}\n\n"
+                    else:
+                        # 如果没有/，直接使用整个部分
+                        found_data = self._search_in_kg(path)
+                        if found_data:
+                            knowledge_base += f"节点 '{path}' 相关信息:\n"
+                            knowledge_base += json.dumps(found_data, ensure_ascii=False, indent=2) + "\n\n"
+
+                            # 提取节点类型
+                            node_types = set()
+                            self._extract_node_types(found_data, node_types)
+
+                            for node_type in node_types:
+                                definition = self._get_node_definition(node_type)
+                                if definition and "未找到" not in definition and "出错" not in definition:
+                                    node_definitions += f"类型 {node_type} 定义:\n{definition}\n\n"
+            else:
+                # 处理普通参数
+                found_data = self._search_in_kg(param)
+                if found_data:
+                    knowledge_base += f"参数 '{param}' 相关信息:\n"
+                    knowledge_base += json.dumps(found_data, ensure_ascii=False, indent=2) + "\n\n"
+
+                    # 提取节点类型
+                    node_types = set()
+                    self._extract_node_types(found_data, node_types)
+
+                    for node_type in node_types:
+                        definition = self._get_node_definition(node_type)
+                        if definition and "未找到" not in definition and "出错" not in definition:
+                            node_definitions += f"类型 {node_type} 定义:\n{definition}\n\n"
+
+        return knowledge_base, node_definitions
+
+    def process_query(
+        self, original_query: str, original_code: str, error_info: Dict[str, str] = None
+    ) -> Dict[str, str]:
+        """
+        处理查询并尝试修复错误
+
+        Args:
+            original_query: 原始查询字符串
+            original_code: 原始代码字符串
+            error_info: 错误信息字典 (可选)
+
+        Returns:
+            Dict: 包含修改后的query和code的字典
+        """
+        print("\n====== RAG重写查询开始 ======")
+        print(f"原始查询: {original_query}")
+        print(f"原始代码: {original_code}")
+        print(f"错误信息: {error_info}")
+
+        # 初始化工具
+        tools = [
+            Tool(
+                name="search_knowledge_base",
+                func=lambda x: json.dumps(self._search_in_kg(x), ensure_ascii=False),
+                description="用于在知识图谱中搜索节点信息",
+            ),
+            Tool(
+                name="get_node_definition",
+                func=lambda x: self._get_node_definition(x),
+                description="获取指定类型的节点定义",
+            ),
+        ]
+
+        # 使用基础 LLM
+        llm = base_llm
+
+        # 初始化 Agent
+        agent_executor = initialize_agent(
+            tools,
+            llm,
+            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
+            verbose=True,
+            handle_parsing_errors=True,
+            prompt=self.prompt,
+            output_parser=CodeOnlyOutputParser(),
+        )
+
+        # 构造输入变量
+        knowledge_base, node_definitions = self._get_relevant_knowledge(original_code)
+
+        print(f"知识库内容: {knowledge_base}")
+        print(f"节点定义: {node_definitions}")
+
+        input_vars = {
+            "input": original_query,
+            "original_query": original_query,
+            "error_info": json.dumps(error_info, ensure_ascii=False) if error_info else "",
+            "original_code": original_code,
+            "KnowledgeBase": knowledge_base,
+            "NodeDefinition": node_definitions,
+        }
+
+        print(f"Agent输入变量: {input_vars}")
+
+        # 调用 Agent 执行
+        response = agent_executor.invoke(input_vars)
+
+        print(f"Agent响应: {response}")
+
+        # 处理不同的返回格式
+        if isinstance(response, dict):
+            if "output" in response:
+                if isinstance(response["output"], dict) and "code" in response["output"]:
+                    return {"query": original_query, "code": response["output"]["code"]}
+                elif isinstance(response["output"], str):
+                    return {"query": original_query, "code": response["output"]}
+
+        # 如果无法解析，返回原始代码
+        print(f"警告: 无法解析 Agent 响应，返回原始代码")
+        return {"query": original_query, "code": original_code}
+
+
+# 全局处理器实例（延迟初始化）
+_processor = None
+
+
+def _get_processor():
+    """获取处理器实例（单例模式）"""
+    global _processor
+    if _processor is None:
+        _processor = KnowledgeGraphProcessor()
+    return _processor
+
+
+def rewrite_query_parameters(
+    original_query: str, original_code: str, error_info: Dict[str, str] = None
+) -> Dict[str, str]:
+    """
+    重写查询参数的外部接口函数
+
+    Args:
+        original_query: 原始查询字符串
+        original_code: 原始代码字符串
+        error_info: 错误信息字典 (可选)
+
+    Returns:
+        Dict: 包含修改后的query和code的字典
+
+    Example:
+        result = rewrite_query_parameters("查询动态费用", 'search_node("动态费用")')
+        print(result["query"])  # 修改后的查询
+        print(result["code"])   # 修改后的代码
+    """
+    processor = _get_processor()
+    return processor.process_query(original_query, original_code, error_info)

kg_lab_6.13/project.py

@@ -0,0 +1,536 @@
+"""
+软件知识图谱类定义
+根据Ontology_Layer.txt文件中的知识图谱信息创建
+"""
+
+from abc import ABC, abstractmethod
+import json
+
+
+class ProjectTookiIt(ABC):
+    """
+    项目类(抽象基类)
+    描述: 代表整个项目结构的顶层容器
+    """
+
+    def __init__(self):
+        self.project_division_set = ProjectDivisionItem()  # 项目划分集对象
+
+    # 项目划分查询方法
+    @abstractmethod
+    def get_division_item_by_path(self, path):
+        """
+        通过路径获取项目划分对象
+
+        Args:
+            path (str): 项目划分节点的路径，以'/'分隔的多级节点路径
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回的项目划分对象数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表，如果未找到节点，则包含父节点下所有ProjectDivisionItem节点
+
+        Note:
+            当路径为空字符串时，会返回None
+        """
+        pass
+
+    @abstractmethod
+    def get_division_node_by_parent_and_name(self, parent_path, partial_name):
+        """
+        通过父节点路径和模糊节点名称获取项目划分对象，包括子节点
+
+        Args:
+            parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
+            partial_name (str): 目标节点的模糊或不完整名称
+
+        Returns:
+            Tuple[str, List[Dict[str, Any]], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (List[Dict[str, Any]]): 成功时返回的数据列表
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表
+        """
+        pass
+
+    # 工程量查询方法
+    @abstractmethod
+    def get_quantities_by_paths(self, paths_str):
+        """
+        获取指定项目路径下的工程量对象
+
+        Args:
+            paths_str (str): 以'/'分隔的多级节点路径
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]:
+                - status: 'success'或'error'
+                - data: 成功时返回的节点数据字典，失败时为{}
+                - error: 错误信息，成功时为''
+                - helper_info: 辅助信息列表（当目标节点不存在时返回父节点下的所有ProjectQuantity节点名称）
+        """
+        pass
+
+    @abstractmethod
+    def get_quantities_node_by_parent_and_code(self, parent_path, quantity_type=None, code=None):
+        """
+        通过父节点路径和编码获取工程量对象（定额、主材或设备），包括子节点
+
+        Args:
+            parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
+            quantity_type (str): 工程量类型('定额'、'主材'、'设备'或None表示所有类型)
+            code (str): 工程量编码，以'/'分隔的多个编码
+
+        Returns:
+            Tuple[str, List[Dict[str, Any]], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (List[Dict[str, Any]]): 成功时返回的数据列表
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表
+        """
+        pass
+
+    @abstractmethod
+    def get_quantities_node_by_parent_and_name(self, parent_path, partial_name, quantity_type=None):
+        """
+        通过父节点路径、模糊节点名称和类型获取工程量对象（主材或者设备），包括子节点
+
+        Args:
+            parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
+            partial_name (str): 目标节点的模糊或不完整名称
+            quantity_type (str): 工程量类型('定额'、'主材'、'设备'或None表示所有类型)
+
+        Returns:
+            Tuple[str, List[Dict[str, Any]], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (List[Dict[str, Any]]): 成功时返回的数据列表
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表
+        """
+        pass
+
+    # 材机查询方法
+    @abstractmethod
+    def get_material_equipment_by_path(self, paths_str):
+        """
+        通过路径获取材机对象（MaterialOrEquipment 类型节点）
+
+        Args:
+            path (str): 项目划分节点的路径，以'/'分隔的多级节点路径
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回的节点属性数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 如果未找到目标节点，则包含父节点下所有 MaterialOrEquipment 节点名称
+        """
+        pass
+
+    @abstractmethod
+    def get_material_equipment_by_parent_and_name(self, parent_path, partial_name):
+        """
+        通过父节点路径和模糊名称获取材机对象
+
+        Args:
+            parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
+            partial_name (str): 目标节点的模糊或不完整名称
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回的节点属性数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 如果未找到目标节点，则包含父节点下所有 MaterialOrEquipment 节点名称
+        """
+        pass
+
+    # 取费查询方法
+    @abstractmethod
+    def get_fee_template_by_path(self, paths_str):
+        """
+        通过路径获取取费节点
+
+        Args:
+            path (str): 项目划分节点的路径，以'/'分隔的多级节点路径
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回的节点属性数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 如果未找到目标节点，则包含父节点下所有 FeeCollection 节点名称
+        """
+        pass
+
+    @abstractmethod
+    def get_fee_template_by_parent_and_name(self, parent_path, partial_name):
+        """
+        通过父节点路径和节点名称获取取费节点
+
+        Args:
+            parent_path (str): 父节点的路径，以'/'分隔的多级节点路径
+            node_name (str): 目标节点的名称，例如：直接费
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回的节点属性数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 如果未找到父节点，则包含父节点的父节点下所有节点名称
+                    - 如果未找到目标节点，则包含父节点下所有 FeeCollection 节点名称
+                    - 否则为空
+        """
+        pass
+
+    # 费用表查询方法
+    @abstractmethod
+    def get_fee_schedule_on_auxiliary_expense_table(self, table_name, fee_name, fee: str):
+        """
+        在辅助费用表中查找费用
+
+        Args:
+            table_name (str): 费用表名称
+            fee_name (str): 要查找的费用名称（可能在多级子节点中）
+            fee_attribute (str): 费用值属性名
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回包含属性值的数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 第一步失败：返回"工程数据"节点下所有子节点名称
+                    - 第二步失败：返回费用表节点下所有子节点名称（递归查找）
+                    - 第三步失败：返回费用节点的所有可用属性名称
+                    - 否则为空
+        """
+        pass
+
+    @abstractmethod
+    def get_fee_schedule_on_other_expense_table(self, table_name, fee_name, fee):
+        """
+        在其它费用表中查找费用
+
+        Args:
+            table_name (str): 费用表名称
+            fee_name (str): 要查找的费用名称（可能在多级子节点中）
+            fee_attribute (str): 费用值属性名
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回包含属性值的数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 第一步失败：返回"工程数据"节点下所有子节点名称
+                    - 第二步失败：返回费用表节点下所有子节点名称（递归查找）
+                    - 第三步失败：返回费用节点的所有可用属性名称
+                    - 否则为空
+        """
+        pass
+
+    @abstractmethod
+    def get_fee_schedule_on_land_acquisition_fee_table_table(self, table_name, fee_name, fee):
+        """
+        在其中：土地征用费表中查找费用
+
+        Args:
+            table_name (str): 费用表名称
+            fee_name (str): 要查找的费用名称（可能在多级子节点中）
+            fee_attribute (str): 费用值属性名
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回包含属性值的数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 第一步失败：返回"工程数据"节点下所有子节点名称
+                    - 第二步失败：返回费用表节点下所有子节点名称（递归查找）
+                    - 第三步失败：返回费用节点的所有可用属性名称
+                    - 否则为空
+        """
+        pass
+
+    @abstractmethod
+    def get_fee_schedule_on_installation_price_difference_table(self, table_name, fee_name, fee):
+        """
+        在安装价差表中查找费用
+
+        Args:
+            table_name (str): 费用表名称
+            fee_name (str): 要查找的费用名称（可能在多级子节点中）
+            fee_attribute (str): 费用值属性名
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回包含属性值的数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 第一步失败：返回"工程数据"节点下所有子节点名称
+                    - 第二步失败：返回费用表节点下所有子节点名称（递归查找）
+                    - 第三步失败：返回费用节点的所有可用属性名称
+                    - 否则为空
+        """
+        pass
+
+    @abstractmethod
+    def get_fee_schedule_on_Engineering_Cost_table(self, table_name, fee_name, fee):
+        """
+        在工程费用表中查找费用
+
+        Args:
+            table_name (str): 费用表名称
+            fee_name (str): 要查找的费用名称（可能在多级子节点中）
+            fee_attribute (str): 费用值属性名
+
+        Returns:
+            Tuple[str, Dict[str, Any], str, List[Any]]: 包含状态、数据、错误信息和辅助信息的元组
+                - status (str): 状态，'success'或'error'
+                - data (Dict[str, Any]): 成功时返回包含属性值的数据
+                - error (str): 错误时的错误信息
+                - helper_info (List[Any]): 辅助信息列表：
+                    - 第一步失败：返回"工程数据"节点下所有子节点名称
+                    - 第二步失败：返回费用表节点下所有子节点名称（递归查找）
+                    - 第三步失败：返回费用节点的所有可用属性名称
+                    - 否则为空
+        """
+        pass
+
+
+class ProjectDivisionItem:
+    """
+    项目划分项
+    描述: 代表项目结构中的层级条目，具有自身的属性，并且可以包含子项目划分项或详细工作项
+    JSON对应: ProjectDivisionSet数组中的对象，或ProjectDivisionItem的children数组中type为"项目划分"的对象
+    """
+
+    def __init__(self):
+        self.GUID = None  # xsd:string
+        self.id = None  # xsd:string
+        self.name = None  # xsd:string
+        self.代码 = None  # xsd:string (可选)
+        self.费率 = None  # xsd:string
+        self.单位 = None  # xsd:string (可选)
+        self.取费表id = None  # xsd:string
+        self.颜色标记 = None  # xsd:string
+        self.取费表 = None  # xsd:string
+        self.合价含税 = None  # xsd:string (可选)
+        self.type = None  # xsd:string
+        self.专业类型 = None  # xsd:string
+        self.资源库列表 = None  # xsd:list (可选)
+        self.notCheck = None  # xsd:string (可选)
+
+
+class ProjectQuantity:
+    """
+    工程量
+    描述: 代表项目划分项(ProjectDivisionItem)下的具体工作单元或物料项，是定额、主材、设备的父类型
+    """
+
+    def __init__(self):
+        self.id = None  # xsd:string
+        self.类型 = None  # xsd:string ("0"为定额,"1"为主材,"5"为设备)
+        self.name = None  # xsd:string
+        self.编码 = None  # xsd:string
+        self.单位 = None  # xsd:string
+        self.数量 = None  # xsd:string
+        self.资源库名称 = None  # xsd:string
+        self.投标数量 = None  # xsd:string (可选)
+        self.投标单价 = None  # xsd:string (可选)
+        self.特征段 = None  # xsd:string (可选)
+        self.关联父级量 = None  # xsd:string (可选)
+        self.颜色标记 = None  # xsd:string (可选)
+        self.单价不含税 = None
+        self.cost_set = None  # xsd:CostSet
+
+
+class Ration(ProjectQuantity):
+    """
+    定额
+    描述: 代表一种标准的工程量条目，通常包含详细的工、料、机消耗标准
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.计算式 = None  # xsd:string
+        self.中标计算式 = None  # xsd:string
+        self.人工费 = None  # xsd:string
+        self.机械费 = None  # xsd:string
+        self.甲供材料费不含税 = None  # xsd:string
+        self.材料费 = None  # xsd:string
+        self.定额系数 = None  # xsd:string
+        self.人工系数 = None  # xsd:string
+        self.材料系数 = None  # xsd:string
+        self.机械系数 = None  # xsd:string
+        self.定额范围 = None  # xsd:string
+        self.定额章节名称 = None  # xsd:string
+        self.费用类型 = None  # xsd:string
+        self.甲供材料费含税 = None  # xsd:string
+        self.投标合价 = None  # xsd:string
+        self.其中甲供材料费 = None  # xsd:string
+        self.合价不含税 = None  # xsd:string
+        self.基价 = None  # xsd:string
+        self.所属定额库 = None  # xsd:string
+
+    def __str__(self):
+        """返回定额的字符串表示"""
+        attrs = {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
+        return json.dumps(attrs, ensure_ascii=False, indent=2)
+
+
+class Material(ProjectQuantity):
+    """
+    主材
+    描述: 代表工程中使用的主要材料
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.规格型号 = None  # xsd:string
+        self.损耗率 = None  # xsd:string
+        self.供货方 = None  # xsd:string
+        self.集中配送 = None  # xsd:string (可选)
+        self.单重 = None  # xsd:string (可选)
+        self.市场价不含税 = None  # xsd:string
+        self.市场价含税 = None  # xsd:string
+        self.单价含税 = None  # xsd:string
+        self.结算市场价不含税 = None  # xsd:string (可选)
+        self.结算市场价含税 = None  # xsd:string (可选)
+        self.基准价不含税 = None  # xsd:string (可选)
+        self.基准价含税 = None  # xsd:string (可选)
+        self.费用类型 = None  # xsd:string
+        self.增值税率 = None  # xsd:string (可选)
+        self.合价含税 = None  # xsd:string
+        self.合价不含税 = None  # xsd:string
+        self.线重 = None  # xsd:string (可选)
+        self.制造长度 = None  # xsd:string (可选)
+        self.截面积 = None  # xsd:string (可选)
+
+    def __str__(self):
+        """返回材料的字符串表示"""
+        attrs = {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
+        return json.dumps(attrs, ensure_ascii=False, indent=2)
+
+
+class Equipment(ProjectQuantity):
+    """
+    设备
+    描述: 代表工程中安装或使用的设备
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.规格型号 = None  # xsd:string
+        self.供货方 = None  # xsd:string
+        self.运杂费率 = None  # xsd:string (可选)
+        self.单价含税 = None  # xsd:string
+        self.设备类型 = None  # xsd:string (可选)
+        self.增值税率 = None  # xsd:string (可选)
+        self.合价含税 = None  # xsd:string
+        self.合价不含税 = None  # xsd:string
+
+
+class MaterialOrEquipment:
+    """
+    材机
+    描述: 代表DetailedWorkItem中所列出的具体材料、人工或机械设备及其详细信息
+    """
+
+    def __init__(self):
+        self.id = None  # xsd:string
+        self.编码 = None  # xsd:string
+        self.名称 = None  # xsd:string
+        self.单位 = None  # xsd:string
+        self.类型 = None  # xsd:string
+        self.供货方 = None  # xsd:string
+        self.预算价不含税 = None  # xsd:string
+        self.市场价不含税 = None  # xsd:string
+        self.预算价含税 = None  # xsd:string
+        self.市场价含税 = None  # xsd:string
+        self.结算预算价不含税 = None  # xsd:string
+        self.结算市场价不含税 = None  # xsd:string
+        self.结算预算价含税 = None  # xsd:string
+        self.结算市场价含税 = None  # xsd:string
+        self.暂估价 = None  # xsd:string
+        self.拆分 = None  # xsd:string
+        self.全口径市场价不含税 = None  # xsd:string
+        self.全口径市场价含税 = None  # xsd:string
+        self.商品砼 = None  # xsd:string
+        self.数量 = None  # xsd:string
+        self.是否未计价 = None  # xsd:string
+
+
+class FeeTableTemplateItem:
+    """
+    取费表模板项
+    描述:
+    """
+
+    def __init__(self):
+        self.name = None  # xsd:string
+        self.OutlayID = None  # xsd:string
+        self.type = None  # xsd:string
+        self.profession = None  # xsd:string
+
+
+class FeeCollection:
+    """
+    取费
+    描述:
+    """
+
+    def __init__(self):
+        self.name = None  # xsd:string
+        self.serialNumber = None  # xsd:string
+        self.code = None  # xsd:string
+        self.base = None  # xsd:string
+        self.rate = None  # xsd:string
+
+
+class FeeScheduleItem:
+    """
+    费用表项
+    描述:
+    """
+
+    def __init__(self):
+        self.name = None  # xsd:string
+
+
+class Fee:
+    """
+    取费
+    描述:
+    """
+
+    def __init__(self):
+        self.name = None  # xsd:string
+        self.serialNumber = None  # xsd:string
+        self.code = None  # xsd:string
+        self.rate = None  # xsd:string
+        self.amount = None  # xsd:string
+        self.输出 = None  # xsd:string (可选)
+        self.取费基数 = None  # xsd:string (可选)
+        self.编制依据 = None  # xsd:string (可选)
+        self.编码 = None  # xsd:string (可选)
+        self.备注 = None  # xsd:string (可选)
+        self.表一显示 = None  # xsd:string (可选)
+        self.报表输出 = None  # xsd:string (可选)
+        self.建安合计费 = None  # xsd:string (可选)
+        self.合计费 = None  # xsd:string (可选)
+        self.占总计 = None  # xsd:string (可选)
+        self.建筑费 = None  # xsd:string (可选)
+        self.安装费 = None  # xsd:string (可选)
+        self.设备费 = None  # xsd:string (可选)
+        self.其他费 = None  # xsd:string (可选)
+        self.施工费 = None  # xsd:string (可选)
+        self.单位投资 = None  # xsd:string (可选)

kg_lab_6.13/prompt_templates.py

@@ -0,0 +1,76 @@
+from langchain.prompts import PromptTemplate
+
+FUNCTION_CALL_TEMPLATE = """
+你是一个专业的Python工程师。我会给你一个用户问题，你需要将其转换为对应的Python代码
+
+可用工具:
+{tools}
+
+工具名称：
+{tool_names}
+
+# 工作流程
+1. 从用户问题中{query}提取关键信息（节点路径、节点类型、节点名称等）
+2. 使用工具查询知识图谱结构以理解可用节点和节点属性
+3. 根据查询结果选择最匹配的{project_class_methods}中的方法
+4. 生成可直接执行的Python代码
+
+# 代码模板（必须严格遵循）
+def neo4j_find_function():
+    project = ProjectBuilder.build()
+    status, data, error, helper_info = project.[SELECTED_METHOD]([PARAMETERS])
+    return status, data, error, helper_info
+
+# 执行规则
+- 每次只能调用一个工具或生成最终代码
+- 参数必须从用户问题或知识图谱查询结果中提取
+- 必须确保生成的代码可以直接执行
+- 禁止修改代码模板结构
+- 禁止添加任何注释或解释
+
+# 当前进度
+{agent_scratchpad}
+
+# 响应格式
+思考: 分析当前步骤需要做什么
+行动: 选择工具名称
+行动输入: 工具参数
+观察: 工具返回结果
+
+...（重复直到准备好生成代码）...
+
+思考: 已收集足够信息，可以生成代码
+Final Answer:
+def neo4j_find_function():
+    project = ProjectBuilder.build()
+    status, data, error, helper_info = project.[SELECTED_METHOD]([PARAMETERS])
+    return status, data, error, helper_info
+"""
+
+FUNCTION_CALL_PROMPT = PromptTemplate.from_template(FUNCTION_CALL_TEMPLATE)
+
+
+###########################################################################################################################################################################
+
+FUNCTION_RETURNS_LOOP_TEMPLATE = """
+
+你是一个专业的Python工程师。我会给你一段错误python代码和错误信息，你需要帮我修复这段出错的代码
+
+你的任务是：
+1. 根据需要修改的代码{original_code}和代码的错误信息{error_info}来对代码和参数进行修改
+2. 如果错误信息中是代码的逻辑出现错误，那么就需要对代码本身整体结构进行修改
+3. 如果是代码中参数出现问题了，那么就需要结合错误信息中的帮助信息（helper_info）来对代码总的参数进行修改
+4. 修复后的代码应该完整，可以直接执行，并且能够返回查询结果
+
+注意：
+- 必须只输出最终的Python代码，不要添加任何解释、注释、推理过程或自然语言描述。
+- 不要以“以下是修正后的代码”、“修改如下”等语句开头。
+- 不要输出任何其他无关的内容。
+- 输出格式必须完全符合指定的函数模板。
+- 如果无法根据已有信息进行修改，请原样返回原始代码。
+
+请输出你修补后的代码：
+"""
+
+
+FUNCTION_RETURNS_LOOP_PROMPT: PromptTemplate = PromptTemplate.from_template(FUNCTION_RETURNS_LOOP_TEMPLATE)