Bitsak/AutoGPT2

1"""Text processing functions"""2from typing import Dict, Generator, Optional3 4from selenium.webdriver.remote.webdriver import WebDriver5 6from autogpt.config import Config7from autogpt.llm_utils import create_chat_completion8from autogpt.memory import get_memory9 10CFG = Config()11MEMORY = get_memory(CFG)12 13 14def split_text(text: str, max_length: int = 8192) -> Generator[str, None, None]:15    """Split text into chunks of a maximum length16 17    Args:18        text (str): The text to split19        max_length (int, optional): The maximum length of each chunk. Defaults to 8192.20 21    Yields:22        str: The next chunk of text23 24    Raises:25        ValueError: If the text is longer than the maximum length26    """27    paragraphs = text.split("\n")28    current_length = 029    current_chunk = []30 31    for paragraph in paragraphs:32        if current_length + len(paragraph) + 1 <= max_length:33            current_chunk.append(paragraph)34            current_length += len(paragraph) + 135        else:36            yield "\n".join(current_chunk)37            current_chunk = [paragraph]38            current_length = len(paragraph) + 139 40    if current_chunk:41        yield "\n".join(current_chunk)42 43 44def summarize_text(45    url: str, text: str, question: str, driver: Optional[WebDriver] = None46) -> str:47    """Summarize text using the OpenAI API48 49    Args:50        url (str): The url of the text51        text (str): The text to summarize52        question (str): The question to ask the model53        driver (WebDriver): The webdriver to use to scroll the page54 55    Returns:56        str: The summary of the text57    """58    if not text:59        return "Error: No text to summarize"60 61    text_length = len(text)62    print(f"Text length: {text_length} characters")63 64    summaries = []65    chunks = list(split_text(text))66    scroll_ratio = 1 / len(chunks)67 68    for i, chunk in enumerate(chunks):69        if driver:70            scroll_to_percentage(driver, scroll_ratio * i)71        print(f"Adding chunk {i + 1} / {len(chunks)} to memory")72 73        memory_to_add = f"Source: {url}\n" f"Raw content part#{i + 1}: {chunk}"74 75        MEMORY.add(memory_to_add)76 77        print(f"Summarizing chunk {i + 1} / {len(chunks)}")78        messages = [create_message(chunk, question)]79 80        summary = create_chat_completion(81            model=CFG.fast_llm_model,82            messages=messages,83        )84        summaries.append(summary)85        print(f"Added chunk {i + 1} summary to memory")86 87        memory_to_add = f"Source: {url}\n" f"Content summary part#{i + 1}: {summary}"88 89        MEMORY.add(memory_to_add)90 91    print(f"Summarized {len(chunks)} chunks.")92 93    combined_summary = "\n".join(summaries)94    messages = [create_message(combined_summary, question)]95 96    return create_chat_completion(97        model=CFG.fast_llm_model,98        messages=messages,99    )100 101 102def scroll_to_percentage(driver: WebDriver, ratio: float) -> None:103    """Scroll to a percentage of the page104 105    Args:106        driver (WebDriver): The webdriver to use107        ratio (float): The percentage to scroll to108 109    Raises:110        ValueError: If the ratio is not between 0 and 1111    """112    if ratio < 0 or ratio > 1:113        raise ValueError("Percentage should be between 0 and 1")114    driver.execute_script(f"window.scrollTo(0, document.body.scrollHeight * {ratio});")115 116 117def create_message(chunk: str, question: str) -> Dict[str, str]:118    """Create a message for the chat completion119 120    Args:121        chunk (str): The chunk of text to summarize122        question (str): The question to answer123 124    Returns:125        Dict[str, str]: The message to send to the chat completion126    """127    return {128        "role": "user",129        "content": f'"""{chunk}""" Using the above text, answer the following'130        f' question: "{question}" -- if the question cannot be answered using the text,'131        " summarize the text.",132    }133