fix parallel transform bug of document (LazyAGI#282)

lazyllm/tools/rag/index.py

@@ -67,6 +67,7 @@ def _parallel_do_embedding(self, nodes: List[DocNode]) -> List[DocNode]:
                         if node.has_missing_embedding(k):
                             future = executor.submit(node.do_embedding, {k: self.embed[k]}) \
                                 if k not in node.embedding_state else executor.submit(node.check_embedding_state, k)
+                            node.embedding_state.add(k)
                             futures.append(future)
             if len(futures) > 0:
                 for future in concurrent.futures.as_completed(futures):

lazyllm/tools/rag/transform.py

@@ -79,13 +79,15 @@ def batch_forward(
     ) -> List[DocNode]:
         documents: List[DocNode] = documents if isinstance(documents, (tuple, list)) else [documents]
 
-        def impl(node):
-            splits = self(node, **kwargs)
-            for s in splits:
-                s.parent = node
-                s.group = node_group
-            node.children[node_group] = splits
-            return splits
+        def impl(node: DocNode):
+            with node._lock:
+                if node_group in node.children: return []
+                splits = self(node, **kwargs)
+                for s in splits:
+                    s.parent = node
+                    s.group = node_group
+                node.children[node_group] = splits
+                return splits
 
         if getattr(self, '_number_workers', 0) > 0:
             pool = ThreadPoolExecutor(max_workers=self._number_workers)