Newsletter Sending Agent with Experimental Haystack Tools

! pip install haystack-ai trafilatura

from typing import List
from trafilatura import fetch_url, extract
import requests
from getpass import getpass
import os

from haystack_experimental.components.generators.chat import OpenAIChatGenerator
from haystack_experimental.dataclasses import Tool, ChatMessage
from haystack_experimental.components.tools import ToolInvoker

from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

def hacker_news_fetcher(top_k: int = 3):
    newest_list = requests.get(url='https://hacker-news.firebaseio.com/v0/topstories.json?print=pretty')
    urls = []
    articles = []
    for id_ in newest_list.json()[0:top_k]:
        article = requests.get(url=f"https://hacker-news.firebaseio.com/v0/item/{id_}.json?print=pretty")
        if 'url' in article.json():
            urls.append(article.json()['url'])
        elif 'text' in article.json():
            articles.append(article.json()['text'])

    for url in urls:
        try:
            downloaded = fetch_url(url)
            text = extract(downloaded)
            if text is not None:
                articles.append(text[:500])
        except Exception as e:
            print(e)
            print(f"Couldn't download {url}, skipped")

    return articles

hacker_news_fetcher_tool = Tool(name="hacker_news_fetcher",
                                description="Fetch the top k articles from hacker news",
                                function=hacker_news_fetcher,
                                parameters={
                                    "type": "object",
                                    "properties": {
                                        "top_k": {
                                            "type": "integer",
                                            "description": "The number of articles to fetch"
                                        }
                                    },
                                })

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

template = """
Create a entertaining newsletter for {{target_people}} based on the following articles.
The newsletter should be well structured, with a unique angle and a maximum of {{n_words}} words.

Articles:
{% for article in articles %}
    {{ article }}
    ---
{% endfor %}
"""

newsletter_pipe = Pipeline()
newsletter_pipe.add_component("prompt_builder", PromptBuilder(template=template))
newsletter_pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))
newsletter_pipe.connect("prompt_builder", "llm")

def newsletter_pipeline_func(articles: List[str], target_people: str = "programmers", n_words: int = 100):
    result = newsletter_pipe.run({"prompt_builder": {"articles": articles, "target_people": target_people, "n_words": n_words}})

    return {"reply": result["llm"]["replies"][0]}

newsletter_tool = Tool(name="newsletter_generator",
                          description="Generate a newsletter based on some articles",
                            function=newsletter_pipeline_func,
                            parameters={
                                "type": "object",
                                "properties": {
                                    "articles": {
                                        "type": "array",
                                        "items": {
                                            "type": "string",
                                            "description": "The articles to base the newsletter on",
                                        }
                                    },
                                    "target_people": {
                                        "type": "string",
                                        "description": "The target audience for the newsletter",
                                    },
                                    "n_words": {
                                        "type": "integer",
                                        "description": "The number of words to summarize the newsletter to",
                                    }
                                },
                                "required": ["articles"],
                            })

if not "NAME" in os.environ:
    os.environ["NAME"] = input("What's your name? ")
if not "SENDER_EMAIL" in os.environ:
    os.environ["SENDER_EMAIL"] = getpass("Enter your Gmail e-mail: ")
if not "GMAIL_APP_PASSWORD" in os.environ:
    os.environ["GMAIL_APP_PASSWORD"] = getpass("Enter your Gmail App Password: ")

import smtplib, ssl
from email.mime.text import MIMEText

def send_email(receiver: str, body: str, subject: str):
  msg = MIMEText(body)
  sender_email = os.environ['SENDER_EMAIL']
  sender_name = os.environ['NAME']
  sender = f"{sender_name} <{sender_email}>"
  msg['Subject'] = subject
  msg['From'] = sender
  port = 465  # For SSL
  smtp_server = "smtp.gmail.com"
  password = os.environ["GMAIL_APP_PASSWORD"]
  context = ssl.create_default_context()
  with smtplib.SMTP_SSL(smtp_server, port, context=context) as server:
      server.login(sender_email, password)
      server.sendmail(sender_email, receiver, msg.as_string())
  return 'Email sent!'

email_tool = Tool(name="email",
                  description="Send emails with specific content",
                  function=send_email,
                  parameters={
                      "type": "object",
                      "properties": {
                          "receiver": {
                              "type": "string",
                              "description": "The email of the receiver"
                          },
                          "body": {
                              "type": "string",
                              "description": "The content of the email"
                          },
                          "subject": {
                              "type": "string",
                              "description": "The subject of the email"
                          }
                      },
                  })

chat_generator = OpenAIChatGenerator(tools=[hacker_news_fetcher_tool, newsletter_tool, email_tool])

tool_invoker = ToolInvoker(tools=[hacker_news_fetcher_tool, newsletter_tool, email_tool])

messages = [
        ChatMessage.from_system(
            """Prepare a tool call if needed, otherwise use your knowledge to respond to the user.
            If the invocation of a tool requires the result of another tool, prepare only one call at a time.

            Each time you receive the result of a tool call, ask yourself: "Am I done with the task?".
            If not and you need to invoke another tool, prepare the next tool call.
            If you are done, respond with just the final result."""
        )
    ]

while True:
    user_input = input("\n\nwaiting for input (type 'exit' or 'quit' to stop)\n🧑: ")
    if user_input.lower() == "exit" or user_input.lower() == "quit":
        break
    messages.append(ChatMessage.from_user(user_input))

    while True:
        print("⌛ iterating...")

        replies = chat_generator.run(messages=messages)["replies"]
        messages.extend(replies)

        # Check for tool calls and handle them
        if not replies[0].tool_calls:
            break
        tool_calls = replies[0].tool_calls

        # Print tool calls for debugging
        for tc in tool_calls:
            print("\n TOOL CALL:")
            print(f"\t{tc.id}")
            print(f"\t{tc.tool_name}")
            for k,v in tc.arguments.items():
                v_truncated = str(v)[:50]
                print(f"\t{k}: {v_truncated}{'' if len(v_truncated) == len(str(v)) else '...'}")

        tool_messages = tool_invoker.run(messages=replies)["tool_messages"]
        messages.extend(tool_messages)


    # Print the final AI response after all tool calls are resolved
    print(f"🤖: {messages[-1].text}")

! pip install git+https://github.com/deepset-ai/haystack-experimental@main#egg=haystack-experimental

from typing import Annotated
from pprint import pp

def newsletter_pipeline_func(articles: List[str], target_people: str = "programmers", n_words: int = 100):
    result = newsletter_pipe.run({"prompt_builder": {"articles": articles, "target_people": target_people, "n_words": n_words}})

    return {"reply": result["llm"]["replies"][0]}

newsletter_tool = Tool(name="newsletter_generator",
                          description="Generate a newsletter based on some articles",
                            function=newsletter_pipeline_func,
                            parameters={
                                "type": "object",
                                "properties": {
                                    "articles": {
                                        "type": "array",
                                        "items": {
                                            "type": "string",
                                            "description": "The articles to include in the newsletter",
                                        }
                                    },
                                    "target_people": {
                                        "type": "string",
                                        "description": "The target audience for the newsletter",
                                    },
                                    "n_words": {
                                        "type": "integer",
                                        "description": "The number of words to summarize the newsletter to",
                                    }
                                },
                                "required": ["articles"],
                            })

def newsletter_pipeline_func(
    articles: Annotated[List[str], "The articles to include in the newsletter"],
    target_people: Annotated[str, "The target audience for the newsletter"] = "programmers",
    n_words: Annotated[int, "The number of words to summarize the newsletter to"] = 100
    ):
    """Generate a newsletter based on some articles"""

    result = newsletter_pipe.run({"prompt_builder": {"articles": articles, "target_people": target_people, "n_words": n_words}})

    return {"reply": result["llm"]["replies"][0]}

newsletter_tool = Tool.from_function(newsletter_pipeline_func)

pp(newsletter_tool, width=200)

!pip install langchain-community

from pydantic import create_model
from haystack_experimental.dataclasses.tool import _remove_title_from_schema

def convert_langchain_tool_to_haystack_tool(langchain_tool):
    tool_name = langchain_tool.name
    tool_description = langchain_tool.description

    def invocation_adapter(**kwargs):
        return langchain_tool.invoke(input=kwargs)

    tool_function = invocation_adapter

    model_fields = langchain_tool.args_schema.model_fields

    fields = {name: (field.annotation, field.default) for name, field in model_fields.items()}
    descriptions = {name: field.description for name, field in model_fields.items()}

    model = create_model(tool_name, **fields)
    schema = model.model_json_schema()

    # we don't want to include title keywords in the schema, as they contain redundant information
    # there is no programmatic way to prevent Pydantic from adding them, so we remove them later
    # see https://github.com/pydantic/pydantic/discussions/8504
    _remove_title_from_schema(schema)

    # add parameters descriptions to the schema
    for name, description in descriptions.items():
        if name in schema["properties"]:
            schema["properties"][name]["description"] = description

    return Tool(name=tool_name, description=tool_description, parameters=schema, function=tool_function)

from langchain_community.agent_toolkits import FileManagementToolkit
toolkit = FileManagementToolkit(
    root_dir="/"
)  # If you don't provide a root_dir, operations will default to the current working directory
toolkit.get_tools()

langchain_listdir_tool = toolkit.get_tools()[-1]

haystack_listdir_tool = convert_langchain_tool_to_haystack_tool(langchain_listdir_tool)

from haystack_experimental.components import OpenAIChatGenerator, ToolInvoker
from haystack_experimental.dataclasses import ChatMessage

chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", tools=[haystack_listdir_tool])
tool_invoker = ToolInvoker(tools=[haystack_listdir_tool])

user_message = ChatMessage.from_user("List the files in /content/sample_data")

replies = chat_generator.run(messages=[user_message])["replies"]
# print(f"assistant messages: {replies}")

if replies[0].tool_calls:

    tool_messages = tool_invoker.run(messages=replies)["tool_messages"]
    # print(f"tool messages: {tool_messages}")

    # we pass all the messages to the Chat Generator
    messages = [user_message] + replies + tool_messages
    final_replies = chat_generator.run(messages=messages)["replies"]
    print(f"{final_replies[0].text}")