Wie füge ich Funktionstools in der Haystack -Pipeline hinzu? ⇐ Python

[Anonymous]

Ich habe gerade angefangen zu lernen, einen Chatbot mit Haystack zu erstellen. Zuvor habe ich Haystack 2.6 verwendet und gerade auf 2.12 aktualisiert. Ich möchte, dass mein Chatbot eine Websuche durchführen kann, um Ergebnisse zu finden, wenn es keine Informationen aus den von mir bereitgestellten Lag -Dokumenten enthält. Arbeiten, wenn ich etwas frage, erwidert es eine Antwort. Wenn ich jedoch eine Frage stelle, die den Chatbot dazu zwingt, die aktuellen Daten durch eine Websuche durchzuführen, heißt es in "Ich hatte keine Echtzeitdaten", was bedeutet, dass das Funktionstool für die Websuche nicht funktioniert. Ich denke, es gibt definitiv einen Fehler oder Problem, wenn ich die Komponente für die Pipeline anschließe, oder ich vermisse einige Komponenten beim Hinzufügen des Werkzeugs.

Code: Select all

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