[bugfix][update_graph]

Author: lightislost

Dockerfile

@@ -1,4 +1,5 @@
 From python:3.9.18-bookworm
+# FROM python:3.9-slim-bookworm
 
 WORKDIR /home/user
 
@@ -11,10 +12,11 @@ COPY ./requirements.txt /home/user/docker_requirements.txt
 # 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 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
+RUN pip install fastapi uvicorn notebook
+# RUN pip install -r /home/user/docker_requirements.txt
 
 CMD ["bash"]

docker-compose.yaml

@@ -0,0 +1,170 @@
+version: '0.1'
+
+services:
+  metad0:
+    image: vesoft/nebula-metad:v3.8.0
+    container_name: metad0
+    environment:
+      USER: root
+    command:
+      - --meta_server_addrs=metad0:9559
+      - --local_ip=metad0
+      - --ws_ip=metad0
+      - --port=9559
+      - --ws_http_port=19559
+      - --data_path=/data/meta
+      - --log_dir=/logs
+      - --v=0
+      - --minloglevel=0
+    healthcheck:
+      test: ["CMD", "curl", "-sf", "http://metad0:19559/status"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 20s
+    ports:
+      - 9559:9559
+      - 19559:19559
+      - 19560
+    volumes:
+      - ./data/meta0:/data/meta
+      - ./logs/meta0:/logs
+    networks:
+      - ekg-net
+    restart: on-failure
+    cap_add:
+      - SYS_PTRACE
+
+  storaged0:
+    image: vesoft/nebula-storaged:v3.8.0
+    container_name: storaged0
+    environment:
+      USER: root
+      TZ:   "${TZ}"
+    command:
+      - --meta_server_addrs=metad0:9559
+      - --local_ip=storaged0
+      - --ws_ip=storaged0
+      - --port=9779
+      - --ws_http_port=19779
+      - --data_path=/data/storage
+      - --log_dir=/logs
+      - --v=0
+      - --minloglevel=0
+    depends_on:
+      - metad0
+    healthcheck:
+      test: ["CMD", "curl", "-sf", "http://storaged0:19779/status"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 20s
+    ports:
+      - 9779:9779
+      - 19779:19779
+      - 19780
+    volumes:
+      - ./data/storage0:/data/storage
+      - ./logs/storage0:/logs
+    networks:
+      - ekg-net
+    restart: on-failure
+    cap_add:
+      - SYS_PTRACE
+
+  graphd:
+    image: vesoft/nebula-graphd:v3.8.0
+    container_name: graphd
+    environment:
+      USER: root
+      TZ:   "${TZ}"
+    command:
+      - --meta_server_addrs=metad0:9559
+      - --port=9669
+      - --local_ip=graphd
+      - --ws_ip=graphd
+      - --ws_http_port=19669
+      - --log_dir=/logs
+      - --v=0
+      - --minloglevel=0
+    depends_on:
+      - storaged0
+    healthcheck:
+      test: ["CMD", "curl", "-sf", "http://graphd:19669/status"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 20s
+    ports:
+      - 9669:9669
+      - 19669:19669
+      - 19670
+    volumes:
+      - ./logs/graph:/logs
+    networks:
+      - ekg-net
+    restart: on-failure
+    cap_add:
+      - SYS_PTRACE
+
+  redis-stack:
+    image: redis/redis-stack:7.4.0-v0
+    container_name: redis
+    ports:
+      - "6379:6379"
+      - "8001:8001"
+    volumes:
+      - ./logs/redis:/var/lib/redis/logs
+    networks:
+      - ekg-net
+    restart: always
+
+
+  ollama:
+    image: ollama/ollama:0.3.6
+    container_name: ollama
+    environment:
+      USER: root
+      TZ:   "${TZ}"
+    ports:
+      - 11434:11434
+    volumes:
+      - //d/models/ollama:/root/.ollama # windows path
+      # - /User/models:/root/.ollama # linux/mac path
+    networks:
+      - ekg-net
+    restart: on-failure
+    cap_add:
+      - SYS_PTRACE
+    # deploy:
+    #   resources:
+    #     reservations:
+    #       devices:
+    #         - driver: nvidia
+    #           count: all  # 或者您想要的数量，例如 1
+    #           capabilities: [gpu]
+
+  ekgservice:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    container_name: ekgservice
+    image: muagent:test
+    environment:
+      USER: root
+      TZ:   "${TZ}"
+    ports:
+      - 5050:3737
+      # - 8080:8888
+    volumes:
+      - ./examples:/home/user/muagent/examples
+      - ./muagent:/home/user/muagent/muagent
+      - ./tests:/home/user/muagent/tests
+    restart: on-failure
+    networks:
+      - ekg-net
+    # command: ["python", "/home/user/muagent/examples/ekg_examples/start.py"]  # 指定要执行的脚本
+    command: ["python", "/home/user/muagent/tests/httpapis/fastapi_test.py"]  # 指定要执行的脚本
+
+networks:
+  ekg-net:

muagent/connector/memory_manager.py

@@ -655,7 +655,7 @@ def get_memory_pool_by_key_content(self, key: str, content: str, ):
         r = self.th.search(content)
         return self.tbasedoc2Memory(r)
 
-    def get_memory_pool_by_all(self, search_key_contents: dict):
+    def get_memory_pool_by_all(self, search_key_contents: dict, limit: int =10):
         '''
         search_key_contents:
             - key: str, key must in message keys
@@ -666,11 +666,17 @@ def get_memory_pool_by_all(self, search_key_contents: dict):
             if not v: continue
             if k == "keyword":
                 querys.append(f"@{k}:{{{v}}}")
+            elif k == "role_tags":
+                tags_str = '|'.join([f"*{tag}*" for tag in v]) if isinstance(v, list) else f"{v}"
+                querys.append(f"@role_tags:{tags_str}")
+            elif k == "start_datetime":
+                query = f"(@start_datetime:[{v[0]} {v[1]}])"
+                querys.append(query)
             else:
                 querys.append(f"@{k}:{v}")
 
         query = f"({')('.join(querys)})" if len(querys) >=2 else "".join(querys)
-        r = self.th.search(query)
+        r = self.th.search(query, limit=limit)
         return self.tbasedoc2Memory(r)
 
     def embedding_retrieval(self, text: str, top_k=1, score_threshold=1.0, chat_index: str = "default", **kwargs) -> List[Message]:
@@ -707,7 +713,7 @@ def text_retrieval(self, text: str, chat_index: str = "default", **kwargs)  -> L
         return self._text_retrieval_from_cache(memory.messages, text)
 
     def datetime_retrieval(self, chat_index: str, datetime: str, text: str = None, n: int = 5, key: str = "start_datetime", **kwargs) -> List[Message]:
-        intput_timestamp = datefromatToTimestamp(datetime, 1)
+        intput_timestamp = dateformatToTimestamp(datetime, 1000, "%Y-%m-%d %H:%M:%S.%f")
         query = f"(@chat_index:{chat_index})(@{key}:[{intput_timestamp-n*60} {intput_timestamp+n*60}])"
         # logger.debug(f"datetime_retrieval query: {query}")
         r = self.th.search(query)
@@ -787,8 +793,8 @@ def localMessage2TbaseMessage(self, message: Message):
         #     if content is not None:
         #         tbase_message["customed_kargs"][key] = content
 
-        tbase_message["start_datetime"] = datefromatToTimestamp(message.start_datetime, 1)
-        tbase_message["end_datetime"] = datefromatToTimestamp(message.end_datetime, 1)
+        tbase_message["start_datetime"] = dateformatToTimestamp(message.start_datetime, 1000, "%Y-%m-%d %H:%M:%S.%f")
+        tbase_message["end_datetime"] = dateformatToTimestamp(message.end_datetime, 1000, "%Y-%m-%d %H:%M:%S.%f")
 
         if self.use_vector and self.embed_config:
             vector_dict = get_embedding(
@@ -830,8 +836,8 @@ def tbasedoc2Memory(self, r_docs) -> Memory:
             memory.append(message)
 
         for message in memory.messages:
-            message.start_datetime = timestampToDateformat(int(message.start_datetime), 1)
-            message.end_datetime = timestampToDateformat(int(message.end_datetime), 1)
+            message.start_datetime = timestampToDateformat(int(message.start_datetime), 1000, "%Y-%m-%d %H:%M:%S.%f")
+            message.end_datetime = timestampToDateformat(int(message.end_datetime), 1000, "%Y-%m-%d %H:%M:%S.%f")
 
         memory.sort_by_key("end_datetime")
         # for message in memory.message:

muagent/connector/schema/message.py

@@ -79,9 +79,9 @@ def check_datetime(cls, values):
         start_datetime = values.get("start_datetime")
         end_datetime = values.get("end_datetime")
         if start_datetime is None:
-            values["start_datetime"] = getCurrentDatetime()
+            values["start_datetime"] = getCurrentDatetime("%Y-%m-%d %H:%M:%S.%f")
         if end_datetime is None:
-            values["end_datetime"] = getCurrentDatetime()
+            values["end_datetime"] = getCurrentDatetime("%Y-%m-%d %H:%M:%S.%f")
         return values
 
     @root_validator(pre=True)
@@ -99,7 +99,7 @@ def check_message_index(cls, values):
     def update_attribute(self, key: str, value):
         if hasattr(self, key):
             setattr(self, key, value)
-            self.end_datetime = getCurrentDatetime()
+            self.end_datetime = getCurrentDatetime("%Y-%m-%d %H:%M:%S.%f")
         else:
             raise AttributeError(f"{key} is not a valid property of {self.__class__.__name__}")
 

muagent/db_handler/graph_db_handler/base_gb_handler.py

@@ -64,5 +64,5 @@ def get_neighbor_nodes(self, attributes: dict, node_type: str = None, return_key
     def get_neighbor_edges(self, attributes: dict, node_type: str = None, return_keys: list = []) -> List[GEdge]:
         pass
 
-    def get_hop_infos(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: 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:
         pass

muagent/db_handler/graph_db_handler/geabase_handler.py

@@ -223,7 +223,7 @@ 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: 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， 表面需要至少两跳
         '''
@@ -267,30 +267,31 @@ def get_hop_infos(self, attributes: dict, node_type: str = None, hop: int = 2, b
         edges = self.convert2GEdges(result.get("e", []))
         return Graph(nodes=nodes, edges=edges, paths=result.get("p", []))
 
-    def get_hop_nodes(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: dict = []) -> List[GNode]:
+    def get_hop_nodes(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: List[dict] = []) -> List[GNode]:
         # 
         result = self.get_hop_infos(attributes, node_type, hop, block_attributes)
         return result.nodes
 
-    def get_hop_edges(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: dict = []) -> List[GEdge]:
+    def get_hop_edges(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: List[dict] = []) -> List[GEdge]:
         # 
         result = self.get_hop_infos(attributes, node_type, hop, block_attributes)
         return result.edges
 
-    def get_hop_paths(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: dict = []) -> List[str]:
+    def get_hop_paths(self, attributes: dict, node_type: str = None, hop: int = 2, block_attributes: List[dict] = []) -> List[str]:
         # 
         result = self.get_hop_infos(attributes, node_type, hop, block_attributes)
         return result.paths
 
-    def deduplicate_paths(self, result, block_attributes: dict = {}, select_attributes: dict = {}, hop:int=None, reverse=False):
+    def deduplicate_paths(self, result, block_attributes: List[dict] = {}, select_attributes: dict = {}, hop:int=None, reverse=False):
         # 获取数据
         n0, n1, e, p = result["n0"], result["n1"], result["e"], result["p"]
         block_node_ids = [
             i["id"]
             for i in n0+n1
+            for block_attribute in block_attributes
             # 这里block为空时也会生效，属于合理情况
             # if block_attributes=={} or all(item in i.items() for item in block_attributes.items())
-            if block_attributes and all(item in i.items() for item in block_attributes.items())
+            if block_attribute and all(item in i.items() for item in block_attribute.items())
         ] + [
             i["id"]
             for i in n0+n1

muagent/httpapis/ekg_construct/api.py

@@ -247,7 +247,7 @@ async def embedding_predict(request: EmbeddingsRequest):
 
 def create_api(llm, embeddings):
     app = init_app(llm, embeddings)
-    uvicorn.run(app, host="localhost", port=3737)
+    uvicorn.run(app, host="127.0.0.1", port=3737)
 
 # def create_api(ekg_construct_service: EKGConstructService):
 #     app = init_app(ekg_construct_service)