Haystack agent pipeline tutorial for customer support

This haystack agent pipeline customer support tutorial walks you through building a production-grade support agent that retrieves knowledge, calls external APIs, and returns structured responses. We’ll use Haystack 2.0’s native agent architecture with ToolInvoker, ChatGenerator, and a custom RAG pipeline — no legacy Pipeline objects, no PromptNode wrappers. By the end you’ll have a runnable agent that classifies tickets, fetches relevant docs, queries an order API, and emits JSON your frontend can consume directly.

Prerequisites

• Python 3.10+
• Haystack 2.7+ (pip install haystack-ai)
• An OpenAI-compatible endpoint (we’ll use gpt-4o-mini for speed)
• Optional: n4n.ai API key if you want automatic fallback across 240+ models without managing provider credentials yourself

pip install haystack-ai python-dotenv requests

Create a .env file:

OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1  # or your n4n.ai endpoint

Project structure

support_agent/
├── main.py
├── tools/
│   ├── __init__.py
│   ├── order_lookup.py
│   └── knowledge_search.py
├── pipelines/
│   ├── __init__.py
│   └── rag_pipeline.py
├── schemas/
│   └── response.py
└── config.py

Configuration and schemas

Start with typed configuration and the response contract your frontend expects.

# config.py
import os
from dataclasses import dataclass

@dataclass
class Settings:
    model: str = "gpt-4o-mini"
    temperature: float = 0.1
    max_tokens: int = 1024
    embedding_model: str = "text-embedding-3-small"
    top_k: int = 4
    api_base: str = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    api_key: str = os.getenv("OPENAI_API_KEY", "")

settings = Settings()

# schemas/response.py
from pydantic import BaseModel, Field
from typing import Literal, Optional
from enum import Enum

class TicketCategory(str, Enum):
    ORDER_ISSUE = "order_issue"
    TECHNICAL = "technical"
    BILLING = "billing"
    GENERAL = "general"

class SupportResponse(BaseModel):
    category: TicketCategory
    answer: str
    confidence: float = Field(ge=0.0, le=1.0)
    sources: list[str] = []
    order_data: Optional[dict] = None
    escalate: bool = False
    escalation_reason: Optional[str] = None

Tool: Order lookup

A realistic support agent needs to query an order system. We’ll simulate a REST endpoint with latency and error modes.

# tools/order_lookup.py
import time
import random
from typing import Optional
from dataclasses import dataclass

@dataclass
class Order:
    order_id: str
    status: str
    items: list[dict]
    shipping_address: dict
    tracking_number: Optional[str] = None
    estimated_delivery: Optional[str] = None

MOCK_ORDERS = {
    "ORD-1001": Order("ORD-1001", "delivered", [{"sku": "WH-1000XM5", "qty": 1}], 
                      {"city": "Austin", "state": "TX"}, "1Z999AA10123456784", "2024-01-15"),
    "ORD-1002": Order("ORD-1002", "shipped", [{"sku": "IPHONE-15-PRO", "qty": 1}], 
                      {"city": "Seattle", "state": "WA"}, "1Z999AA10123456785", "2024-01-20"),
    "ORD-1003": Order("ORD-1003", "processing", [{"sku": "MBP-16-M3", "qty": 1}], 
                      {"city": "Boston", "state": "MA"}, None, None),
}

def lookup_order(order_id: str) -> dict:
    """Simulate order API with realistic latency and occasional failures."""
    time.sleep(random.uniform(0.1, 0.3))
    
    if random.random() < 0.05:
        raise ConnectionError("Order service temporarily unavailable")
    
    order = MOCK_ORDERS.get(order_id)
    if not order:
        return {"error": "Order not found", "order_id": order_id}
    
    return {
        "order_id": order.order_id,
        "status": order.status,
        "items": order.items,
        "shipping_address": order.shipping_address,
        "tracking_number": order.tracking_number,
        "estimated_delivery": order.estimated_delivery,
    }

