使用工具发送新闻通讯的 Agent

! 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 import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.components.tools import ToolInvoker
from haystack.tools import Tool

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 = [ChatMessage.from_user("""
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", ChatPromptBuilder(template=template))
newsletter_pipe.add_component("llm", OpenAIChatGenerator(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].text}

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}")

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].text}

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"],
                            })

from haystack.tools import create_tool_from_function
 
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].text}

newsletter_tool = create_tool_from_function(newsletter_pipeline_func)

pp(newsletter_tool, width=200)

! pip install langchain-community

from pydantic import create_model
from haystack.tools.from_function 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.components.generators.chat import OpenAIChatGenerator
from haystack.components.tools import ToolInvoker
from haystack.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}")