[Python] How to add function tools in the haystack pipeline?

I just started learning to create a chatbot using Haystack. Previously I was using Haystack 2.6 and just upgraded to 2.12.

Then I just compiled, copied, and tried all the codes that I found so they could be used in my chatbot.

Then I want to add a function calling tools in my chatbot. I want my chatbot can do a web searches to find results if it doesn't have any information from the RAG documents that I provided.

So I'm trying to integrate the DuckDuckGo into my chatbot.

I just following this (Haystack Agents) example from Haystack Docs but it's using Agents, and I'm not.

Actually, the chatbot is working, if I'm asking something it's returning a response. But when I ask a question that forces the chatbot to look into the current data by doing a web search, it says "I didn't have real-time data" which means the function tool for web search is not working.

I have no idea why the web search tool is not working. I think definitely there is an error or problem when I connect the component for the pipeline, or maybe I miss some components when adding the tool.

This is my pipeline:

import logging
import uuid
from typing import List
from haystack import Pipeline, Document
from haystack.utils import Secret
from haystack.tools import ComponentTool
from haystack.dataclasses import ChatMessage, ChatRole
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.tools import ToolInvoker
from haystack.components.routers import ConditionalRouter
from get_chatbot_data import get_all_data
from duckduckgo_api_haystack import DuckduckgoApiWebSearch

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ChatbotRAGPipeline:
def __init__(self, model="gpt-4o-mini"):
self.model = model
self.prompt_template = self._create_system_prompt()
self.document_store = self._initialize_document_store()
self.llm = self._initialize_language_model()
self.retriever = self._initialize_retriever()
self.prompt_builder = self._initialize_prompt_builder()
self.web_route = self._initialize_web_route()
self.web_tool = self._initialize_web_tool()
self.rag_pipeline = self._initialize_rag_pipeline()

def _initialize_web_route(self):
routes = [
{
"condition": "{{replies[0].tool_calls | length > 0}}",
"output": "{{replies}}",
"output_name": "there_are_tool_calls",
"output_type": List[ChatMessage],
},
{
"condition": "{{replies[0].tool_calls | length == 0}}",
"output": "{{replies}}",
"output_name": "final_replies",
"output_type": List[ChatMessage],
},
]
return routes

def _initialize_web_tool(self):
web_tool = ComponentTool(
name="WebSearch",
description="Use this to search the web when you need real-time or recent information.",
component=DuckduckgoApiWebSearch(top_k=10)
)
return web_tool

def _create_system_prompt(self):
return """
You are an AI assistant designed to provide **clear, concise, and accurate** responses. Your goal is to **help users efficiently** while providing recommendations only if the question relates to something you can recommend. Your responses should be **direct, informative, and polite**, without unnecessary details or filler content.

**Conversation Context**:
{% if conversation_history %}
Previous conversation history:
{{ conversation_history }}
{% else %}
This is a new conversation.
{% endif %}

**Documents**:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}

**Image Description**
{% if image_description %}
An image description has been provided. Use the description to assist with the response, including any user-related details that might be inferred.
{{ image_description }}
{% else %}
No image description provided.
{% endif %}

**Question**: {{ question }}

**Answer**:
"""

def _initialize_document_store(self):
try:
datas, _ = get_all_data()
documents = [
Document(
content=data["content"],
meta=data["meta"],
id=str(uuid.uuid4())
) for data in datas
]
document_store = InMemoryDocumentStore()
document_store.write_documents(documents)
return document_store
except Exception as e:
raise RuntimeError(f"Error initializing document store: {e}")

def _initialize_language_model(self):
api_key = "API_KEY"
model_name = self.model
return OpenAIGenerator(api_key=Secret.from_token(api_key), model=model_name)

def _initialize_retriever(self):
retriever = InMemoryBM25Retriever(document_store=self.document_store, top_k=50)
return retriever

def _initialize_prompt_builder(self):
prompt_builder = PromptBuilder(template=self.prompt_template)
return prompt_builder

def _initialize_rag_pipeline(self):
pipeline = Pipeline()
pipeline.add_component("retriever", self.retriever)
pipeline.add_component("prompt_builder", self.prompt_builder)
pipeline.add_component("llm", self.llm)
pipeline.add_component("router", ConditionalRouter(self.web_route))
pipeline.add_component("tool_invoker", ToolInvoker(tools=[self.web_tool]))

pipeline.connect("retriever.documents", "prompt_builder.documents")
pipeline.connect("prompt_builder", "llm.prompt")
pipeline.connect("llm.replies", "router.replies")
pipeline.connect("router.there_are_tool_calls", "tool_invoker")
return pipeline

Continue reading...