Tool: Knowledge search (RAG)

We’ll build a minimal in-memory document store for FAQ content. In production you’d swap this for Pinecone, Weaviate, or pgvector.

# tools/knowledge_search.py
from haystack import Document
from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack import Pipeline
from config import settings

FAQ_DOCS = [
    Document(content="Return policy: 30 days from delivery for full refund. Items must be unused with original packaging.", meta={"topic": "returns"}),
    Document(content="Shipping: Free standard shipping on orders over $50. Express shipping $15 flat rate.", meta={"topic": "shipping"}),
    Document(content="Warranty: All electronics carry 1-year manufacturer warranty. Extended warranty available at checkout.", meta={"topic": "warranty"}),
    Document(content="Payment methods: We accept Visa, Mastercard, Amex, PayPal, Apple Pay, and Google Pay.", meta={"topic": "payment"}),
    Document(content="Order changes: Cancellations allowed within 1 hour of placement. After that, contact support for return.", meta={"topic": "cancellation"}),
    Document(content="International shipping: Available to 40+ countries. Duties and taxes calculated at checkout.", meta={"topic": "international"}),
]

def build_knowledge_pipeline() -> Pipeline:
    doc_store = InMemoryDocumentStore()
    
    # Embed and index FAQ docs
    embedder = SentenceTransformersDocumentEmbedder(model=settings.embedding_model)
    embedder.warm_up()
    docs_with_embeddings = embedder.run(FAQ_DOCS)["documents"]
    doc_store.write_documents(docs_with_embeddings)
    
    # Retrieval pipeline
    rag = Pipeline()
    rag.add_component("text_embedder", SentenceTransformersTextEmbedder(model=settings.embedding_model))
    rag.add_component("retriever", InMemoryEmbeddingRetriever(document_store=doc_store, top_k=settings.top_k))
    rag.connect("text_embedder.embedding", "retriever.query_embedding")
    
    return rag

knowledge_pipeline = build_knowledge_pipeline()

def search_knowledge(query: str) -> list[dict]:
    result = knowledge_pipeline.run({"text_embedder": {"text": query}})
    docs = result["retriever"]["documents"]
    return [{"content": d.content, "topic": d.meta.get("topic"), "score": d.score} for d in docs]

Checkpoint — test knowledge search:

# Quick test in REPL
from tools.knowledge_search import search_knowledge
print(search_knowledge("How do I return an item?"))

Expected output:

[
  {'content': 'Return policy: 30 days from delivery for full refund. Items must be unused with original packaging.', 'topic': 'returns', 'score': 0.87},
  {'content': 'Order changes: Cancellations allowed within 1 hour of placement. After that, contact support for return.', 'topic': 'cancellation', 'score': 0.72}
]

Agent pipeline assembly

Now the core: a Haystack 2.0 agent that uses ChatGenerator with tool calling, ToolInvoker, and our custom components.

# pipelines/rag_pipeline.py
from haystack import Pipeline
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.tools import ToolInvoker
from haystack.dataclasses import ChatMessage
from haystack.utils import Secret
from tools.order_lookup import lookup_order
from tools.knowledge_search import search_knowledge
from schemas.response import SupportResponse, TicketCategory
from config import settings
import json

SYSTEM_PROMPT = """You are a customer support agent for an electronics retailer.
Classify the user's issue, retrieve relevant information, and respond with structured JSON.

Categories: order_issue, technical, billing, general

Tools available:
- lookup_order: Get order status, tracking, items. Use when user mentions order ID.
- search_knowledge: Search FAQ for policies, shipping, returns, warranty, payments.

Rules:
1. Always classify the ticket category first.
2. If user provides order ID, call lookup_order.
3. For policy questions, call search_knowledge.
4. Respond ONLY with valid JSON matching the SupportResponse schema.
5. Set escalate=true only for: fraud suspicion, legal threats, repeated failures, or explicit request.
6. Confidence reflects how well you resolved the issue (0.0-1.0)."""

