imbajin commented on code in PR #370:
URL: https://github.com/apache/hugegraph-ai/pull/370#discussion_r3564247455
##########
hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract.py:
##########
@@ -90,26 +98,86 @@ def run(self, context: Dict[str, Any]) -> Dict[str,
List[Any]]:
context["vertices"] = []
if "edges" not in context:
context["edges"] = []
+
items = []
- for chunk in chunks:
- proceeded_chunk = self.extract_property_graph_by_llm(schema, chunk)
- log.debug(
- "[LLM] %s input: %s \n output:%s",
- self.__class__.__name__,
- chunk,
- proceeded_chunk,
- )
- items.extend(self._extract_and_filter_label(schema,
proceeded_chunk))
+ if self.max_workers == 1 or len(chunks) <= 1:
+ chunk_results = [
+ self._extract_chunk_items(schema, chunk, index, len(chunks))
for index, chunk in enumerate(chunks)
+ ]
+ else:
+ chunk_results = self._extract_chunks_concurrently(schema, chunks)
+
+ for chunk_items in chunk_results:
+ items.extend(chunk_items)
+
items = filter_item(schema, items)
for item in items:
if item["type"] == "vertex":
context["vertices"].append(item)
elif item["type"] == "edge":
context["edges"].append(item)
-
context["call_count"] = context.get("call_count", 0) + len(chunks)
return context
+ def _extract_chunks_concurrently(self, schema, chunks):
+ worker_count = min(self.max_workers, len(chunks))
+ chunk_results = [None] * len(chunks)
+ with ThreadPoolExecutor(max_workers=worker_count) as executor:
+ future_to_index = {
+ executor.submit(
+ self._extract_chunk_items,
+ schema,
+ chunk,
+ index,
+ len(chunks),
+ ): index
+ for index, chunk in enumerate(chunks)
+ }
+ for future in as_completed(future_to_index):
+ index = future_to_index[future]
+ chunk_results[index] = future.result()
Review Comment:
⚠️ A failed chunk does not stop queued LLM calls. Every chunk is submitted
above, and when `future.result()` raises, leaving the `ThreadPoolExecutor`
context waits for running work while queued calls may still start, delaying the
error and consuming unnecessary rate limit or paid requests. Please cancel
futures that have not started on the first failure, define the policy for
already-running calls, and add a regression test proving later queued chunks
are not invoked after an early failure.
##########
hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract.py:
##########
@@ -90,26 +98,86 @@ def run(self, context: Dict[str, Any]) -> Dict[str,
List[Any]]:
context["vertices"] = []
if "edges" not in context:
context["edges"] = []
+
items = []
- for chunk in chunks:
- proceeded_chunk = self.extract_property_graph_by_llm(schema, chunk)
- log.debug(
- "[LLM] %s input: %s \n output:%s",
- self.__class__.__name__,
- chunk,
- proceeded_chunk,
- )
- items.extend(self._extract_and_filter_label(schema,
proceeded_chunk))
+ if self.max_workers == 1 or len(chunks) <= 1:
+ chunk_results = [
+ self._extract_chunk_items(schema, chunk, index, len(chunks))
for index, chunk in enumerate(chunks)
+ ]
+ else:
+ chunk_results = self._extract_chunks_concurrently(schema, chunks)
+
+ for chunk_items in chunk_results:
+ items.extend(chunk_items)
+
items = filter_item(schema, items)
for item in items:
if item["type"] == "vertex":
context["vertices"].append(item)
elif item["type"] == "edge":
context["edges"].append(item)
-
context["call_count"] = context.get("call_count", 0) + len(chunks)
return context
+ def _extract_chunks_concurrently(self, schema, chunks):
+ worker_count = min(self.max_workers, len(chunks))
+ chunk_results = [None] * len(chunks)
+ with ThreadPoolExecutor(max_workers=worker_count) as executor:
+ future_to_index = {
+ executor.submit(
+ self._extract_chunk_items,
+ schema,
+ chunk,
+ index,
+ len(chunks),
+ ): index
+ for index, chunk in enumerate(chunks)
+ }
+ for future in as_completed(future_to_index):
+ index = future_to_index[future]
+ chunk_results[index] = future.result()
+ return chunk_results
+
+ def _extract_chunk_items(self, schema, chunk, chunk_index, chunk_count):
+ try:
+ proceeded_chunk = self.extract_property_graph_by_llm(schema, chunk)
+ self._extract_property_graph_json(proceeded_chunk)
+ self._extract_property_graph_json(proceeded_chunk)
Review Comment:
🧹 The same response is parsed and validated twice consecutively, and both
return values are discarded. This doubles the regex and JSON-decoding work for
every chunk without adding coverage. Please remove the duplicate call or parse
once and reuse the result.
--
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]
