def run()

in hugegraph-llm/src/hugegraph_llm/operators/llm_op/property_graph_extract.py [0:0]

• 20 lines of code
• 7 McCabe index (conditional complexity)

    def run(self, context: Dict[str, Any]) -> Dict[str, List[Any]]:
        schema = context["schema"]
        chunks = context["chunks"]
        if "vertices" not in context:
            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))
        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