Streamline run_pipeline_with_config call pattern

python/graphrag/examples/custom_input/run.py

@@ -20,7 +20,7 @@ async def run():
 
     # Grab the last result from the pipeline, should be our entity extraction
     outputs = []
-    async for output in await run_pipeline_with_config(
+    async for output in run_pipeline_with_config(
         config_or_path=config, dataset=dataset
     ):
         outputs.append(output)

python/graphrag/examples/custom_set_of_available_verbs/run.py

@@ -20,7 +20,7 @@ async def run_with_config():
     )
 
     outputs = []
-    async for output in await run_pipeline_with_config(
+    async for output in run_pipeline_with_config(
         config_or_path=config_path, dataset=dataset
     ):
         outputs.append(output)

python/graphrag/examples/custom_set_of_available_workflows/run.py

@@ -27,7 +27,7 @@ async def run_with_config():
 
     # Grab the last result from the pipeline, should be our entity extraction
     tables = []
-    async for table in await run_pipeline_with_config(
+    async for table in run_pipeline_with_config(
         config_or_path=config_path,
         dataset=dataset,
         additional_workflows=custom_workflows,

python/graphrag/examples/entity_extraction/with_graph_intelligence/run.py

@@ -37,7 +37,7 @@ async def run_with_config():
 
     # Grab the last result from the pipeline, should be our entity extraction
     tables = []
-    async for table in await run_pipeline_with_config(
+    async for table in run_pipeline_with_config(
         config_or_path=config_path, dataset=dataset
     ):
         tables.append(table)

python/graphrag/examples/entity_extraction/with_nltk/run.py

@@ -36,7 +36,7 @@ async def run_with_config():
 
     # Grab the last result from the pipeline, should be our entity extraction
     tables = []
-    async for table in await run_pipeline_with_config(
+    async for table in run_pipeline_with_config(
         config_or_path=config_path, dataset=dataset
     ):
         tables.append(table)

python/graphrag/examples/interdependent_workflows/run.py

@@ -24,7 +24,7 @@ async def run_with_config():
     )
 
     tables = []
-    async for table in await run_pipeline_with_config(
+    async for table in run_pipeline_with_config(
         config_or_path=config_path, dataset=dataset
     ):
         tables.append(table)

python/graphrag/examples/multiple_workflows/run.py

@@ -33,7 +33,7 @@ async def run_with_config():
         os.path.dirname(os.path.abspath(__file__)), "./pipeline.yml"
     )
 
-    async for result in await run_pipeline_with_config(pipeline_path, dataset=dataset):
+    async for result in run_pipeline_with_config(pipeline_path, dataset=dataset):
         print(f"Workflow {result.workflow} result\n: ")
         print(result.result)
 

python/graphrag/examples/single_verb/run.py

@@ -19,7 +19,7 @@ async def run_with_config():
     )
 
     tables = []
-    async for table in await run_pipeline_with_config(
+    async for table in run_pipeline_with_config(
         config_or_path=config_path, dataset=dataset
     ):
         tables.append(table)

python/graphrag/examples/use_built_in_workflows/run.py

@@ -38,7 +38,7 @@ async def run_with_config():
 
     # Grab the last result from the pipeline, should be our entity extraction
     tables = []
-    async for table in await run_pipeline_with_config(
+    async for table in run_pipeline_with_config(
         config_or_path=config_path, dataset=dataset
     ):
         tables.append(table)

python/graphrag/examples/various_levels_of_configs/workflows_and_inputs.py

@@ -23,7 +23,7 @@ async def main():
     # run the pipeline with the config, and override the dataset with the one we just created
     # and grab the last result from the pipeline, should be the last workflow that was run (our nodes)
     tables = []
-    async for table in await run_pipeline_with_config(pipeline_path):
+    async for table in run_pipeline_with_config(pipeline_path):
         tables.append(table)
     pipeline_result = tables[-1]
 

python/graphrag/examples/various_levels_of_configs/workflows_and_inputs_with_custom_handlers.py

@@ -37,7 +37,7 @@ async def main():
     # run the pipeline with the config, and override the dataset with the one we just created
     # and grab the last result from the pipeline, should be the last workflow that was run (our nodes)
     pipeline_result = []
-    async for result in await run_pipeline_with_config(
+    async for result in run_pipeline_with_config(
         pipeline_path,
         storage=custom_storage,
         callbacks=custom_reporter,

python/graphrag/examples/various_levels_of_configs/workflows_only.py

@@ -40,7 +40,7 @@ async def main():
         os.path.dirname(os.path.abspath(__file__)), "./pipelines/workflows_only.yml"
     )
     tables = []
-    async for table in await run_pipeline_with_config(pipeline_path, dataset=dataset):
+    async for table in run_pipeline_with_config(pipeline_path, dataset=dataset):
         tables.append(table)
     pipeline_result = tables[-1]
 

python/graphrag/graphrag/index/cli.py

@@ -83,7 +83,7 @@ def handle_signal(signum, _):
 
         async def execute():
             nonlocal encountered_errors
-            async for output in await run_pipeline_with_config(
+            async for output in run_pipeline_with_config(
                 pipeline_config,
                 debug=verbose,
                 resume=resume,  # type: ignore

python/graphrag/graphrag/index/run.py

@@ -146,7 +146,7 @@ def _create_postprocess_steps(
         msg = "No dataset provided!"
         raise ValueError(msg)
 
-    return run_pipeline(
+    async for table in run_pipeline(
         workflows=workflows,
         dataset=dataset,
         storage=storage,
@@ -158,7 +158,8 @@ def _create_postprocess_steps(
         additional_workflows=additional_workflows,
         progress_reporter=progress_reporter,
         emit=emit,
-    )
+    ):
+        yield table
 
 
 async def run_pipeline(

python/graphrag/tests/integration/_pipeline/test_run.py

@@ -15,9 +15,7 @@ async def test_megapipeline(self):
             os.path.dirname(os.path.abspath(__file__)),
             "./megapipeline.yml",
         )
-        pipeline_result = [
-            gen async for gen in await run_pipeline_with_config(pipeline_path)
-        ]
+        pipeline_result = [gen async for gen in run_pipeline_with_config(pipeline_path)]
 
         errors = []
         for result in pipeline_result: