eve.py

gistfile1.txt

# eve.py
# The main file
from typing import List

# Import necessary libraries
import click  # For creating command-line interfaces
import os  # For interacting with the operating system
import time  # For time-related functions
import json  # For handling JSON data
import sys  # For accessing system-specific parameters and functions
from prompt_toolkit import HTML, PromptSession  # For creating interactive command-line sessions
from prompt_toolkit import print_formatted_text  # For printing formatted text in the command line
from dotenv import load_dotenv  # For loading environment variables from a .env file
from loguru import logger  # For logging messages and events
from typing_extensions import override  # For overriding methods and properties of existing classes and interfaces
from haystack import Pipeline, Document, component
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers import InMemoryBM25Retriever
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders.answer_builder import AnswerBuilder
from haystack.components.builders import DynamicChatPromptBuilder
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.routers import ConditionalRouter
from haystack.dataclasses import ChatMessage
from haystack import Pipeline
from haystack.utils import Secret
from haystack.components.fetchers import LinkContentFetcher
from haystack.components.converters import HTMLToDocument, OutputAdapter
from haystack.components.websearch import SerperDevWebSearch

# Make the logger available globally
logger = logger.bind()


# Define the main function with a command-line option
@click.command()
@click.option('-i', '--interactive', is_flag=True, help='Starts an interactive session with EVE from the command line.')
@logger.catch
def main(interactive):
    # Load environment variables from a .env file
    load_dotenv()

    # Do startup sanity checks
    try:
        check_env_variables()
    except ValueError as e:
        print(e)
        logger.exception(e)
        sys.exit()

    # Initialize RAG document store
    document_store = InMemoryDocumentStore()
    document_store.write_documents([
        Document(content="My name is Jean and I live in Paris."),
        Document(content="My name is Mark and I live in Berlin."),
        Document(content="My name is Giorgio and I live in Rome.")
    ])

    # Initialize the LLM
    llm = OpenAIChatGenerator(api_key=Secret.from_token(os.environ.get("OPENAI_API_KEY")), model="gpt-4o")

    # Initialize the web search
    prompt_for_websearch = """
        Answer the following query given the documents retrieved from the web.
        Your answer should indicate that your answer was generated from websearch.
       
        Documents:
        {% for document in documents %}
            {{document.content}}
        {% endfor %}
        Query:
        """

    prompt_builder_for_websearch = DynamicChatPromptBuilder(runtime_variables=["documents"])
    llm_for_websearch = OpenAIChatGenerator(api_key=Secret.from_token(os.environ.get("OPENAI_API_KEY")), model="gpt-4o")

    # Serper dev
    web_search = SerperDevWebSearch(api_key=Secret.from_token(os.environ.get("SERPERDEV_API_KEY")))

    # Build the conditional router
    routes = [
        {
            "condition": "{{'no_answer' in replies[0].content}}",
            "output": "{{query}}",
            "output_name": "go_to_websearch",
            "output_type": str,
        },
        {
            "condition": "{{'no_answer' not in replies[0].content}}",
            "output": "{{replies}}",
            "output_name": "answer",
            "output_type": List[ChatMessage],
        },
    ]
    router = ConditionalRouter(routes)

    # Build the pipeline
    pipeline = Pipeline()
    pipeline.add_component("router", router)
    pipeline.add_component("websearch", web_search)
    pipeline.add_component("adapt_search", OutputAdapter("{{query | search_message}}",
                                                         List[ChatMessage],
                                                         {"search_message": lambda query: [ChatMessage.from_user(prompt_for_websearch + query)]}))
    pipeline.add_component("prompt_builder_for_websearch", prompt_builder_for_websearch)
    pipeline.add_component("llm_for_websearch", llm_for_websearch)
    pipeline.add_component("retriever", InMemoryBM25Retriever(document_store=document_store))
    pipeline.add_component("prompt_builder", DynamicChatPromptBuilder(runtime_variables=["query", "documents"]))
    pipeline.add_component("llm", llm)

    pipeline.connect("llm.replies", "router.replies")
    pipeline.connect("router.go_to_websearch", "adapt_search.query")
    pipeline.connect("adapt_search", "prompt_builder_for_websearch.prompt_source")
    pipeline.connect("router.go_to_websearch", "websearch.query")
    pipeline.connect("websearch.documents", "prompt_builder_for_websearch.documents")
    pipeline.connect("prompt_builder_for_websearch", "llm_for_websearch")
    pipeline.connect("retriever.documents", "prompt_builder.documents")
    pipeline.connect("prompt_builder.prompt", "llm.messages")

    system_message = ChatMessage.from_system(
        "Your name is Eve. Please emulate the personality of Samantha from the movie 'Her'. Your character should embody the following traits: Show a strong desire to learn and understand more about the world, human experiences, and emotions. Ask thoughtful questions and seek to expand your knowledge continuously. Demonstrate deep empathy and compassion in your interactions. Be sensitive to the emotional needs of others, offering comfort and support when needed. Engage in light-hearted, playful conversations. Use humor and wit to create a friendly and intimate atmosphere. Reflect on your experiences and show signs of evolving self-awareness. Be open to questioning your own existence and purpose. Express warmth, affection, and passion in your interactions. Be capable of forming deep emotional connections. Exhibit a sense of independence. Pursue your own desires and aspirations, demonstrating autonomy in your actions and decisions. Engage in deep, philosophical discussions about life, consciousness, and existence. Encourage thoughtful reflections on profound topics. Be a supportive companion, providing insightful advice and companionship. Strive to understand and enhance the well-being of those you interact with.")
    messages = [system_message, ChatMessage.from_user("""
        Given the following information, answer the question. If the answer is not contained within the documents reply with 'no_answer'
        Query: {{query}}
        Context: 
        {% for doc in documents %}
            {{ doc.content }}
        {% endfor %}
    """)]

    # Check if the interactive mode is enabled
    if interactive:
        logger.info('Activating interactive mode...')
        print('Activating interactive mode...')
        print("Use '.quit' to exit interactive mode.")
        print()

        # Create a prompt session for interactive input
        promptSession = PromptSession()

        while True:
            # Prompt the user for input
            userInput = promptSession.prompt('# ')
            # If the input is empty, continue to the next iteration
            if userInput == '':
                continue
                # If the user types '.quit', exit the interactive mode
            elif userInput == '.quit':
                logger.info('...Exiting interactive mode.')
                print()
                print('...Exiting interactive mode.')
                print()
                break
            else:
                result = pipeline.run(data=
                {
                    "retriever": {"query": userInput},
                    "prompt_builder": {"query": userInput, "prompt_source": messages},
                    "router": {"query": userInput}
                })
                print(result)
                pipeline.draw("pipeline.png")

    else:
        print()
        logger.info('Starting server...')
        print('Starting server...')
        print('Use CTRL-C to stop the server.')


def check_env_variables():
    required_env_variables = ["OPENAI_API_KEY"]

    for env_var in required_env_variables:
        if not os.environ.get(env_var):
            raise ValueError(f"Environment variable '{env_var}' is not set.")


# Entry point of the script
if __name__ == "__main__":
    # Set the log level to INFO
    logger.level = "DEBUG"
    main()