教程：使用 Haystack 创建多智能体系统

%%bash

pip install -q haystack-ai duckduckgo-api-haystack

from getpass import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass("Enter your OpenAI API key:")
if not os.environ.get("NOTION_API_KEY"):
    os.environ["NOTION_API_KEY"] = getpass("Enter your NOTION API key:")

from haystack.tools import ComponentTool
from duckduckgo_api_haystack import DuckduckgoApiWebSearch


def doc_to_string(documents) -> str:
    """
    Handles the tool output before conversion to ChatMessage.
    """
    result_str = ""
    for document in documents:
        result_str += f"File Content for {document.meta['link']}\n\n {document.content}"

    if len(result_str) > 150_000:  # trim if the content is too large
        result_str = result_str[:150_000] + "...(large file can't be fully displayed)"

    return result_str


web_search = ComponentTool(
    component=DuckduckgoApiWebSearch(top_k=5, backend="lite"),
    name="web_search",
    description="Search the web",
    outputs_to_string={"source": "documents", "handler": doc_to_string},
)

wiki_search = ComponentTool(
    component=DuckduckgoApiWebSearch(top_k=5, backend="lite", allowed_domain="https://en.wikipedia.org"),
    name="wiki_search",
    description="Search Wikipedia",
    outputs_to_string={"source": "documents", "handler": doc_to_string},
)

# from getpass import getpass
# import os

# from haystack.components.websearch import SerperDevWebSearch
# from haystack.tools import ComponentTool

# if not os.environ.get("SERPERDEV_API_KEY"):
#     os.environ["SERPERDEV_API_KEY"] = getpass("Enter your SERPER API key:")

# web_search = ComponentTool(
#     component=SerperDevWebSearch(top_k=5),
#     name="web_search",
#     description="Search the web",
#     outputs_to_string={"source": "documents", "handler": doc_to_string},
# )

# wiki_search = ComponentTool(
#     component=SerperDevWebSearch(top_k=5, allowed_domains=["https://www.wikipedia.org/", "https://en.wikipedia.org"]),
#     name="wiki_search",
#     description="Search Wikipedia",
#     outputs_to_string={"source": "documents", "handler": doc_to_string},
# )

from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
from haystack.components.generators.utils import print_streaming_chunk
from haystack.components.generators.chat import OpenAIChatGenerator

research_agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
    system_prompt="""
    You are a research agent that can find information on web or specifically on wikipedia.
    Use wiki_search tool if you need facts and use web_search tool for latest news on topics.
    Use one tool at a time. Try different queries if you need more information.
    Only use the retrieved context, do not use your own knowledge.
    Summarize the all retrieved information before returning response to the user.
    """,
    tools=[web_search, wiki_search],
    streaming_callback=print_streaming_chunk,
)

result = research_agent.run(
    messages=[ChatMessage.from_user("Can you tell me about Florence Nightingale's contributions to nursery?")]
)

print("Final Answer:", result["last_message"].text)

from haystack import component
from typing import Optional
from haystack.utils import Secret
import requests


@component
class NotionPageCreator:
    """
    Create a page in Notion using provided title and content.
    """

    def __init__(
        self,
        page_id: str,
        notion_version: str = "2022-06-28",
        api_key: Secret = Secret.from_env_var("NOTION_API_KEY"),  # to use the environment variable NOTION_API_KEY
    ):
        """
        Initialize with the target Notion database ID and API version.
        """
        self.api_key = api_key
        self.notion_version = notion_version
        self.page_id = page_id

    @component.output_types(success=bool, status_code=int, error=Optional[str])
    def run(self, title: str, content: str):
        """
        :param title: The title of the Notion page.
        :param content: The content of the Notion page.
        """
        headers = {
            "Authorization": f"Bearer {self.api_key.resolve_value()}",
            "Content-Type": "application/json",
            "Notion-Version": self.notion_version,
        }

        payload = {
            "parent": {"page_id": self.page_id},
            "properties": {"title": [{"text": {"content": title}}]},
            "children": [
                {
                    "object": "block",
                    "type": "paragraph",
                    "paragraph": {"rich_text": [{"type": "text", "text": {"content": content}}]},
                }
            ],
        }

        response = requests.post("https://api.notion.com/v1/pages", headers=headers, json=payload)

        if response.status_code == 200 or response.status_code == 201:
            return {"success": True, "status_code": response.status_code}
        else:
            return {"success": False, "status_code": response.status_code, "error": response.text}

