--- title: "Pipeline" id: pipeline-api description: "Arranges components and integrations in flow." slug: "/pipeline-api" --- ## pipeline ### PipelineStreamHandle Handle returned by `Pipeline.stream()`. Async-iterable over `StreamingChunk`s produced by streaming components in the pipeline. After iteration ends, `result` holds the final pipeline output dict. By default, iteration cleans up automatically: if the consumer abandons iteration, the underlying pipeline task is cancelled. `aclose()` is also available for explicit cleanup. #### result ```python result: dict[str, Any] ``` Final pipeline output dict, available only after a successful, complete run. Raises a `RuntimeError` if the pipeline has not finished or was cancelled. If the pipeline failed, re-raises the original exception. #### aclose ```python aclose() -> None ``` Cancel the underlying pipeline task. Bounded by `_CLEANUP_TIMEOUT_SECONDS` so that components cannot block cleanup indefinitely. ### Pipeline Bases: PipelineBase Orchestration engine that runs components according to the execution graph. Supports both a synchronous run path (`run`) and an asynchronous run path (`run_async`, `run_async_generator`, `stream`). #### run ```python run( data: dict[str, Any], include_outputs_from: set[str] | None = None, *, break_point: Breakpoint | None = None, pipeline_snapshot: PipelineSnapshot | None = None, snapshot_callback: SnapshotCallback | None = None ) -> dict[str, Any] ``` Runs the Pipeline with given input data. `run` executes synchronously and blocks the calling thread until the run completes. In an async context, use `run_async` instead. Usage: ```python from haystack import Pipeline, Document from haystack.components.builders.answer_builder import AnswerBuilder from haystack.components.builders.chat_prompt_builder import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.dataclasses import ChatMessage from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.utils import Secret # Write documents to InMemoryDocumentStore document_store = InMemoryDocumentStore() document_store.write_documents([ Document(content="My name is Jean and I live in Paris."), Document(content="My name is Mark and I live in Berlin."), Document(content="My name is Giorgio and I live in Rome.") ]) retriever = InMemoryBM25Retriever(document_store=document_store) prompt_template = """ Given these documents, answer the question. Documents: {% for doc in documents %} {{ doc.content }} {% endfor %} Question: {{question}} Answer: """ template = [ChatMessage.from_user(prompt_template)] prompt_builder = ChatPromptBuilder( template=template, required_variables=["question", "documents"], variables=["question", "documents"] ) llm = OpenAIChatGenerator() rag_pipeline = Pipeline() rag_pipeline.add_component("retriever", retriever) rag_pipeline.add_component("prompt_builder", prompt_builder) rag_pipeline.add_component("llm", llm) rag_pipeline.connect("retriever", "prompt_builder.documents") rag_pipeline.connect("prompt_builder", "llm") question = "Who lives in Paris?" results = rag_pipeline.run( { "retriever": {"query": question}, "prompt_builder": {"question": question}, } ) print(results["llm"]["replies"][0].text) # Jean lives in Paris ``` **Parameters:** - **data** (dict\[str, Any\]) – A dictionary of inputs for the pipeline's components. Each key is a component name and its value is a dictionary of that component's input parameters: ``` data = { "comp1": {"input1": 1, "input2": 2}, } ``` For convenience, this format is also supported when input names are unique: ``` data = { "input1": 1, "input2": 2, } ``` - **include_outputs_from** (set\[str\] | None) – Set of component names whose individual outputs are to be included in the pipeline's output. For components that are invoked multiple times (in a loop), only the last-produced output is included. - **break_point** (Breakpoint | None) – A breakpoint that pauses execution before the specified component runs by raising a `BreakpointException` carrying a `PipelineSnapshot` of the current pipeline state. - **pipeline_snapshot** (PipelineSnapshot | None) – A snapshot of a previously interrupted pipeline execution to resume from. Can be combined with `break_point` to step through a pipeline: resume from the snapshot and pause again at the next breakpoint. The `break_point` must target a different component or visit count than the one the snapshot was created at, otherwise it would trigger again before any progress is made. - **snapshot_callback** (SnapshotCallback | None) – Optional callback function that is invoked when a pipeline snapshot is created. The callback receives a `PipelineSnapshot` object and can return an optional string (e.g., a file path or identifier). If provided, the callback is used instead of the default file-saving behavior, allowing custom handling of snapshots (e.g., saving to a database, sending to a remote service). If not provided, the default behavior saves snapshots to a JSON file. **Returns:** - dict\[str, Any\] – A dictionary where each entry corresponds to a component name and its output. If `include_outputs_from` is `None`, this dictionary will only contain the outputs of leaf components, i.e., components without outgoing connections. **Raises:** - ValueError – If invalid inputs are provided to the pipeline. - PipelineRuntimeError – If the Pipeline contains cycles with unsupported connections that would cause it to get stuck and fail running. Or if a Component fails or returns output in an unsupported type. - PipelineMaxComponentRuns – If a Component reaches the maximum number of times it can be run in this Pipeline. - PipelineBreakpointException – When a pipeline_breakpoint is triggered. Contains the component name, state, and partial results. #### run_async_generator ```python run_async_generator( data: dict[str, Any], include_outputs_from: set[str] | None = None, concurrency_limit: int = 4, ) -> AsyncGenerator[dict[str, Any], None] ``` Executes the pipeline step by step asynchronously, yielding partial outputs when any component finishes. Usage: ```python from haystack import Document from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack.utils import Secret from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.builders.prompt_builder import PromptBuilder from haystack import Pipeline import asyncio # Write documents to InMemoryDocumentStore document_store = InMemoryDocumentStore() document_store.write_documents([ Document(content="My name is Jean and I live in Paris."), Document(content="My name is Mark and I live in Berlin."), Document(content="My name is Giorgio and I live in Rome.") ]) prompt_template = [ ChatMessage.from_user( ''' Given these documents, answer the question. Documents: {% for doc in documents %} {{ doc.content }} {% endfor %} Question: {{question}} Answer: ''') ] # Create and connect pipeline components retriever = InMemoryBM25Retriever(document_store=document_store) prompt_builder = ChatPromptBuilder(template=prompt_template) llm = OpenAIChatGenerator() rag_pipeline = Pipeline() rag_pipeline.add_component("retriever", retriever) rag_pipeline.add_component("prompt_builder", prompt_builder) rag_pipeline.add_component("llm", llm) rag_pipeline.connect("retriever", "prompt_builder.documents") rag_pipeline.connect("prompt_builder", "llm") # Prepare input data question = "Who lives in Paris?" data = { "retriever": {"query": question}, "prompt_builder": {"question": question}, } # Process results as they become available async def process_results(): async for partial_output in rag_pipeline.run_async_generator( data=data, include_outputs_from={"retriever", "llm"} ): # Each partial_output contains the results from a completed component if "retriever" in partial_output: print("Retrieved documents:", len(partial_output["retriever"]["documents"])) if "llm" in partial_output: print("Generated answer:", partial_output["llm"]["replies"][0]) asyncio.run(process_results()) ``` **Parameters:** - **data** (dict\[str, Any\]) – Initial input data to the pipeline. - **concurrency_limit** (int) – The maximum number of components that are allowed to run concurrently. - **include_outputs_from** (set\[str\] | None) – Set of component names whose individual outputs are to be included in the pipeline's output. For components that are invoked multiple times (in a loop), only the last-produced output is included. **Returns:** - AsyncGenerator\[dict\[str, Any\], None\] – An async iterator containing partial (and final) outputs. **Raises:** - ValueError – If invalid inputs are provided to the pipeline, or if `concurrency_limit` is less than 1. - PipelineMaxComponentRuns – If a component exceeds the maximum number of allowed executions within the pipeline. - PipelineRuntimeError – If the Pipeline contains cycles with unsupported connections that would cause it to get stuck and fail running. Or if a Component fails or returns output in an unsupported type. #### run_async ```python run_async( data: dict[str, Any], include_outputs_from: set[str] | None = None, concurrency_limit: int = 4, ) -> dict[str, Any] ``` Provides an asynchronous interface to run the pipeline with provided input data. This method allows the pipeline to be integrated into an asynchronous workflow, enabling non-blocking execution of pipeline components. Usage: ```python import asyncio from haystack import Document from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack import Pipeline from haystack.dataclasses import ChatMessage from haystack.document_stores.in_memory import InMemoryDocumentStore # Write documents to InMemoryDocumentStore document_store = InMemoryDocumentStore() document_store.write_documents([ Document(content="My name is Jean and I live in Paris."), Document(content="My name is Mark and I live in Berlin."), Document(content="My name is Giorgio and I live in Rome.") ]) prompt_template = [ ChatMessage.from_user( ''' Given these documents, answer the question. Documents: {% for doc in documents %} {{ doc.content }} {% endfor %} Question: {{question}} Answer: ''') ] retriever = InMemoryBM25Retriever(document_store=document_store) prompt_builder = ChatPromptBuilder(template=prompt_template) llm = OpenAIChatGenerator() rag_pipeline = Pipeline() rag_pipeline.add_component("retriever", retriever) rag_pipeline.add_component("prompt_builder", prompt_builder) rag_pipeline.add_component("llm", llm) rag_pipeline.connect("retriever", "prompt_builder.documents") rag_pipeline.connect("prompt_builder", "llm") # Ask a question question = "Who lives in Paris?" async def run_inner(data, include_outputs_from): return await rag_pipeline.run_async(data=data, include_outputs_from=include_outputs_from) data = { "retriever": {"query": question}, "prompt_builder": {"question": question}, } results = asyncio.run(run_inner(data, include_outputs_from={"retriever", "llm"})) print(results["llm"]["replies"]) # [ChatMessage(_role=, _content=[TextContent(text='Jean lives in Paris.')], # _name=None, _meta={'model': 'gpt-5-mini', 'index': 0, 'finish_reason': 'stop', 'usage': # {'completion_tokens': 6, 'prompt_tokens': 69, 'total_tokens': 75, # 'completion_tokens_details': CompletionTokensDetails(accepted_prediction_tokens=0, # audio_tokens=0, reasoning_tokens=0, rejected_prediction_tokens=0), 'prompt_tokens_details': # PromptTokensDetails(audio_tokens=0, cached_tokens=0)}})] ``` **Parameters:** - **data** (dict\[str, Any\]) – A dictionary of inputs for the pipeline's components. Each key is a component name and its value is a dictionary of that component's input parameters: ``` data = { "comp1": {"input1": 1, "input2": 2}, } ``` For convenience, this format is also supported when input names are unique: ``` data = { "input1": 1, "input2": 2, } ``` - **include_outputs_from** (set\[str\] | None) – Set of component names whose individual outputs are to be included in the pipeline's output. For components that are invoked multiple times (in a loop), only the last-produced output is included. - **concurrency_limit** (int) – The maximum number of components that should be allowed to run concurrently. **Returns:** - dict\[str, Any\] – A dictionary where each entry corresponds to a component name and its output. If `include_outputs_from` is `None`, this dictionary will only contain the outputs of leaf components, i.e., components without outgoing connections. **Raises:** - ValueError – If invalid inputs are provided to the pipeline, or if `concurrency_limit` is less than 1. - PipelineRuntimeError – If the Pipeline contains cycles with unsupported connections that would cause it to get stuck and fail running. Or if a Component fails or returns output in an unsupported type. - PipelineMaxComponentRuns – If a Component reaches the maximum number of times it can be run in this Pipeline. #### stream ```python stream( data: dict[str, Any], *, streaming_components: list[str] | None = None, include_outputs_from: set[str] | None = None, concurrency_limit: int = 4, cancel_on_abandon: bool = True ) -> PipelineStreamHandle ``` Run the pipeline and return a handle that streams `StreamingChunk`s as they arrive. Iterate the handle with `async for` to consume chunks; after iteration ends, `handle.result` holds the final pipeline output dict (same as `run_async`). By default, if iteration is abandoned, the underlying pipeline task is cancelled automatically. Pass `cancel_on_abandon=False` to instead let the pipeline run to completion. For every async-capable component that exposes a `streaming_callback` input socket, a forwarder is injected at runtime that pushes chunks onto the handle's queue. If a `streaming_callback` is provided at component init or at runtime (inside `data`, e.g. `data={"llm": {"streaming_callback": cb}}`), it is also invoked for each chunk. Async callbacks are preferred; a sync callback is accepted but will run synchronously on the event loop and may block it. Usage: ```python import asyncio from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack import Pipeline from haystack.dataclasses import ChatMessage pipe = Pipeline() pipe.add_component( "prompt_builder", ChatPromptBuilder(template=[ChatMessage.from_user("Tell me about {{topic}}")]), ) pipe.add_component("llm", OpenAIChatGenerator()) pipe.connect("prompt_builder.prompt", "llm.messages") async def main(): handle = pipe.stream(data={"prompt_builder": {"topic": "Italy"}}) async for chunk in handle: print(chunk.content, end="", flush=True) return handle.result result = asyncio.run(main()) print(result["llm"]["replies"]) ``` **Parameters:** - **data** (dict\[str, Any\]) – A dictionary of inputs for the pipeline's components. Each key is a component name and its value is a dictionary of that component's input parameters: ``` data = { "comp1": {"input1": 1, "input2": 2}, } ``` For convenience, this format is also supported when input names are unique: ``` data = { "input1": 1, "input2": 2, } ``` - **streaming_components** (list\[str\] | None) – Names of components to stream from. If `None` (default), every streaming-capable component is forwarded. If a list, only the listed components are forwarded; unknown names or names of components that do not support streaming raise `ValueError`. - **include_outputs_from** (set\[str\] | None) – Set of component names whose individual outputs are to be included in the pipeline's output. For components that are invoked multiple times (in a loop), only the last-produced output is included. - **concurrency_limit** (int) – The maximum number of components that should be allowed to run concurrently. - **cancel_on_abandon** (bool) – If `True` (default), the underlying pipeline task is cancelled when iteration is abandoned. If `False`, the pipeline runs to completion even when the consumer stops reading. **Returns:** - PipelineStreamHandle – A `PipelineStreamHandle` that is async-iterable over `StreamingChunk`s. After iteration ends, `handle.result` holds the final pipeline output dict (same shape as `run_async`). **Raises:** - ValueError – If `streaming_components` contains unknown component names or components that do not support streaming, or if invalid inputs are provided to the pipeline, or if `concurrency_limit` is less than 1. - PipelineRuntimeError – Surfaced during iteration. If the Pipeline contains cycles with unsupported connections that would cause it to get stuck and fail running, or if a Component fails or returns output in an unsupported type. - PipelineMaxComponentRuns – Surfaced during iteration. If a Component reaches the maximum number of times it can be run in this Pipeline.