nw9663644-eng commented on code in PR #370:
URL: https://github.com/apache/hugegraph-ai/pull/370#discussion_r3563598142
##########
hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract.py:
##########
@@ -90,26 +97,60 @@ 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)
+ except Exception as exc:
+ raise RuntimeError(f"Graph extraction failed for chunk
{chunk_index + 1}/{chunk_count}: {exc}") from exc
+
+ log.debug(
+ "[LLM] %s input: %s \noutput:%s",
+ self.__class__.__name__,
+ chunk,
+ proceeded_chunk,
+ )
+ return self._extract_and_filter_label(schema, proceeded_chunk)
Review Comment:
Thanks for the review. I updated the chunk extraction path so malformed LLM
responses are no longer silently treated as empty results.
Each chunk response is now validated for JSON extraction and expected
property-graph shape before filtering. If a chunk returns malformed JSON or an
invalid graph format, extraction fails with chunk context such as `Graph
extraction failed for chunk 2/5: ...`.
I also added regression tests for malformed chunk output, backend
concurrency cap validation, and REST request validation.
--
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]