notion_writer = NotionPageCreator(page_id="<your_page_id>")
notion_writer.run(title="My first page", content="The content of my first page")

    def run(self, title: str, content: str):
        """
        :param title: The title of the Notion page.
        :param content: The content of the Notion page.
        """

from haystack.tools import ComponentTool

notion_writer = ComponentTool(
    component=NotionPageCreator(page_id="<your_page_id>"),
    name="notion_writer",
    description="Use this tool to write/save content to Notion.",
)
notion_writer.parameters  # see how parameters are automatically generated by the ComponentTool

from haystack import Pipeline, component, Document, SuperComponent
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from typing import List


@component
class DocumentAdapter:
    @component.output_types(documents=List[Document])
    def run(self, content: str, title: str):
        return {"documents": [Document(content=content, meta={"title": title})]}


document_store = InMemoryDocumentStore()

doc_store_writer_pipeline = Pipeline()
doc_store_writer_pipeline.add_component("adapter", DocumentAdapter())
doc_store_writer_pipeline.add_component("writer", DocumentWriter(document_store=document_store))
doc_store_writer_pipeline.connect("adapter", "writer")

doc_store_writer = ComponentTool(
    component=SuperComponent(doc_store_writer_pipeline),
    name="doc_store_writer",
    description="Use this tool to write/save content to document store",
    parameters={
        "type": "object",
        "properties": {
            "title": {"type": "string", "description": "The title of the Document"},
            "content": {"type": "string", "description": "The content of the Document"},
        },
        "required": ["title", "content"],
    },
)
doc_store_writer.parameters

from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
from haystack.components.generators.utils import print_streaming_chunk
from haystack.components.generators.chat import OpenAIChatGenerator

writer_agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
    system_prompt="""
    You are a writer agent that saves given information to different locations.
    Do not change the provided content before saving.
    Infer the title from the text if not provided.
    When you need to save provided information to Notion, use notion_writer tool.
    When you need to save provided information to document store, use doc_store_writer tool
    If no location is mentioned, use notion_writer tool to save the information.
    """,
    tools=[doc_store_writer, notion_writer],
    streaming_callback=print_streaming_chunk,
    exit_conditions=["notion_writer", "doc_store_writer"],
)

result = writer_agent.run(
    messages=[
        ChatMessage.from_user(
            """
Save this text on Notion:

Florence Nightingale is widely recognized as the founder of modern nursing, and her contributions significantly transformed the field.
Florence Nightingale's legacy endures, as she set professional standards that have shaped nursing into a respected and essential component of the healthcare system. Her influence is still felt in nursing education and practice today.
"""
        )
    ]
)

from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk

research_tool = ComponentTool(
    component=research_agent,
    description="Use this tool to find information on web or specifically on wikipedia",
    name="research_tool",
    outputs_to_string={"source": "last_message"},
)
writer_tool = ComponentTool(
    component=writer_agent,
    description="Use this tool to write content into document store or Notion",
    name="writer_tool",
    outputs_to_string={"source": "last_message"},
)

main_agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
    system_prompt="""
    You are an assistant that has access to several tools.
    Understand the user query and use relevant tool to answer the query.
    You can use `research_tool` to make research on web and wikipedia and `writer_tool` to save information into the document store or Notion.
    """,
    streaming_callback=print_streaming_chunk,
    tools=[research_tool, writer_tool],
)

from haystack.dataclasses import ChatMessage

result = main_agent.run(
    messages=[
        ChatMessage.from_user(
            """
            Can you research the history of the Silk Road?
            """
        )
    ]
)

result["last_message"].text

result = main_agent.run(
    messages=[
        ChatMessage.from_user(
            """
            Summarize how RAG pipelines work and save it in Notion
            """
        )
    ]
)