diff --git a/docs-website/docs/concepts/pipelines.mdx b/docs-website/docs/concepts/pipelines.mdx index f3311f8b19f..25ccfd8286a 100644 --- a/docs-website/docs/concepts/pipelines.mdx +++ b/docs-website/docs/concepts/pipelines.mdx @@ -42,11 +42,52 @@ See [Pipeline Loops](pipelines/pipeline-loops.mdx) for a deeper explanation of h -### Async Pipelines +### Async Execution and Streaming -The AsyncPipeline enables parallel execution of Haystack components when their dependencies allow it. This improves performance in complex pipelines with independent operations. For example, it can run multiple Retrievers or LLM calls simultaneously, execute independent pipeline branches in parallel, and efficiently handle I/O-bound operations that would otherwise cause delays. Through concurrent execution, the AsyncPipeline significantly reduces total processing time compared to sequential execution. +Pipelines execute components in parallel when their dependencies allow it. This improves performance in complex pipelines with independent operations. For example, a pipeline can run multiple Retrievers or LLM calls simultaneously, execute independent pipeline branches in parallel, and efficiently handle I/O-bound operations that would otherwise cause delays. You can cap the number of components running at the same time with the `concurrency_limit` init parameter. -Find out more in our [AsyncPipeline](pipelines/asyncpipeline.mdx) documentation. +Besides the blocking `run` method, every pipeline offers three ways to run asynchronously: + +- `run_async`: Executes the pipeline in a single non-blocking call, ideal for integrating a pipeline into a larger async application or service. +- `run_async_generator`: Yields partial outputs as components complete their tasks, which is useful for monitoring progress, debugging, and handling outputs incrementally. +- `stream`: Runs the pipeline and returns a handle that streams [`StreamingChunk`](/reference/data-classes-api#streamingchunk) objects as they are produced — a convenient way to stream LLM output from an async application, such as an API endpoint. Iterate the handle with `async for` to consume the chunks; after iteration ends, `handle.result` holds the final pipeline output (the same dictionary returned by `run_async`). + +```python +import asyncio + +from haystack import Pipeline +from haystack.components.builders import ChatPromptBuilder +from haystack.components.generators.chat import OpenAIChatGenerator +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()) +``` + +By default, chunks from every streaming-capable component are forwarded; pass `streaming_components` with a list of component names to stream only specific components. If the consumer abandons iteration, the underlying pipeline run is cancelled automatically; pass `cancel_on_abandon=False` to let it run to completion instead. + +If a `streaming_callback` is set on a component (at init or at runtime through `data`), it is still invoked for each chunk in addition to the chunks being pushed to the handle. When streaming, components accept a sync `streaming_callback` in `run_async` too — see the [Choosing the Right Generator guide](../pipeline-components/generators/guides-to-generators/choosing-the-right-generator.mdx#sync-and-async-callbacks) for details. + +#### Error Handling and Task Cancellation + +If a component raises an error while sibling components are still running concurrently, the pipeline cancels and drains those in-flight tasks before re-raising the original error, so no tasks keep running in the background. The same cleanup applies when you stop iterating `run_async_generator` early (for example, by breaking out of the loop and closing the generator) or when the run itself is cancelled. + +Note that cancellation only interrupts components that run natively async. Sync components are offloaded to a worker thread, which cannot be interrupted and runs to completion in the background. Their outputs are discarded, so the pipeline state stays consistent, but the component's side effects still complete. ## SuperComponents diff --git a/docs-website/docs/concepts/pipelines/asyncpipeline.mdx b/docs-website/docs/concepts/pipelines/asyncpipeline.mdx deleted file mode 100644 index e113b331bac..00000000000 --- a/docs-website/docs/concepts/pipelines/asyncpipeline.mdx +++ /dev/null @@ -1,163 +0,0 @@ ---- -title: "AsyncPipeline" -id: asyncpipeline -slug: "/asyncpipeline" -description: "Use AsyncPipeline to run multiple Haystack components at the same time for faster processing." ---- - -# AsyncPipeline - -Use AsyncPipeline to run multiple Haystack components at the same time for faster processing. - -The `AsyncPipeline` in Haystack introduces asynchronous execution capabilities, enabling concurrent component execution when dependencies allow. This optimizes performance, particularly in complex pipelines where multiple independent components can run in parallel. - -The `AsyncPipeline` provides significant performance improvements in scenarios such as: - -- Hybrid retrieval pipelines, where multiple Retrievers can run in parallel, -- Multiple LLM calls that can be executed concurrently, -- Complex pipelines with independent branches of execution, -- I/O-bound operations that benefit from asynchronous execution. - -## Key Features - -### Concurrent Execution - -The `AsyncPipeline` schedules components based on input readiness and dependency resolution, ensuring efficient parallel execution when possible. For example, in a hybrid retrieval scenario, multiple Retrievers can run simultaneously if they do not depend on each other. - -### Execution Methods - -The `AsyncPipeline` offers three ways to run your pipeline: - -#### Synchronous Run (`run`) - -Executes the pipeline synchronously with the provided input data. This method is blocking, making it suitable for environments where asynchronous execution is not possible or desired. Although components execute concurrently internally, the method blocks until the pipeline completes. - -#### Asynchronous Run (`run_async`) - -Executes the pipeline in an asynchronous manner, allowing non-blocking execution. This method is ideal when integrating the pipeline into an async workflow, enabling smooth operation within larger async applications or services. - -#### Asynchronous Generator (`run_async_generator`) - -Allows step-by-step execution by yielding partial outputs as components complete their tasks. This is particularly useful for monitoring progress, debugging, and handling outputs incrementally. It differs from `run_async`, which executes the pipeline in a single async call. - -In an `AsyncPipeline`, components such as A and B will run in parallel _only if they have no shared dependencies_ and can process inputs independently. - -### Concurrency Control - -You can control the maximum number of components that run simultaneously using the `concurrency_limit` parameter to ensure controlled resource usage. - -You can find more details in our [API Reference](/reference/pipeline-api#pipeline), or directly in the pipeline's [GitHub code](https://github.com/deepset-ai/haystack/blob/main/haystack/core/pipeline/pipeline.py). - -## Example - -The examples on this page use Sentence Transformers embedders that have moved to the `sentence-transformers-haystack` package. Install it to run the examples: - -```shell -pip install sentence-transformers-haystack -``` - -```python -import asyncio - -from haystack import AsyncPipeline, Document -from haystack.components.builders import ChatPromptBuilder -from haystack_integrations.components.embedders.sentence_transformers import ( - SentenceTransformersDocumentEmbedder, - SentenceTransformersTextEmbedder, -) -from haystack.components.generators.chat import OpenAIChatGenerator -from haystack.components.joiners import DocumentJoiner -from haystack.components.retrievers import ( - InMemoryBM25Retriever, - InMemoryEmbeddingRetriever, -) -from haystack.dataclasses import ChatMessage -from haystack.document_stores.in_memory import InMemoryDocumentStore - -documents = [ - Document(content="Khufu is the largest pyramid."), - Document(content="Khafre is the middle pyramid."), - Document(content="Menkaure is the smallest pyramid."), -] - -docs_embedder = SentenceTransformersDocumentEmbedder() - -document_store = InMemoryDocumentStore() -document_store.write_documents(docs_embedder.run(documents=documents)["documents"]) - -prompt_template = [ - ChatMessage.from_system( - """ - You are a precise, factual QA assistant. - According to the following documents: - {% for document in documents %} - {{document.content}} - {% endfor %} - - If an answer cannot be deduced from the documents, say "I don't know based on these documents". - - When answering: - - be concise - - list the documents that support your answer - - Answer the given question. - """, - ), - ChatMessage.from_user("{{query}}"), - ChatMessage.from_system("Answer:"), -] - -hybrid_rag_retrieval = AsyncPipeline() -hybrid_rag_retrieval.add_component("text_embedder", SentenceTransformersTextEmbedder()) -hybrid_rag_retrieval.add_component( - "embedding_retriever", - InMemoryEmbeddingRetriever(document_store=document_store, top_k=3), -) -hybrid_rag_retrieval.add_component( - "bm25_retriever", - InMemoryBM25Retriever(document_store=document_store, top_k=3), -) -hybrid_rag_retrieval.add_component("document_joiner", DocumentJoiner()) -hybrid_rag_retrieval.add_component( - "prompt_builder", - ChatPromptBuilder(template=prompt_template), -) -hybrid_rag_retrieval.add_component("llm", OpenAIChatGenerator()) - -hybrid_rag_retrieval.connect( - "text_embedder.embedding", - "embedding_retriever.query_embedding", -) -hybrid_rag_retrieval.connect("bm25_retriever.documents", "document_joiner.documents") -hybrid_rag_retrieval.connect( - "embedding_retriever.documents", - "document_joiner.documents", -) -hybrid_rag_retrieval.connect("document_joiner.documents", "prompt_builder.documents") -hybrid_rag_retrieval.connect("prompt_builder.prompt", "llm.messages") - -question = "Which pyramid is neither the smallest nor the biggest?" - -data = { - "prompt_builder": {"query": question}, - "text_embedder": {"text": question}, - "bm25_retriever": {"query": question}, -} - - -async def process_results(): - async for partial_output in hybrid_rag_retrieval.run_async_generator( - data=data, - include_outputs_from={"document_joiner", "llm"}, - ): - if "document_joiner" in partial_output: - print( - "Retrieved documents:", - len(partial_output["document_joiner"]["documents"]), - ) - if "llm" in partial_output: - print("Generated answer:", partial_output["llm"]["replies"][0]) - - -asyncio.run(process_results()) -``` diff --git a/docs-website/docs/concepts/pipelines/pipeline-loops.mdx b/docs-website/docs/concepts/pipelines/pipeline-loops.mdx index 04c07966552..562cf6eb8f0 100644 --- a/docs-website/docs/concepts/pipelines/pipeline-loops.mdx +++ b/docs-website/docs/concepts/pipelines/pipeline-loops.mdx @@ -37,7 +37,7 @@ There are two main ways a loop ends: The pipeline finishes when the work queue is empty and no component can run again (for example, the router stops feeding inputs back into the loop). 2. **Reaching the maximum run count** - Every pipeline has a per-component run limit, controlled by the `max_runs_per_component` parameter of the `Pipeline` (or `AsyncPipeline`) constructor, which is `100` by default. If any component exceeds this limit, Haystack raises a `PipelineMaxComponentRuns` error. + Every pipeline has a per-component run limit, controlled by the `max_runs_per_component` parameter of the `Pipeline` constructor, which is `100` by default. If any component exceeds this limit, Haystack raises a `PipelineMaxComponentRuns` error. You can set this limit to a lower value: diff --git a/docs-website/docs/pipeline-components/builders/chatpromptbuilder.mdx b/docs-website/docs/pipeline-components/builders/chatpromptbuilder.mdx index cce7b879bd8..d102df0af9f 100644 --- a/docs-website/docs/pipeline-components/builders/chatpromptbuilder.mdx +++ b/docs-website/docs/pipeline-components/builders/chatpromptbuilder.mdx @@ -117,6 +117,32 @@ This template format allows you to define `ChatMessage` sequences using Jinja2 s Compared to using a list of `ChatMessage`s, this format is more flexible and allows including structured parts like images in the templatized `ChatMessage`; to better understand this use case, check out the [multimodal example](#multimodal) in the Usage section below. +#### The `insert` Tag + +String templates also support an `{% insert %}` tag. It is a placeholder that evaluates an expression to a `ChatMessage` or a list of `ChatMessage` objects and expands it into the prompt, so messages provided at runtime can be interleaved with literal `{% message %}` blocks. For example, you can wrap the runtime messages with a system message above and a templated user message below, then pass the messages (and any template variables) at run time: + +```python +from haystack.components.builders import ChatPromptBuilder +from haystack.dataclasses import ChatMessage + +template = """ +{% message role="system" %}You are a helpful assistant.{% endmessage %} +{% insert messages %} +{% message role="user" %}{{ query }}{% endmessage %} +""" + +builder = ChatPromptBuilder(template=template) +result = builder.run( + messages=[ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello!")], + query="What's the weather?", +) +# result["prompt"] -> [system, user "Hi", assistant "Hello!", user "What's the weather?"] +``` + +All content types (tool calls, tool call results, images, reasoning, `name`, and `meta`) round trip without loss. A missing or empty value expands to nothing. + +The expression can be a plain variable (`{% insert messages %}`), a slice or index (`{% insert messages[-1:] %}`, `{% insert messages[-1] %}`), or a combination of variables (`{% insert previous + current %}`). Multiple `{% insert %}` tags can be used in a single template, so the runtime messages can be split, reordered, or repeated across different positions. + ### Jinja2 Time Extension `PromptBuilder` supports the Jinja2 TimeExtension, which allows you to work with datetime formats. diff --git a/docs-website/docs/tools/pipelinetool.mdx b/docs-website/docs/tools/pipelinetool.mdx index 3d46c64a980..3ee72c258c9 100644 --- a/docs-website/docs/tools/pipelinetool.mdx +++ b/docs-website/docs/tools/pipelinetool.mdx @@ -27,11 +27,11 @@ It replaces the older workflow of first wrapping a pipeline in a `SuperComponent `ComponentTool`. `PipelineTool` builds the tool parameter schema from the pipeline’s input sockets and uses the underlying components’ docstrings for input descriptions. You can choose which pipeline inputs and outputs to expose with -`input_mapping` and `output_mapping`. It works with both `Pipeline` and `AsyncPipeline` and can be used in a pipeline with `ToolInvoker` or directly with the `Agent` component. +`input_mapping` and `output_mapping`. It can be used in a pipeline with `ToolInvoker` or directly with the `Agent` component. ### Parameters -- `pipeline` is mandatory and must be a `Pipeline` or `AsyncPipeline` instance. +- `pipeline` is mandatory and must be a `Pipeline` instance. - `name` is mandatory and specifies the tool name. - `description` is mandatory and explains what the tool does. - `input_mapping` is optional. It maps tool input names to pipeline input socket paths. If omitted, a default mapping is created from all pipeline inputs. diff --git a/docs-website/docs/tools/searchabletoolset.mdx b/docs-website/docs/tools/searchabletoolset.mdx index d6c971da6dc..3cfa6a32ae2 100644 --- a/docs-website/docs/tools/searchabletoolset.mdx +++ b/docs-website/docs/tools/searchabletoolset.mdx @@ -110,7 +110,7 @@ toolset = SearchableToolset( ### Reusing the toolset across multiple agent runs -When reusing the same `SearchableToolset` instance across multiple agent runs, you can call `clear()` to reset any tools discovered in the previous run: +You can safely reuse the same `SearchableToolset` instance across multiple agent runs, including concurrent ones. Each `Agent` run operates on an isolated, run-scoped copy of the toolset (created with [`spawn()`](toolset.mdx#run-scoped-copies-and-tool-selection)), so tools discovered in one run do not persist into, or collide with, other runs — every run starts fresh from the catalog: ```python agent = Agent( @@ -120,8 +120,8 @@ agent = Agent( result1 = agent.run(messages=[ChatMessage.from_user("What's the weather in Milan?")]) -# Reset discovered tools before the next run -toolset.clear() - +# The next run starts fresh: tools discovered in the previous run are not carried over result2 = agent.run(messages=[ChatMessage.from_user("Search for news about AI.")]) ``` + +If you drive the toolset directly (outside an `Agent`), you can call `clear()` to reset the discovered tools yourself. diff --git a/docs-website/docs/tools/toolset.mdx b/docs-website/docs/tools/toolset.mdx index 48200d803a9..27147ac9226 100644 --- a/docs-website/docs/tools/toolset.mdx +++ b/docs-website/docs/tools/toolset.mdx @@ -78,6 +78,17 @@ math_toolset.add(multiply_numbers) math_toolset.add(another_toolset) ``` +### Run-Scoped Copies and Tool Selection + +A `Toolset` is never mutated in place during an [`Agent`](../pipeline-components/agents-1/agent.mdx) run. Each run operates on an isolated, run-scoped copy of the configured `Toolset`, created with the `spawn()` method. This makes concurrent runs that share the same `Toolset` instance safe: per-run state, such as an active tool-name selection or a [`SearchableToolset`](searchabletoolset.mdx)'s discovered tools, cannot leak or collide across runs. + +You can also restrict an `Agent` to a subset of tools at runtime by passing tool names, for example `agent.run(tools=["tool_a", "tool_b"])`. When a `Toolset` is configured, the selection is applied to the live (run-scoped) `Toolset` rather than flattening it into a static list, so dynamic behavior like a `SearchableToolset`'s search and lazy loading keeps working over the selected subset. + +Two methods support this and can be overridden when subclassing: + +- `get_selectable_tools()`: Returns every tool available for name-based selection, ignoring any active selection restriction. Override it if your subclass's iteration does not surface every selectable tool. +- `spawn()`: Returns an isolated, run-scoped copy of the `Toolset`. Override it if your subclass holds additional run-scoped state. + ## Usage You can use `Toolset` wherever you can use Tools in Haystack. diff --git a/docs-website/docusaurus.config.js b/docs-website/docusaurus.config.js index c6c57f3ea3f..d9b4037ce45 100644 --- a/docs-website/docusaurus.config.js +++ b/docs-website/docusaurus.config.js @@ -185,6 +185,10 @@ j=d.createElement(s),dl=l!='dataLayer'?'&l='+l:'';j.async=true;j.src= from: '/docs/pipeline', to: '/docs/pipelines', }, + { + from: '/docs/asyncpipeline', + to: '/docs/pipelines', + }, { from: '/docs/prompt_node', to: '/docs/promptbuilder', diff --git a/docs-website/sidebars.js b/docs-website/sidebars.js index 3b1431d36c3..c137c1d80f5 100644 --- a/docs-website/sidebars.js +++ b/docs-website/sidebars.js @@ -66,7 +66,6 @@ export default { 'concepts/pipelines/debugging-pipelines', 'concepts/pipelines/pipeline-breakpoints', 'concepts/pipelines/pipeline-loops', - 'concepts/pipelines/asyncpipeline', 'concepts/pipelines/smart-pipeline-connections', ], },