From 627cbeeb2096eb4cd5da45015d37fceb7fe7840a Mon Sep 17 00:00:00 2001 From: Federico Minutoli Date: Sat, 11 May 2024 00:13:27 +0200 Subject: [PATCH 1/4] feat(parallel-exeuction): add asyncio event loop dispatcher with semaphore for parallel graph instances TODO: still untested --- scrapegraphai/nodes/graph_iterator_node.py | 99 +++++++++++++++++----- 1 file changed, 77 insertions(+), 22 deletions(-) diff --git a/scrapegraphai/nodes/graph_iterator_node.py b/scrapegraphai/nodes/graph_iterator_node.py index 663adc62..4ed7ad8e 100644 --- a/scrapegraphai/nodes/graph_iterator_node.py +++ b/scrapegraphai/nodes/graph_iterator_node.py @@ -2,12 +2,18 @@ GraphIterator Module """ -from typing import List, Optional +import asyncio import copy -from tqdm import tqdm +from typing import List, Optional + +from tqdm.asyncio import tqdm + from .base_node import BaseNode +_default_batchsize = 4 + + class GraphIteratorNode(BaseNode): """ A node responsible for instantiating and running multiple graph instances in parallel. @@ -23,12 +29,20 @@ class GraphIteratorNode(BaseNode): node_name (str): The unique identifier name for the node, defaulting to "Parse". """ - def __init__(self, input: str, output: List[str], node_config: Optional[dict]=None, node_name: str = "GraphIterator"): + def __init__( + self, + input: str, + output: List[str], + node_config: Optional[dict] = None, + node_name: str = "GraphIterator", + ): super().__init__(node_name, "node", input, output, 2, node_config) - self.verbose = False if node_config is None else node_config.get("verbose", False) + self.verbose = ( + False if node_config is None else node_config.get("verbose", False) + ) - def execute(self, state: dict) -> dict: + def execute(self, state: dict) -> dict: """ Executes the node's logic to instantiate and run multiple graph instances in parallel. @@ -43,37 +57,78 @@ class GraphIteratorNode(BaseNode): KeyError: If the input keys are not found in the state, indicating that the necessary information for running the graph instances is missing. """ + batchsize = self.node_config.get("batchsize", _default_batchsize) if self.verbose: - print(f"--- Executing {self.node_name} Node ---") + print(f"--- Executing {self.node_name} Node with batchsize {batchsize} ---") - # Interpret input keys based on the provided input expression + try: + eventloop = asyncio.get_event_loop() + except RuntimeError: + eventloop = None + + if eventloop and eventloop.is_running(): + state = eventloop.run_until_complete(self._async_execute(state, batchsize)) + else: + state = asyncio.run(self._async_execute(state, batchsize)) + + return state + + async def _async_execute(self, state: dict, batchsize: int) -> dict: + """asynchronously executes the node's logic with multiple graph instances + running in parallel, using a semaphore of some size for concurrency regulation + + Args: + state: The current state of the graph. + batchsize: The maximum number of concurrent instances allowed. + + Returns: + The updated state with the output key containing the results + aggregated out of all parallel graph instances. + + Raises: + KeyError: If the input keys are not found in the state. + """ + + # interprets input keys based on the provided input expression input_keys = self.get_input_keys(state) - # Fetching data from the state based on the input keys + # fetches data from the state based on the input keys input_data = [state[key] for key in input_keys] user_prompt = input_data[0] urls = input_data[1] graph_instance = self.node_config.get("graph_instance", None) + if graph_instance is None: - raise ValueError("Graph instance is required for graph iteration.") - - # set the prompt and source for each url + raise ValueError("graph instance is required for concurrent execution") + + # sets the prompt for the graph instance graph_instance.prompt = user_prompt - graphs_instances = [] + + participants = [] + + # semaphore to limit the number of concurrent tasks + semaphore = asyncio.Semaphore(batchsize) + + async def _async_run(graph): + async with semaphore: + return await asyncio.to_thread(graph.run) + + # creates a deepcopy of the graph instance for each endpoint for url in urls: - # make a copy of the graph instance - copy_graph_instance = copy.copy(graph_instance) - copy_graph_instance.source = url - graphs_instances.append(copy_graph_instance) + instance = copy.deepcopy(graph_instance) + instance.source = url - # run the graph for each url and use tqdm for progress bar - graphs_answers = [] - for graph in tqdm(graphs_instances, desc="Processing Graph Instances", disable=not self.verbose): - result = graph.run() - graphs_answers.append(result) + participants.append(instance) + + futures = [_async_run(graph) for graph in participants] + + answers = await tqdm.gather( + *futures, desc="processing graph instances", disable=not self.verbose + ) + + state.update({self.output[0]: answers}) - state.update({self.output[0]: graphs_answers}) return state From dedc73304755c2d540a121d143173f60fb448bbb Mon Sep 17 00:00:00 2001 From: Marco Perini Date: Mon, 13 May 2024 18:46:34 +0200 Subject: [PATCH 2/4] fix(asyncio): replaced deepcopy with copy due to serialization problems --- examples/openai/search_graph_openai.py | 2 +- scrapegraphai/nodes/graph_iterator_node.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/examples/openai/search_graph_openai.py b/examples/openai/search_graph_openai.py index ffff1cb5..7f40ebde 100644 --- a/examples/openai/search_graph_openai.py +++ b/examples/openai/search_graph_openai.py @@ -28,7 +28,7 @@ graph_config = { # ************************************************ search_graph = SearchGraph( - prompt="List me the best escursions near Trento", + prompt="List me Chioggia's famous dishes", config=graph_config ) diff --git a/scrapegraphai/nodes/graph_iterator_node.py b/scrapegraphai/nodes/graph_iterator_node.py index 4ed7ad8e..517dfe0c 100644 --- a/scrapegraphai/nodes/graph_iterator_node.py +++ b/scrapegraphai/nodes/graph_iterator_node.py @@ -11,7 +11,7 @@ from tqdm.asyncio import tqdm from .base_node import BaseNode -_default_batchsize = 4 +_default_batchsize = 2 class GraphIteratorNode(BaseNode): @@ -118,7 +118,7 @@ class GraphIteratorNode(BaseNode): # creates a deepcopy of the graph instance for each endpoint for url in urls: - instance = copy.deepcopy(graph_instance) + instance = copy.copy(graph_instance) instance.source = url participants.append(instance) From a8d5e7db050e15306780ffca47f998ebaf5c1216 Mon Sep 17 00:00:00 2001 From: Marco Perini Date: Mon, 13 May 2024 23:49:48 +0200 Subject: [PATCH 3/4] feat(batchsize): tested different batch sizes and systems --- scrapegraphai/nodes/graph_iterator_node.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scrapegraphai/nodes/graph_iterator_node.py b/scrapegraphai/nodes/graph_iterator_node.py index 517dfe0c..8a71319a 100644 --- a/scrapegraphai/nodes/graph_iterator_node.py +++ b/scrapegraphai/nodes/graph_iterator_node.py @@ -11,7 +11,7 @@ from tqdm.asyncio import tqdm from .base_node import BaseNode -_default_batchsize = 2 +_default_batchsize = 16 class GraphIteratorNode(BaseNode): From fa4edb47033121b81cdcc1c910f0386cba5a2f2e Mon Sep 17 00:00:00 2001 From: semantic-release-bot Date: Mon, 13 May 2024 21:52:03 +0000 Subject: [PATCH 4/4] ci(release): 0.11.0-beta.8 [skip ci] ## [0.11.0-beta.8](https://github.com/VinciGit00/Scrapegraph-ai/compare/v0.11.0-beta.7...v0.11.0-beta.8) (2024-05-13) ### Features * **parallel-exeuction:** add asyncio event loop dispatcher with semaphore for parallel graph instances ([627cbee](https://github.com/VinciGit00/Scrapegraph-ai/commit/627cbeeb2096eb4cd5da45015d37fceb7fe7840a)) * **batchsize:** tested different batch sizes and systems ([a8d5e7d](https://github.com/VinciGit00/Scrapegraph-ai/commit/a8d5e7db050e15306780ffca47f998ebaf5c1216)) ### Bug Fixes * **asyncio:** replaced deepcopy with copy due to serialization problems ([dedc733](https://github.com/VinciGit00/Scrapegraph-ai/commit/dedc73304755c2d540a121d143173f60fb448bbb)) --- CHANGELOG.md | 13 +++++++++++++ pyproject.toml | 2 +- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4afb2d4a..60c8e01c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,16 @@ +## [0.11.0-beta.8](https://github.com/VinciGit00/Scrapegraph-ai/compare/v0.11.0-beta.7...v0.11.0-beta.8) (2024-05-13) + + +### Features + +* **parallel-exeuction:** add asyncio event loop dispatcher with semaphore for parallel graph instances ([627cbee](https://github.com/VinciGit00/Scrapegraph-ai/commit/627cbeeb2096eb4cd5da45015d37fceb7fe7840a)) +* **batchsize:** tested different batch sizes and systems ([a8d5e7d](https://github.com/VinciGit00/Scrapegraph-ai/commit/a8d5e7db050e15306780ffca47f998ebaf5c1216)) + + +### Bug Fixes + +* **asyncio:** replaced deepcopy with copy due to serialization problems ([dedc733](https://github.com/VinciGit00/Scrapegraph-ai/commit/dedc73304755c2d540a121d143173f60fb448bbb)) + ## [0.11.0-beta.7](https://github.com/VinciGit00/Scrapegraph-ai/compare/v0.11.0-beta.6...v0.11.0-beta.7) (2024-05-13) diff --git a/pyproject.toml b/pyproject.toml index 07b14714..c41c3112 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [tool.poetry] name = "scrapegraphai" -version = "0.11.0b7" +version = "0.11.0b8" description = "A web scraping library based on LangChain which uses LLM and direct graph logic to create scraping pipelines." authors = [