feat: ekg add new node and attributes

Author: lightislost

Dockerfile

@@ -4,15 +4,6 @@ WORKDIR /home/user
 
 COPY ./requirements.txt /home/user/docker_requirements.txt
 
-
-# RUN apt-get update
-# RUN apt-get install -y iputils-ping telnetd net-tools vim tcpdump
-# RUN echo telnet stream tcp nowait telnetd /usr/sbin/tcpd /usr/sbin/in.telnetd /etc/inetd.conf
-# RUN service inetutils-inetd start
-# service inetutils-inetd status
-
-# RUN wget https://oss-cdn.nebula-graph.com.cn/package/3.6.0/nebula-graph-3.6.0.ubuntu1804.amd64.deb
-# RUN dpkg -i nebula-graph-3.6.0.ubuntu1804.amd64.deb
 RUN pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
 RUN pip install -r /home/user/docker_requirements.txt --retries 5 --timeout 120
 

docker-compose.yaml

@@ -153,10 +153,12 @@ services:
       USER:
         root
     ports:
-      # - 5050:3737
       - 8080:8080
     networks:
       - ekg-net
+    volumes:
+      - ./runtime:/home/user/runtime
+    command: ["bash", "-c", "cd /home/user/runtime && mvn package && java -jar /home/user/runtime/bootstrap/muagent-runtime.jar"]
 
   ekgservice:
     build:

examples/ekg_examples/who_is_spy_game.py

@@ -145,11 +145,17 @@ def delete_edge(self, src_id, dst_id, edge_type: str = None) -> GbaseExecStatus:
 
     def delete_edges(self, id_pairs: List, edge_type: str = None) -> GbaseExecStatus:
         # geabase 不支持直接根据边关系进行检索
-        src_id, dst_id, timestamp = self.get_current_edgeID(src_id, dst_id, edge_type)
-        # src_id, dst_id = double_hashing(src_id), double_hashing(dst_id)
-        gql = f"MATCH ()-[e:{edge_type}{{@src_id:{src_id}, @dst_id:{dst_id}}}]->() DELETE e"
-        gql = f"MATCH (n:opsgptkg_intent )-[r]->(t1) DELETE r"
-        return self._get_crud_status(self.execute(gql))
+        try:
+            src_id, dst_id, timestamp = self.get_current_edgeID(src_id, dst_id, edge_type)
+            # src_id, dst_id = double_hashing(src_id), double_hashing(dst_id)
+            gql = f"MATCH ()-[e:{edge_type}{{@src_id:{src_id}, @dst_id:{dst_id}}}]->() DELETE e"
+            gql = f"MATCH (n:opsgptkg_intent )-[r]->(t1) DELETE r"
+            return self._get_crud_status(self.execute(gql))
+        except Exception as e:
+            return GbaseExecStatus(
+                errorMessage=e, 
+                errorCode=-1, 
+            )
 
     def get_nodeIDs(self, attributes: dict, node_type: str) -> List[int]:
         result = self.get_current_nodes(attributes, node_type)
