Haystack AI Orchestration Framework

Haystack is an open-source AI orchestration framework for building production-ready LLM applications. Key capabilities include:

• Production-ready Agents: Supports lifecycle hooks (e.g., before_llm, before_tool, on_exit) for guardrails, and built-in monitoring for step_count, token_usage, and tool calls. Includes Agent Pack for ready-made agents and SkillToolset for progressive skill discovery.
• Context Engineering: Explicit control over retrieval, ranking, filtering, combining, structuring, and routing of information.
• Native Async Support: Pipeline supports synchronous and asynchronous execution with token-by-token streaming. Agent supports concurrent tool calls.
• Modularity: Built-in components for retrieval, indexing, tool calling, memory, and evaluation, with support for loops, branches, and conditional logic.
• Model Agnostic: Integrates with OpenAI, Mistral, Anthropic, Cohere, Hugging Face, Google, Azure OpenAI, AWS Bedrock, and local models.
• Extensibility: Consistent interfaces for building and sharing custom components.

The CometAPIChatGenerator provides access to over 500 AI models (including OpenAI, Anthropic, Google, xAI, and DeepSeek) through a single unified API gateway. It allows you to switch between different model providers within a single pipeline using a single API key.

Key Details:

• Package Name: cometapi-haystack
• Typical Pipeline Position: After a ChatPromptBuilder.
• Mandatory Initialization: Requires an api_key (can be set via the COMET_API_KEY environment variable).
• Mandatory Run Input: A list of messages as ChatMessage objects.
• Output: A list of ChatMessage objects in the replies key.
• Model Examples: gpt-5-mini (default), claude-sonnet-4-5, gemini-2.5-pro, grok-4.3, deepseek-chat.

from haystack_integrations.components.generators.cometapi import CometAPIChatGenerator

generator = CometAPIChatGenerator(api_key="YOUR_API_KEY", model="gpt-4o")
# Use in a pipeline or call run directly
# response = generator.run(messages=[...])

A Toolset is a collection of related Tool objects that can be managed as a single unit. It is useful for organizing tools into logical groups (e.g., a math_toolset) and passing them to components like an Agent or chat generators.

Toolset implements the collection interface (__iter__, __contains__, __len__, __getitem__), so it behaves like a list of tools and is compatible with any component expecting an iterable of tools.

To create a toolset, use the @tool decorator to define individual tools and then wrap them in a Toolset object.

from typing import Annotated
from haystack.tools import tool, Toolset
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator

