imbajin commented on code in PR #370:
URL: https://github.com/apache/hugegraph-ai/pull/370#discussion_r3562079166
##########
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:
This still treats malformed model output as a successful empty chunk in
several cases. _extract_chunk_items only wraps exceptions raised by
extract_property_graph_by_llm, but _extract_and_filter_label returns [] when no
JSON is found, when the graph shape is invalid, or when JSON decoding fails.
With multiple chunks, one bad chunk can therefore be silently dropped and the
overall extraction still returns success, which loses the chunk-level failure
visibility described in the PR. Please make parse/format failures
distinguishable from a valid empty extraction and raise or report them with the
chunk index.
--
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]