@@ -236,22 +242,30 @@ def check_neighbor_exist(self, attributes: dict, node_type: str = None, check_at
         filter_result = [i for i in result if all([item in i.attributes.items() for item in check_attributes.items()])]
         return len(filter_result) > 0
 
-    def get_hop_infos(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: List[dict] = [], select_attributes: dict = {}, reverse=False) -> Graph:
+    def get_hop_infos(
+            self, 
+            attributes: dict, 
+            node_type: str = None, 
+            hop: int = 2, 
+            block_attributes: List[dict] = [], 
+            select_attributes: dict = {}, 
+            reverse=False
+        ) -> Graph:
         '''
-        hop >= 2， 表面需要至少两跳
+        hop >= 1
         '''
         hop_max = self.hop_max
         # 
         where_str = ' and '.join([f"n0.{k}='{v}'" for k, v in attributes.items()])
         if reverse:
-            gql = f"MATCH p = (n0:{node_type} WHERE {where_str})<-[e]-{{1,{min(hop, hop_max)}}}(n1) RETURN n0, n1, e, p"
+            gql = f"MATCH p = (n0:{node_type} WHERE {where_str})<-[e]-{{0,{min(hop, hop_max)}}}(n1) RETURN n0, n1, e, p"
         else:
-            gql = f"MATCH p = (n0:{node_type} WHERE {where_str})-[e]->{{1,{min(hop, hop_max)}}}(n1) RETURN n0, n1, e, p"
+            gql = f"MATCH p = (n0:{node_type} WHERE {where_str})-[e]->{{0,{min(hop, hop_max)}}}(n1) RETURN n0, n1, e, p"
         last_node_ids, last_node_types = [], []
 
         result = {}
         iter_index = 0
-        while hop > 1:
+        while hop >= 1:
             if last_node_ids == [] and iter_index==0:
                 # 
                 result = self.execute(gql)
@@ -272,7 +286,8 @@ def get_hop_infos(self, attributes: dict, node_type: str = None, hop: int = 2, b
 
                     result = self.merge_hotinfos(result, _result, reverse=reverse)
             # 
-            last_node_ids, last_node_types, result = self.deduplicate_paths(result, block_attributes, select_attributes, hop=min(hop, hop_max)+iter_index*hop_max, reverse=reverse)
+            last_node_ids, last_node_types, result = self.deduplicate_paths(
+                result, block_attributes, select_attributes, hop=min(hop, hop_max)+iter_index*hop_max, reverse=reverse)
             hop -= hop_max
             iter_index += 1
 
@@ -381,6 +396,7 @@ def decode_result(self, geabase_result, gql: str) -> Dict:
                 for col_data, rk, sk in zip(row["columns"], return_keys, save_keys):
                     _decode_func = decode_geabase_result_func_by_key.get(sk, self.decode_attribute)
                     # print(sk, json.dumps(col_data, ensure_ascii=False, indent=2))
+                    if col_data is None or col_data == {}: continue
                     decode_reuslt = _decode_func(col_data, rk)
                     if ".attr" in sk:
                         attr_dict.setdefault(sk, {}).update(decode_reuslt)
@@ -447,8 +463,17 @@ def decode_edge(self, col_data, k) -> Dict:
         def _decode_edge(data):
             edgeVal= data.get("edgeVal", {})
             edge_val_json = {
-                **{"SRCID": int(edgeVal.get("srcId", "")), "DSTID": int(edgeVal.get("dstId", "")), "type": edgeVal.get("type", "")}, 
-                **{k: v.get("strVal", "") if "strVal" in v else v.get("intVal", "0") for k, v in edgeVal.get("props", {}).items()}
+                **{
+                    "SRCID": int(edgeVal.get("srcId", "")), 
+                    "DSTID": int(edgeVal.get("dstId", "")), 
+                    "type": edgeVal.get("type", ""),
+                    "timestamp": int(edgeVal.get("timestamp", "1"))
+                    }, 
+                **{
+                    k: v.get("strVal", "") 
+                    if "strVal" in v else v.get("intVal", "0") 
+                    for k, v in edgeVal.get("props", {}).items()
+                }
             }
             # 存在业务逻辑
             edge_val_json["start_id"] = edge_val_json.pop("original_src_id1__")

muagent/db_handler/graph_db_handler/geabase_handler.py

@@ -10,6 +10,7 @@
 
 from muagent.service.ekg_construct.ekg_construct_base import EKGConstructService
 from muagent.schemas.apis.ekg_api_schema import *
+from muagent.schemas.ekg import *
 from muagent.service.ekg_reasoning.src.graph_search.graph_search_main import main
 
 
@@ -23,8 +24,39 @@ def wrapping_reponse(result, errorMessage="ok", success=0):
         success=success
     )
 
+def autofill_nodes(nodes: List[GNode]):
+    '''
+    兼容
+    '''
+    new_nodes = []
+    for node in nodes:
+        schema = TYPE2SCHEMA.get(node.type,)
+        node.attributes.update(node.attributes.pop("extra", {}))
+        node_data = schema(
+            **{**{"id": node.id, "type": node.type}, **node.attributes}
+        )
+        node_data = {
+            k:v
+            for k, v in node_data.dict().items()
+            if k not in ["type", "ID", "id", "extra"]
+        }
+        new_nodes.append(GNode(**{
+            "id": node.id, 
+            "type": node.type,
+            "attributes": {**node_data, **node.attributes}
+        }))
+    return new_nodes
+    
 # 
-def init_app(llm, llm_config, embeddings, ekg_construct_service: EKGConstructService, memory_manager, geabase_handler, intention_router):
+def init_app(
+        llm, 
+        llm_config, 
+        embeddings, 
+        ekg_construct_service: EKGConstructService, 
+        memory_manager, 
+        geabase_handler, 
+        intention_router
+):
 
     app = FastAPI()
 
@@ -179,8 +211,6 @@ async def text2graph(request: EKGT2GRequest):
 
 
     # ~/ekg/graph/update
-    # @app.post("/ekg/graph/update", response_model=EKGResponse)
-    # async def update_graph(request: UpdateGraphRequest):
     @app.post("/ekg/graph/update", response_model=EKGAIResponse)
     async def update_graph(request: EKGFeaturesRequest):
         logger.info(request.features.query)
@@ -196,14 +226,17 @@ async def update_graph(request: EKGFeaturesRequest):
 
         # 将 origin_nodes 和 nodes 转换为 GNode 对象
         origin_nodes = [GNode(**n) for n in origin_nodes]
+        origin_nodes = autofill_nodes(origin_nodes)
         nodes = [GNode(**n) for n in nodes]
-
+        nodes = autofill_nodes(nodes)
+                                      
         # 处理 origin_edges，给每个 edge 设置 type 字段
         origin_edges = [
             GEdge(
                 start_id=e['start_id'], 
                 end_id=e['end_id'],
-                type=f"{nodeid2type_dict.get(e['start_id'], 'unknown')}_route_{nodeid2type_dict.get(e['end_id'], 'unknown')}",  # 使用默认值 'unknown' 以防 id 不在字典中
+                 # 使用默认值 'unknown' 以防 id 不在字典中
+                type=f"{nodeid2type_dict.get(e['start_id'], 'unknown')}_route_{nodeid2type_dict.get(e['end_id'], 'unknown')}",
                 attributes=e.get("attributes", {})
             )
             for e in origin_edges
@@ -220,34 +253,6 @@ async def update_graph(request: EKGFeaturesRequest):
             for e in edges
         ]
 
-        # 将 GEdge 和 GNode 对象转换回字典以保持原有 JSON 格式
-        # origin_edges_dict = [
-        #     {
-        #         "start_id": edge.start_id,
-        #         "end_id": edge.end_id,
-        #         "type": edge.type,
-        #         "attributes": edge.attributes
-        #     }
-        #     for edge in origin_edges
-        # ]
-
-        # edges_dict = [
-        #     {
-        #         "start_id": edge.start_id,
-        #         "end_id": edge.end_id,
-        #         "type": edge.type,
-        #         "attributes": edge.attributes
-        #     }
-        #     for edge in edges
-        # ]
-
-        # # 更新 query 的内容，将 edges 和 originEdges 部分重新保存
-        # request.features.query['originEdges'] = origin_edges_dict
-        # request.features.query['edges'] = edges_dict
-
-
-        # query = UpdateGraphRequest(**request.features.query)
-
         # 添加预测逻辑的代码
         errorMessage = "ok"
         successCode = True
@@ -318,9 +323,6 @@ def get_graph(request: EKGFeaturesRequest):
                     nodeid=query.nodeid, node_type=query.nodeType, 
                     hop=query.hop
                 )
-
-            # nodes = graph.nodes.dict()
-            # edges = graph.edges.dict()
             nodes = graph.nodes
             edges = graph.edges
         except Exception as e:

muagent/httpapis/ekg_construct/api.py

@@ -4,6 +4,7 @@
 __all__ = [
     "EKGEdgeSchema", "EKGNodeSchema", 
     "EKGTaskNodeSchema", "EKGIntentNodeSchema", "EKGAnalysisNodeSchema", "EKGScheduleNodeSchema", "EKGPhenomenonNodeSchema",
+    "EKGToolTypeSchema", "EKGToolSchema", "EKGAgentSchema",
     "EKGNodeTbaseSchema", "EKGEdgeTbaseSchema", "EKGTbaseData", 
     "EKGGraphSlsSchema", "EKGSlsData",
     "SHAPE2TYPE", "TYPE2SCHEMA",