@tool
def add(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
    '''Add two numbers.'''
    return a + b

@tool
def subtract(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
    '''Subtract b from a.'''
    return a - b

# Create a toolset with the math tools
math_toolset = Toolset([add, subtract])

# Use the toolset with an Agent
agent = Agent(chat_generator=OpenAIChatGenerator(), tools=math_toolset)

PipelineTool is a utility that wraps an existing Haystack Pipeline so it can be used as a tool by an LLM (typically via an Agent).

Key features:

• Automatic Schema Generation: It builds the tool's JSON parameter schema using the pipeline's input sockets and uses component docstrings for descriptions.
• Mapping: You can use input_mapping and output_mapping to control which pipeline sockets are exposed to the tool.
• Async Support: It natively supports async invocation (e.g., when called from Agent.run_async) because it leverages the pipeline's run_async method.
• Replacement: It replaces the older pattern of wrapping a pipeline in a SuperComponent and then using ComponentTool.

from haystack.tools import PipelineTool

retrieval_tool = PipelineTool(
    pipeline=retrieval_pipeline,
    name="retrieval_tool",
    description="Search short articles about Nikola Tesla...",
    input_mapping={"query": ["bm25_retriever.query", "ranker.query"]},
    output_mapping={"ranker.documents": "documents"}
)

The SearchableToolset allows an LLM to discover and use tools from large catalogs using BM25-based search. Instead of overwhelming the LLM context by exposing all tools at once, it provides a search_tools bootstrap tool. The LLM uses this tool to find and load specific tools as needed.

Key Behavior:

• Discovery Mode: If the catalog size is above the search_threshold, the agent only sees the search_tools tool and must search to find others.
• Passthrough Mode: If the catalog size is below search_threshold, all tools are exposed directly without discovery.

Important Methods:

• warm_up(): Indexes the catalog and creates the search_tools bootstrap tool. It is idempotent.
• spawn(selected_tool_names=...): Returns an isolated copy for a single run. This is recommended for concurrent runs to prevent discovered tools from colliding between different agent sessions.
• clear(): Resets discovered tools, useful for long-running applications to control memory or start fresh.

from typing import Annotated
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import SearchableToolset, tool

@tool
def get_weather(city: Annotated[str, "The city to get the weather for"]) -> str:
    '''Get the current weather for a city.'''
    return f"The weather in {city} is 22°C and sunny."

@tool
def search_web(query: Annotated[str, "The query to search the web for"]) -> str:
    '''Search the web for a query.'''
    return f"Top result for '{query}': ..."

@tool
def convert_currency(
    amount: Annotated[float, "The amount to convert"],
    to_currency: Annotated[str, "The currency to convert to, e.g. 'EUR'"],
) -> str:
    '''Convert an amount in USD to another currency.'''
    return f"{amount} USD is {amount * 0.9} {to_currency}"

# search_threshold=2 means a catalog of 2+ tools activates discovery
toolset = SearchableToolset(catalog=[get_weather, search_web, convert_currency], search_threshold=2)

agent = Agent(chat_generator=OpenAIChatGenerator(), tools=toolset)

result = agent.run(messages=[ChatMessage.from_user("What's the weather in Milan?")])
print(result["last_message"].text)

The AgentTool class wraps a Haystack Agent so that it can be used as a tool by another Agent. This enables multi-agent architectures where a 'coordinator' agent delegates specialized tasks to 'specialist' agents.

Key Benefits:

• Context Isolation: The specialist agent performs its own internal reasoning, searches, and tool calls within its own context. Only the final result is passed back to the coordinator, keeping the coordinator's context window small and efficient.
• Specialization: You can create agents with specific system prompts and toolsets and expose them as modular tools.

from haystack.tools import AgentTool

# researcher is an existing Agent instance
research_tool = AgentTool(
    agent=researcher,
    name="research",
    description="Research a question on the web and report the findings",
)

# coordinator can now use research_tool in its tools list
coordinator = Agent(
    chat_generator=chat_gen,
    tools=[research_tool],
    system_prompt="You coordinate specialists...",
)

The ValkeyDocumentStore is a Haystack DocumentStore implementation that uses Valkey with the Valkey Search module to provide persistent storage and vector similarity search. It is designed for high-performance retrieval requiring both semantic search and metadata filtering.

Key Features:

• Vector Search: Uses the HNSW algorithm for similarity search.
• Distance Metrics: Supports l2, cosine, and ip (inner product).
• Metadata Filtering: Supports filtering on both tag (string) and numeric fields.
• Operational Modes: Supports both synchronous and asynchronous operations, as well as cluster and standalone modes.
• Batching: Supports batch operations for efficient document management.

from haystack import Document
from haystack_integrations.document_stores.valkey import ValkeyDocumentStore

document_store = ValkeyDocumentStore(
    nodes_list=[("localhost", 6379)],
    index_name="my_documents",
    embedding_dim=768,
    distance_metric="cosine"
)

When an Agent has access to a very large number of tools, exposing them all at once can overwhelm the LLM's context window. SearchableToolset solves this by providing a search_tools bootstrap tool. Instead of seeing every tool, the Agent uses this search tool to find and load specific tools from a BM25-indexed catalog as needed.

If the number of tools in the catalog is below the search_threshold, the toolset acts as a simple passthrough, exposing all tools directly without the discovery mechanism.

from typing import Annotated
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import SearchableToolset, tool

@tool
def get_weather(city: Annotated[str, "The city to get the weather for"]) -> str:
    '''Get the current weather for a city.'''
    return f"The weather in {city} is 22°C and sunny."

@tool
def search_web(query: Annotated[str, "The query to search the web for"]) -> str:
    '''Search the web for a query.'''
    return f"Top result for '{query}': ..."

@tool
def convert_currency(
    amount: Annotated[float, "The amount to convert"],
    to_currency: Annotated[str, "The currency to convert to, e.g. 'EUR'"],
) -> str:
    '''Convert an amount in USD to another currency.'''
    return f"{amount} USD is {amount * 0.9} {to_currency}"

# search_threshold=2 means a catalog of 2+ tools activates discovery
toolset = SearchableToolset(catalog=[get_weather, search_web, convert_currency], search_threshold=2)

agent = Agent(chat_generator=OpenAIChatGenerator(), tools=toolset)

result = agent.run(messages=[ChatMessage.from_user("What's the weather in Milan?")])
print(result["last_message"].text)

A Toolset is a collection of related Tool objects that can be managed as a single unit. It serves two primary purposes:

1. Grouping: You can organize multiple tools into a single collection to simplify management in Haystack pipelines, such as passing a single Toolset to an Agent instead of a list of individual tools.
2. Dynamic Loading: Toolset acts as a base class for custom implementations that load tools dynamically from external sources like OpenAPI URLs or MCP servers.

Toolset implements the collection interface (__iter__, __contains__, __len__, __getitem__), meaning it behaves like a list of tools and is compatible with components like Agent or Haystack chat generators that expect an iterable of tools.

from typing import Annotated
from haystack.tools import tool, Toolset
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator

@tool
def add(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
    '''Add two numbers.'''
    return a + b

@tool
def subtract(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
    '''Subtract b from a.'''
    return a - b

# Create a toolset with the math tools
math_toolset = Toolset([add, subtract])

# Use the toolset with an Agent
agent = Agent(chat_generator=OpenAIChatGenerator(), tools=math_toolset)

An AgentTool is a specialized tool that wraps a Haystack Agent, allowing it to be used as a tool by another Agent. This is the primary building block for multi-agent systems where one Agent (the coordinator) delegates specific tasks to another Agent (the specialist).

When an Agent uses an AgentTool, it only sees the final reply from the wrapped Agent, keeping the internal reasoning steps of the specialist out of the coordinator's context. By default, the task is delegated as a single user message and returned as text.

# Example of a specialist agent being wrapped as a tool
researcher = Agent(
    chat_generator=OpenAIResponsesChatGenerator(model="gpt-5.4-mini"),
    system_prompt="You are a research specialist.",
    tools=[ComponentTool(component=SerperDevWebSearch(), name="web_search", description="...")]
)

research_tool = AgentTool(
    agent=researcher,
    name="research",
    description="Research a question on the web and report the findings"
)

# The coordinator agent uses the research_tool
coordinator = Agent(
    chat_generator=OpenAIResponsesChatGenerator(model="gpt-5.4"),
    tools=[research_tool],
    system_prompt="You coordinate specialists."
)