async def _summarize_communities(
        self,
        offset: int,
        limit: int,
        max_summary_input_length: int,
        generation_config: GenerationConfig,
        collection_id: UUID,
        leiden_params: dict,
    ) -> AsyncGenerator[dict, None]:
        """Does the community summary logic from
        GraphCommunitySummaryPipe._run_logic.

        Yields each summary dictionary as it completes.
        """
        start_time = time.time()
        logger.info(
            f"Starting community summarization for collection={collection_id}"
        )

        # get all entities & relationships
        (
            all_entities,
            _,
        ) = await self.providers.database.graphs_handler.get_entities(
            parent_id=collection_id,
            offset=0,
            limit=-1,
            include_embeddings=False,
        )
        (
            all_relationships,
            _,
        ) = await self.providers.database.graphs_handler.get_relationships(
            parent_id=collection_id,
            offset=0,
            limit=-1,
            include_embeddings=False,
        )

        # We can optionally re-run the clustering to produce fresh community assignments
        (
            _,
            community_clusters,
        ) = await self.providers.database.graphs_handler._cluster_and_add_community_info(
            relationships=all_relationships,
            leiden_params=leiden_params,
            collection_id=collection_id,
        )

        # Group clusters
        clusters: dict[Any, list[str]] = {}
        for item in community_clusters:
            cluster_id = item["cluster"]
            node_name = item["node"]
            clusters.setdefault(cluster_id, []).append(node_name)

        # create an async job for each cluster
        tasks: list[Coroutine[Any, Any, dict]] = []

        tasks.extend(
            self._process_community_summary(
                community_id=uuid.uuid4(),
                nodes=nodes,
                all_entities=all_entities,
                all_relationships=all_relationships,
                max_summary_input_length=max_summary_input_length,
                generation_config=generation_config,
                collection_id=collection_id,
            )
            for nodes in clusters.values()
        )

        total_jobs = len(tasks)
        results_returned = 0
        total_errors = 0

        for coro in asyncio.as_completed(tasks):
            summary = await coro
            results_returned += 1
            if results_returned % 50 == 0:
                logger.info(
                    f"Community summaries: {results_returned}/{total_jobs} done in {time.time() - start_time:.2f}s"
                )
            if "error" in summary:
                total_errors += 1
            yield summary

        if total_errors > 0:
            logger.warning(
                f"{total_errors} communities failed summarization out of {total_jobs}"
            )
Does the community summary logic from
        GraphCommunitySummaryPipe._run_logic.

        Yields each summary dictionary as it completes.
        
py/core/main/services/graph_service.py