def build_agent_pipeline() -> Pipeline:
    llm = OpenAIChatGenerator(
        model=settings.model,
        api_base_url=settings.api_base,
        api_key=Secret.from_token(settings.api_key),
        generation_kwargs={"temperature": settings.temperature, "max_tokens": settings.max_tokens},
        tools=[lookup_order, search_knowledge],
    )
    
    tool_invoker = ToolInvoker(tools=[lookup_order, search_knowledge])
    
    pipe = Pipeline()
    pipe.add_component("llm", llm)
    pipe.add_component("tools", tool_invoker)
    
    # Agent loop: LLM -> tools -> LLM -> tools ... until final answer
    pipe.connect("llm.replies", "tools.messages")
    pipe.connect("tools.tool_messages", "llm.messages")
    
    return pipe

agent = build_agent_pipeline()

Main entry point

Wire it together with a clean CLI and proper error handling.

# main.py
import json
import sys
from pipelines.rag_pipeline import agent
from schemas.response import SupportResponse
from haystack.dataclasses import ChatMessage

def run_support_agent(user_message: str) -> SupportResponse:
    messages = [
        ChatMessage.from_system("""You are a customer support agent for an electronics retailer.
Classify the user's issue, retrieve relevant information, and respond with structured JSON.

Categories: order_issue, technical, billing, general

Tools available:
- lookup_order: Get order status, tracking, items. Use when user mentions order ID.
- search_knowledge: Search FAQ for policies, shipping, returns, warranty, payments.

Rules:
1. Always classify the ticket category first.
2. If user provides order ID, call lookup_order.
3. For policy questions, call search_knowledge.
4. Respond ONLY with valid JSON matching the SupportResponse schema.
5. Set escalate=true only for: fraud suspicion, legal threats, repeated failures, or explicit request.
6. Confidence reflects how well you resolved the issue (0.0-1.0)."""),
        ChatMessage.from_user(user_message),
    ]
    
    result = agent.run({"llm": {"messages": messages}})
    
    # Final LLM response is in tools.tool_messages (last iteration) or llm.replies
    final_messages = result.get("tools", {}).get("tool_messages", [])
    if not final_messages:
        final_messages = result.get("llm", {}).get("replies", [])
    
    if not final_messages:
        raise RuntimeError("Agent produced no response")
    
    last_msg = final_messages[-1]
    content = last_msg.text
    
    # Parse JSON from response (LLM may wrap in markdown)
    if "```json" in content:
        content = content.split("```json")[1].split("```")[0].strip()
    elif "```" in content:
        content = content.split("```")[1].split("```")[0].strip()
    
    return SupportResponse.model_validate_json(content)

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python main.py \"Your support question here\"")
        sys.exit(1)
    
    query = " ".join(sys.argv[1:])
    response = run_support_agent(query)
    print(json.dumps(response.model_dump(), indent=2))

Checkpoint — run the agent

python main.py "Where is my order ORD-1002?"

Expected output (order data will vary slightly due to mock):

{
  "category": "order_issue",
  "answer": "Your order ORD-1002 is currently shipped. It contains 1x IPHONE-15-PRO and is estimated to arrive by 2024-01-20. Tracking number: 1Z999AA10123456785. You can track it on the carrier's website.",
  "confidence": 0.95,
  "sources": [],
  "order_data": {
    "order_id": "ORD-1002",
    "status": "shipped",
    "items": [{"sku": "IPHONE-15-PRO", "qty": 1}],
    "shipping_address": {"city": "Seattle", "state": "WA"},
    "tracking_number": "1Z999AA10123456785",
    "estimated_delivery": "2024-01-20"
  },
  "escalate": false,
  "escalation_reason": null
}

python main.py "What's your return policy for headphones?"

Expected output:

