imbajin commented on code in PR #370:
URL: https://github.com/apache/hugegraph-ai/pull/370#discussion_r3577836774
##########
hugegraph-llm/src/hugegraph_llm/api/models/graph_extract_requests.py:
##########
@@ -44,6 +46,12 @@ class GraphExtractRequest(BaseModel):
extract_type: Literal["property_graph"] = Query("property_graph",
description="Extraction type.")
language: Literal["zh", "en"] = Query("zh", description="Language for
chunk splitting.")
split_type: Literal["document", "paragraph", "sentence"] =
Query("document", description="Chunk split granularity.")
+ graph_extract_max_workers: int = Query(
Review Comment:
🧹 REST validation coerces some raw inputs before the shared validator can
enforce its contract. A direct probe shows `graph_extract_max_workers=true` and
`"1.0"` are both accepted here as integer `1`, while
`validate_graph_extract_max_workers()` rejects those same raw values. This
makes REST behavior inconsistent with the flow/helper entry points and can
silently turn malformed input into a valid request. Please add a
`mode="before"` field validator (or equivalent strict raw-input validation)
that applies the shared rules, and cover boolean and decimal-string request
values with regression tests.
##########
hugegraph-llm/src/tests/document/test_graph_extract_concurrency.py:
##########
@@ -0,0 +1,371 @@
+# 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 json
+import re
+import threading
+import time
+
+import gradio as gr
+import pytest
+from pydantic import ValidationError
+
+from hugegraph_llm.flows.graph_extract import GraphExtractFlow
+from hugegraph_llm.operators.llm_op.property_graph_extract import
PropertyGraphExtract
+from hugegraph_llm.state.ai_state import WkFlowInput
+from hugegraph_llm.utils import graph_index_utils
+from hugegraph_llm.utils.graph_extract_config import
validate_graph_extract_max_workers
+
+SCHEMA = {
+ "vertexlabels": [
+ {
+ "id": 1,
+ "name": "person",
+ "id_strategy": "PRIMARY_KEY",
+ "primary_keys": ["name"],
+ "nullable_keys": [],
+ "properties": ["name"],
+ }
+ ],
+ "edgelabels": [],
+}
+
+
+class CountingLLM:
+ def __init__(self, delay=0.02, fail_on=None, malformed_on=None):
+ self.delay = delay
+ self.fail_on = fail_on
+ self.malformed_on = malformed_on
+ self.active = 0
+ self.max_active = 0
+ self.lock = threading.Lock()
+ self.calls = []
+
+ def generate(self, prompt):
+ chunk = self._chunk_from_prompt(prompt)
+ with self.lock:
+ self.active += 1
+ self.max_active = max(self.max_active, self.active)
+ try:
+ self.calls.append(chunk)
+ if chunk == self.fail_on:
+ raise RuntimeError("boom")
+ if chunk == self.malformed_on:
+ return "this is not json"
+ time.sleep(self.delay)
+ return json.dumps(
+ {
+ "vertices": [
+ {
+ "label": "person",
+ "type": "vertex",
+ "properties": {"name": chunk},
+ }
+ ],
+ "edges": [],
+ }
+ )
+ finally:
+ with self.lock:
+ self.active -= 1
+
+ @staticmethod
+ def _chunk_from_prompt(prompt):
+ match = re.search(r"## Text:\s*(.*?)\s*## Graph schema", prompt,
re.DOTALL)
+ if not match:
+ raise AssertionError(f"Could not identify chunk in prompt:
{prompt}")
+ return match.group(1).strip()
+
+
+def test_property_graph_extract_respects_configured_concurrency_limit():
+ llm = CountingLLM()
+ extractor = PropertyGraphExtract(llm, example_prompt="", max_workers=2)
+
+ result = extractor.run({"schema": SCHEMA, "chunks": ["a", "b", "c", "d"]})
+
+ assert llm.max_active <= 2
+ assert llm.max_active > 1
+ assert result["call_count"] == 4
+
+
+def test_property_graph_extract_serial_mode_keeps_one_active_call():
+ llm = CountingLLM()
+ extractor = PropertyGraphExtract(llm, example_prompt="", max_workers=1)
+
+ result = extractor.run({"schema": SCHEMA, "chunks": ["a", "b", "c"]})
+
+ assert llm.max_active == 1
+ assert result["call_count"] == 3
+
+
+def test_property_graph_extract_preserves_chunk_merge_order_with_concurrency():
+ llm = CountingLLM()
Review Comment:
🧹 This order-preservation test gives every chunk the same delay, so it does
not deterministically make completion order differ from input order. An
implementation that accidentally merges futures in completion order could still
pass when these calls finish first/second/third. Please coordinate the fake
calls with distinct events or delays so they finish in a fixed reverse order,
then assert that the merged result remains in input order.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
