You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hugegraph.apache.org by "weijinglin (via GitHub)" <gi...@apache.org> on 2025/09/13 05:34:37 UTC
[PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
weijinglin opened a new pull request, #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301
This PR refactors the Scheduler class to introduce a more flexible and extensible workflow scheduling mechanism. The main changes include:
- Introduced a pipeline pool using a dictionary to manage different workflow types (e.g., build_vector_index, graph_extract), each with its own GPipelineManager, flow, prepare, and post-processing functions.
- Added a schedule_flow method to dynamically select and execute workflows based on the flow name, supporting pipeline reuse and resource management.
- Refactored the build_vector_index and graph_extract flows to separate preparation, execution, and post-processing logic, improving modularity and maintainability.
- Updated related utility functions (graph_index_utils.py, vector_index_utils.py) to use the new schedule_flow interface.
- Improved error handling and logging for schema parsing and pipeline execution.
These changes lay the foundation for supporting more complex and agentic workflows in the future, while also improving the efficiency and scalability of the current pipeline execution framework.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301#discussion_r2447061378
##########
hugegraph-llm/pyproject.toml:
##########
@@ -85,3 +86,4 @@ allow-direct-references = true
[tool.uv.sources]
hugegraph-python-client = { workspace = true }
+pycgraph = { git = "https://github.com/ChunelFeng/CGraph.git", subdirectory = "python", rev = "main", marker = "sys_platform == 'linux'" }
Review Comment:
The pycgraph dependency is restricted to Linux only with `marker = \"sys_platform == 'linux'\"`. This may prevent the code from running on other platforms (Windows, macOS). Consider documenting this platform limitation or providing alternatives for cross-platform support.
##########
hugegraph-llm/src/hugegraph_llm/state/ai_state.py:
##########
@@ -0,0 +1,81 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from PyCGraph import GParam, CStatus
+
+from typing import Union, List, Optional, Any
+
+
+class WkFlowInput(GParam):
+ texts: Union[str, List[str]] = None # texts input used by ChunkSplit Node
+ language: str = None # language configuration used by ChunkSplit Node
+ split_type: str = None # split type used by ChunkSplit Node
+ example_prompt: str = None # need by graph information extract
+ schema: str = None # Schema information requeired by SchemaNode
Review Comment:
Corrected spelling of 'requeired' to 'required'.
```suggestion
schema: str = None # Schema information required by SchemaNode
```
##########
hugegraph-llm/src/hugegraph_llm/flows/scheduler.py:
##########
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+from typing import Dict, Any
+from PyCGraph import GPipelineManager
+from hugegraph_llm.flows.build_vector_index import BuildVectorIndexFlow
+from hugegraph_llm.flows.common import BaseFlow
+from hugegraph_llm.flows.graph_extract import GraphExtractFlow
+from hugegraph_llm.utils.log import log
+
+
+class Scheduler:
+ pipeline_pool: Dict[str, Any] = None
+ max_pipeline: int
+
+ def __init__(self, max_pipeline: int = 10):
+ self.pipeline_pool = {}
+ # pipeline_pool act as a manager of GPipelineManager which used for pipeline management
+ self.pipeline_pool["build_vector_index"] = {
+ "manager": GPipelineManager(),
+ "flow": BuildVectorIndexFlow(),
+ }
+ self.pipeline_pool["graph_extract"] = {
+ "manager": GPipelineManager(),
+ "flow": GraphExtractFlow(),
+ }
+ self.max_pipeline = max_pipeline
+
+ # TODO: Implement Agentic Workflow
+ def agentic_flow(self):
+ pass
+
+ def schedule_flow(self, flow: str, *args, **kwargs):
+ if flow not in self.pipeline_pool:
+ raise ValueError(f"Unsupported workflow {flow}")
+ manager = self.pipeline_pool[flow]["manager"]
+ flow: BaseFlow = self.pipeline_pool[flow]["flow"]
+ pipeline = manager.fetch()
+ if pipeline is None:
+ # call coresponding flow_func to create new workflow
Review Comment:
Corrected spelling of 'coresponding' to 'corresponding'.
```suggestion
# call corresponding flow_func to create new workflow
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301#discussion_r2447070061
##########
hugegraph-llm/src/hugegraph_llm/models/embeddings/init_embedding.py:
##########
@@ -17,10 +17,40 @@
Review Comment:
⚠️ **Important: Duplicate function definition**
The `get_embedding` function is defined twice in this file - once as a standalone function (lines 18-51) and once as a method in the `Embeddings` class (lines 60-78). This creates maintenance issues and confusion.
**Recommendation:**
1. Remove the duplicate standalone function if it's not being used
2. Or deprecate the `Embeddings` class if migrating to the new pattern
3. Update all callers to use the consistent API
##########
hugegraph-llm/src/hugegraph_llm/operators/common_op/check_schema.py:
##########
@@ -59,64 +63,270 @@ def _validate_schema(self, schema: Dict[str, Any]) -> None:
check_type(schema, dict, "Input data is not a dictionary.")
if "vertexlabels" not in schema or "edgelabels" not in schema:
log_and_raise("Input data does not contain 'vertexlabels' or 'edgelabels'.")
- check_type(schema["vertexlabels"], list, "'vertexlabels' in input data is not a list.")
- check_type(schema["edgelabels"], list, "'edgelabels' in input data is not a list.")
+ check_type(
+ schema["vertexlabels"], list, "'vertexlabels' in input data is not a list."
+ )
+ check_type(
+ schema["edgelabels"], list, "'edgelabels' in input data is not a list."
+ )
+
+ def _process_property_labels(self, schema: Dict[str, Any]) -> (list, set):
+ property_labels = schema.get("propertykeys", [])
+ check_type(
+ property_labels,
+ list,
+ "'propertykeys' in input data is not of correct type.",
+ )
+ property_label_set = {label["name"] for label in property_labels}
+ return property_labels, property_label_set
+
+ def _process_vertex_labels(
+ self, schema: Dict[str, Any], property_labels: list, property_label_set: set
+ ) -> None:
+ for vertex_label in schema["vertexlabels"]:
+ self._validate_vertex_label(vertex_label)
+ properties = vertex_label["properties"]
+ primary_keys = self._process_keys(
+ vertex_label, "primary_keys", properties[:1]
+ )
+ if len(primary_keys) == 0:
+ log_and_raise(f"'primary_keys' of {vertex_label['name']} is empty.")
+ vertex_label["primary_keys"] = primary_keys
+ nullable_keys = self._process_keys(
+ vertex_label, "nullable_keys", properties[1:]
+ )
+ vertex_label["nullable_keys"] = nullable_keys
+ self._add_missing_properties(
+ properties, property_labels, property_label_set
+ )
+
+ def _process_edge_labels(
+ self, schema: Dict[str, Any], property_labels: list, property_label_set: set
+ ) -> None:
+ for edge_label in schema["edgelabels"]:
+ self._validate_edge_label(edge_label)
+ properties = edge_label.get("properties", [])
+ self._add_missing_properties(
+ properties, property_labels, property_label_set
+ )
+
+ def _validate_vertex_label(self, vertex_label: Dict[str, Any]) -> None:
+ check_type(vertex_label, dict, "VertexLabel in input data is not a dictionary.")
+ if "name" not in vertex_label:
+ log_and_raise("VertexLabel in input data does not contain 'name'.")
+ check_type(
+ vertex_label["name"], str, "'name' in vertex_label is not of correct type."
+ )
+ if "properties" not in vertex_label:
+ log_and_raise("VertexLabel in input data does not contain 'properties'.")
+ check_type(
+ vertex_label["properties"],
+ list,
+ "'properties' in vertex_label is not of correct type.",
+ )
+ if len(vertex_label["properties"]) == 0:
+ log_and_raise("'properties' in vertex_label is empty.")
+
+ def _validate_edge_label(self, edge_label: Dict[str, Any]) -> None:
+ check_type(edge_label, dict, "EdgeLabel in input data is not a dictionary.")
+ if (
+ "name" not in edge_label
+ or "source_label" not in edge_label
+ or "target_label" not in edge_label
+ ):
+ log_and_raise(
+ "EdgeLabel in input data does not contain 'name', 'source_label', 'target_label'."
+ )
+ check_type(
+ edge_label["name"], str, "'name' in edge_label is not of correct type."
+ )
+ check_type(
+ edge_label["source_label"],
+ str,
+ "'source_label' in edge_label is not of correct type.",
+ )
+ check_type(
+ edge_label["target_label"],
+ str,
+ "'target_label' in edge_label is not of correct type.",
+ )
Review Comment:
⚠️ **Important: Silent failure in key processing**
The `_process_keys` method silently filters out keys that don't exist in the properties list (`new_keys = [key for key in keys if key in label['properties']]`). This could hide configuration errors where users specify invalid keys.
**Issues:**
1. Typos in key names go undetected
2. No feedback when configured keys are ignored
3. Could lead to unexpected schema behavior
**Recommendation:**
Add validation to warn or error on invalid keys:
```python
def _process_keys(self, label: Dict[str, Any], key_type: str, default_keys: list) -> list:
keys = label.get(key_type, default_keys)
check_type(keys, list, f"'{key_type}' in {label['name']} is not of correct type.")
invalid_keys = [key for key in keys if key not in label['properties']]
if invalid_keys:
log.warning(f"Keys {invalid_keys} in {key_type} are not present in properties for {label['name']}")
new_keys = [key for key in keys if key in label['properties']]
return new_keys
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301#discussion_r2447073674
##########
hugegraph-llm/src/hugegraph_llm/flows/common.py:
##########
@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from abc import ABC, abstractmethod
+
+from hugegraph_llm.state.ai_state import WkFlowInput
+
+
+class BaseFlow(ABC):
+ """
+ Base class for flows, defines three interface methods: prepare, build_flow, and post_deal.
+ """
+
+ @abstractmethod
+ def prepare(self, prepared_input: WkFlowInput, *args, **kwargs):
Review Comment:
🧹 **Minor: Missing type hints**
The `BaseFlow` abstract class and its implementations lack type hints for parameters and return values. This reduces code maintainability and IDE support.
**Suggestion:**
```python
from typing import Any
from PyCGraph import GPipeline
@abstractmethod
def prepare(self, prepared_input: WkFlowInput, *args: Any, **kwargs: Any) -> None:
"""Pre-processing interface."""
pass
@abstractmethod
def build_flow(self, *args: Any, **kwargs: Any) -> GPipeline:
"""Interface for building the flow."""
pass
@abstractmethod
def post_deal(self, pipeline: GPipeline | None = None) -> str:
"""Post-processing interface."""
pass
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301#discussion_r2447068804
##########
hugegraph-llm/src/hugegraph_llm/flows/build_vector_index.py:
##########
@@ -0,0 +1,55 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from hugegraph_llm.flows.common import BaseFlow
+from hugegraph_llm.state.ai_state import WkFlowInput
+
+import json
+from PyCGraph import GPipeline
+
+from hugegraph_llm.operators.document_op.chunk_split import ChunkSplitNode
+from hugegraph_llm.operators.index_op.build_vector_index import BuildVectorIndexNode
+from hugegraph_llm.state.ai_state import WkFlowState
+
+
+class BuildVectorIndexFlow(BaseFlow):
+ def __init__(self):
+ pass
+
+ def prepare(self, prepared_input: WkFlowInput, texts):
Review Comment:
‼️ **Critical: Hardcoded configuration breaks flexibility**
The `prepare` method hardcodes `language = "zh"` (Chinese) and `split_type = "paragraph"`. This contradicts the PR's goal of "flexible and extensible workflow scheduling."
**Issues:**
1. Non-Chinese users cannot use this flow without code modification
2. No way to customize split strategy for different document types
3. Breaks the principle of configuration over convention
**Recommendation:**
Make these configurable parameters:
```python
def prepare(self, prepared_input: WkFlowInput, texts, language="zh", split_type="paragraph"):
prepared_input.texts = texts
prepared_input.language = language
prepared_input.split_type = split_type
```
##########
hugegraph-llm/src/hugegraph_llm/flows/scheduler.py:
##########
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+from typing import Dict, Any
+from PyCGraph import GPipelineManager
+from hugegraph_llm.flows.build_vector_index import BuildVectorIndexFlow
+from hugegraph_llm.flows.common import BaseFlow
+from hugegraph_llm.flows.graph_extract import GraphExtractFlow
+from hugegraph_llm.utils.log import log
+
+
+class Scheduler:
+ pipeline_pool: Dict[str, Any] = None
+ max_pipeline: int
+
+ def __init__(self, max_pipeline: int = 10):
+ self.pipeline_pool = {}
+ # pipeline_pool act as a manager of GPipelineManager which used for pipeline management
+ self.pipeline_pool["build_vector_index"] = {
+ "manager": GPipelineManager(),
+ "flow": BuildVectorIndexFlow(),
+ }
+ self.pipeline_pool["graph_extract"] = {
+ "manager": GPipelineManager(),
+ "flow": GraphExtractFlow(),
+ }
+ self.max_pipeline = max_pipeline
+
+ # TODO: Implement Agentic Workflow
+ def agentic_flow(self):
+ pass
+
+ def schedule_flow(self, flow: str, *args, **kwargs):
+ if flow not in self.pipeline_pool:
+ raise ValueError(f"Unsupported workflow {flow}")
+ manager = self.pipeline_pool[flow]["manager"]
+ flow: BaseFlow = self.pipeline_pool[flow]["flow"]
+ pipeline = manager.fetch()
+ if pipeline is None:
+ # call coresponding flow_func to create new workflow
+ pipeline = flow.build_flow(*args, **kwargs)
Review Comment:
‼️ **Critical: Incomplete error recovery logic**
When `pipeline.init()` or `pipeline.run()` fails, the code raises an exception but never calls `manager.add(pipeline)` to return the pipeline to the pool. This creates a **resource leak** where failed pipelines are never recycled.
**Impact:** After `max_pipeline` failures, the system runs out of pipelines and cannot process new requests.
**Recommendation:**
Use try-finally to ensure pipelines are always returned:
```python
pipeline = flow.build_flow(*args, **kwargs)
try:
status = pipeline.init()
if status.isErr():
raise RuntimeError(f"Error in flow init: {status.getInfo()}")
status = pipeline.run()
if status.isErr():
raise RuntimeError(f"Error in flow execution: {status.getInfo()}")
res = flow.post_deal(pipeline)
return res
finally:
manager.add(pipeline)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301#discussion_r2447072956
##########
hugegraph-llm/src/hugegraph_llm/flows/__init__.py:
##########
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
Review Comment:
🧹 **Minor: Empty `__init__.py` with only license header**
This file contains only the Apache license header with no actual code. While functional, consider adding a brief module docstring to describe the flows package purpose.
**Suggestion:**
```python
# ... (license header) ...
"""
Workflow orchestration module for HugeGraph AI.
This package provides flexible workflow scheduling and pipeline management
for various AI tasks including vector indexing and graph extraction.
"""
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "weijinglin (via GitHub)" <gi...@apache.org>.
weijinglin closed pull request #301: Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301#discussion_r2447071277
##########
hugegraph-llm/src/hugegraph_llm/flows/scheduler.py:
##########
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+from typing import Dict, Any
+from PyCGraph import GPipelineManager
+from hugegraph_llm.flows.build_vector_index import BuildVectorIndexFlow
+from hugegraph_llm.flows.common import BaseFlow
+from hugegraph_llm.flows.graph_extract import GraphExtractFlow
+from hugegraph_llm.utils.log import log
+
+
+class Scheduler:
+ pipeline_pool: Dict[str, Any] = None
+ max_pipeline: int
Review Comment:
⚠️ **Important: Unused parameter**
The `max_pipeline` parameter is stored in `__init__` but never used anywhere in the class. The `GPipelineManager` is created without any size limits, so this parameter has no effect.
**Recommendation:**
1. Either implement pipeline pool size limits using this parameter
2. Or remove it if not needed yet
3. Document the intended behavior for future implementation
##########
hugegraph-llm/src/hugegraph_llm/flows/scheduler.py:
##########
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+from typing import Dict, Any
+from PyCGraph import GPipelineManager
+from hugegraph_llm.flows.build_vector_index import BuildVectorIndexFlow
+from hugegraph_llm.flows.common import BaseFlow
+from hugegraph_llm.flows.graph_extract import GraphExtractFlow
+from hugegraph_llm.utils.log import log
+
+
+class Scheduler:
+ pipeline_pool: Dict[str, Any] = None
+ max_pipeline: int
+
+ def __init__(self, max_pipeline: int = 10):
+ self.pipeline_pool = {}
+ # pipeline_pool act as a manager of GPipelineManager which used for pipeline management
+ self.pipeline_pool["build_vector_index"] = {
+ "manager": GPipelineManager(),
+ "flow": BuildVectorIndexFlow(),
+ }
+ self.pipeline_pool["graph_extract"] = {
+ "manager": GPipelineManager(),
+ "flow": GraphExtractFlow(),
+ }
+ self.max_pipeline = max_pipeline
+
+ # TODO: Implement Agentic Workflow
+ def agentic_flow(self):
+ pass
+
+ def schedule_flow(self, flow: str, *args, **kwargs):
+ if flow not in self.pipeline_pool:
+ raise ValueError(f"Unsupported workflow {flow}")
+ manager = self.pipeline_pool[flow]["manager"]
+ flow: BaseFlow = self.pipeline_pool[flow]["flow"]
+ pipeline = manager.fetch()
+ if pipeline is None:
+ # call coresponding flow_func to create new workflow
+ pipeline = flow.build_flow(*args, **kwargs)
+ status = pipeline.init()
+ if status.isErr():
+ error_msg = f"Error in flow init: {status.getInfo()}"
+ log.error(error_msg)
+ raise RuntimeError(error_msg)
+ status = pipeline.run()
+ if status.isErr():
+ error_msg = f"Error in flow execution: {status.getInfo()}"
+ log.error(error_msg)
+ raise RuntimeError(error_msg)
+ res = flow.post_deal(pipeline)
Review Comment:
⚠️ **Important: Inconsistent error handling**
The code uses both `raise ValueError` and `raise RuntimeError` inconsistently. Also, the `post_deal` method can fail (e.g., JSON serialization errors) but is not wrapped in error handling.
**Recommendation:**
1. Define a clear exception hierarchy for the flows module
2. Wrap `post_deal` in try-catch to handle serialization errors
3. Document which exceptions callers should expect
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] Refactor: Refactor Scheduler to Support Dynamic Workflow Scheduling and Pipeline Pooling [incubator-hugegraph-ai]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #301:
URL: https://github.com/apache/incubator-hugegraph-ai/pull/301#discussion_r2447068104
##########
hugegraph-llm/pyproject.toml:
##########
@@ -85,3 +86,4 @@ allow-direct-references = true
[tool.uv.sources]
hugegraph-python-client = { workspace = true }
+pycgraph = { git = "https://github.com/ChunelFeng/CGraph.git", subdirectory = "python", rev = "main", marker = "sys_platform == 'linux'" }
Review Comment:
‼️ **Critical: Platform-specific dependency with insufficient error handling**
The `pycgraph` dependency is marked with `marker = "sys_platform == 'linux'"`, meaning it will only install on Linux systems. However, the code imports and uses `PyCGraph` unconditionally without any platform checks or error handling.
**Impact:** The application will crash immediately on non-Linux systems (macOS, Windows) when trying to import modules from `hugegraph_llm.flows`.
**Recommendation:**
1. Add platform compatibility checks and graceful degradation
2. Provide clear error messages for unsupported platforms
3. Consider making CGraph support optional with a feature flag
4. Document platform requirements in README/docs
##########
hugegraph-llm/src/hugegraph_llm/flows/scheduler.py:
##########
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+from typing import Dict, Any
+from PyCGraph import GPipelineManager
+from hugegraph_llm.flows.build_vector_index import BuildVectorIndexFlow
+from hugegraph_llm.flows.common import BaseFlow
+from hugegraph_llm.flows.graph_extract import GraphExtractFlow
+from hugegraph_llm.utils.log import log
+
+
+class Scheduler:
+ pipeline_pool: Dict[str, Any] = None
+ max_pipeline: int
+
+ def __init__(self, max_pipeline: int = 10):
+ self.pipeline_pool = {}
+ # pipeline_pool act as a manager of GPipelineManager which used for pipeline management
+ self.pipeline_pool["build_vector_index"] = {
+ "manager": GPipelineManager(),
+ "flow": BuildVectorIndexFlow(),
+ }
+ self.pipeline_pool["graph_extract"] = {
+ "manager": GPipelineManager(),
+ "flow": GraphExtractFlow(),
+ }
+ self.max_pipeline = max_pipeline
+
+ # TODO: Implement Agentic Workflow
+ def agentic_flow(self):
+ pass
+
+ def schedule_flow(self, flow: str, *args, **kwargs):
Review Comment:
‼️ **Critical: Missing thread safety in pipeline access**
The `schedule_flow` method accesses and modifies the pipeline pool without proper synchronization. While `SchedulerSingleton` uses a lock for instance creation, the `schedule_flow` method itself doesn't protect concurrent access to the shared `pipeline_pool` dictionary.
**Race condition scenarios:**
1. Multiple threads calling `schedule_flow` for the same flow type
2. Concurrent `fetch()` and `release()` operations on the same manager
3. Pipeline state corruption during concurrent runs
**Recommendation:**
Add proper locking around pipeline operations:
```python
def schedule_flow(self, flow: str, *args, **kwargs):
if flow not in self.pipeline_pool:
raise ValueError(f"Unsupported workflow {flow}")
with threading.Lock(): # Add lock here
manager = self.pipeline_pool[flow]["manager"]
# ... rest of the logic
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org