{
  "category": "general",
  "answer": "Our return policy allows returns within 30 days of delivery for a full refund. Items must be unused and in their original packaging. This applies to headphones and all other electronics.",
  "confidence": 0.92,
  "sources": ["returns"],
  "order_data": null,
  "escalate": false,
  "escalation_reason": null
}

python main.py "I want to speak to a manager. This is fraud!"

Expected output:

{
  "category": "billing",
  "answer": "I understand you're concerned about potential fraud. I'm escalating this to a senior support specialist who can investigate your account and any suspicious charges. They'll reach out within 2 hours.",
  "confidence": 0.7,
  "sources": [],
  "order_data": null,
  "escalate": true,
  "escalation_reason": "User explicitly requested escalation and mentioned fraud"
}

Handling streaming and timeouts

Production agents need streaming for perceived latency. Haystack 2.0 supports streaming via OpenAIChatGenerator with a callback.

# Add to pipelines/rag_pipeline.py
from haystack.dataclasses import StreamingChunk
from typing import Callable

def build_streaming_agent(on_token: Callable[[str], None] | None = None) -> Pipeline:
    llm = OpenAIChatGenerator(
        model=settings.model,
        api_base_url=settings.api_base,
        api_key=Secret.from_token(settings.api_key),
        generation_kwargs={"temperature": settings.temperature, "max_tokens": settings.max_tokens},
        tools=[lookup_order, search_knowledge],
        streaming_callback=lambda chunk: on_token(chunk.content) if on_token and isinstance(chunk, StreamingChunk) else None,
    )
    # ... rest same as build_agent_pipeline

Usage:

def print_token(token: str):
    print(token, end="", flush=True)

streaming_agent = build_streaming_agent(on_token=print_token)
result = streaming_agent.run({"llm": {"messages": messages}})
print()  # newline after stream

Observability: logging tool calls

Wrap ToolInvoker to log every invocation — critical for debugging agent loops.

# tools/logging_invoker.py
from haystack.components.tools import ToolInvoker
from haystack.dataclasses import ChatMessage
import logging
import time

logger = logging.getLogger(__name__)

class LoggingToolInvoker(ToolInvoker):
    def run(self, messages: list[ChatMessage]):
        for msg in messages:
            if msg.tool_calls:
                for tc in msg.tool_calls:
                    logger.info(f"TOOL CALL: {tc.tool_name}({tc.arguments})")
                    start = time.time()
        
        result = super().run(messages)
        
        for msg in result["tool_messages"]:
            if msg.tool_call_result:
                logger.info(f"TOOL RESULT: {msg.tool_call_result.origin.name} -> {msg.tool_call_result.result[:200]}... (took {time.time()-start:.2f}s)")
        
        return result

Swap ToolInvoker for LoggingToolInvoker in build_agent_pipeline().

Common failure modes and fixes

Extending for production

1. Replace in-memory store — swap InMemoryDocumentStore for PineconeDocumentStore or WeaviateDocumentStore with same retriever interface.
2. Add authentication — wrap run_support_agent in FastAPI with JWT validation.
3. Rate limiting — use slowapi or provider-level limits; n4n.ai honors x-ratelimit-* headers and falls back automatically when a provider is degraded.
4. Evaluation — log every (query, response, human_feedback) tuple to a dataset; run periodic haystack.eval benchmarks against golden sets.
5. Multi-turn — persist messages history in Redis keyed by session_id; pass full history to agent.run().

Full file reference

support_agent/
├── config.py
├── main.py
├── tools/
│   ├── __init__.py
│   ├── order_lookup.py
│   ├── knowledge_search.py
│   └── logging_invoker.py
├── pipelines/
│   ├── __init__.py
│   └── rag_pipeline.py
└── schemas/
    └── response.py

Run the full test suite:

python -m pytest tests/ -v  # add your own tests

This haystack agent pipeline customer support tutorial gives you a working foundation. The agent architecture — ChatGenerator + ToolInvoker + typed Pydantic output — scales from this demo to thousands of daily tickets without restructuring. Swap components, add tools, measure, iterate.