Spaces:
Sleeping
Sleeping
"""Text processing functions""" | |
from typing import Dict, Generator, Optional | |
from selenium.webdriver.remote.webdriver import WebDriver | |
from autogpt.config import Config | |
from autogpt.llm_utils import create_chat_completion | |
from autogpt.memory import get_memory | |
CFG = Config() | |
MEMORY = get_memory(CFG) | |
def split_text(text: str, max_length: int = 8192) -> Generator[str, None, None]: | |
"""Split text into chunks of a maximum length | |
Args: | |
text (str): The text to split | |
max_length (int, optional): The maximum length of each chunk. Defaults to 8192. | |
Yields: | |
str: The next chunk of text | |
Raises: | |
ValueError: If the text is longer than the maximum length | |
""" | |
paragraphs = text.split("\n") | |
current_length = 0 | |
current_chunk = [] | |
for paragraph in paragraphs: | |
if current_length + len(paragraph) + 1 <= max_length: | |
current_chunk.append(paragraph) | |
current_length += len(paragraph) + 1 | |
else: | |
yield "\n".join(current_chunk) | |
current_chunk = [paragraph] | |
current_length = len(paragraph) + 1 | |
if current_chunk: | |
yield "\n".join(current_chunk) | |
def summarize_text( | |
url: str, text: str, question: str, driver: Optional[WebDriver] = None | |
) -> str: | |
"""Summarize text using the OpenAI API | |
Args: | |
url (str): The url of the text | |
text (str): The text to summarize | |
question (str): The question to ask the model | |
driver (WebDriver): The webdriver to use to scroll the page | |
Returns: | |
str: The summary of the text | |
""" | |
if not text: | |
return "Error: No text to summarize" | |
text_length = len(text) | |
print(f"Text length: {text_length} characters") | |
summaries = [] | |
chunks = list(split_text(text)) | |
scroll_ratio = 1 / len(chunks) | |
for i, chunk in enumerate(chunks): | |
if driver: | |
scroll_to_percentage(driver, scroll_ratio * i) | |
print(f"Adding chunk {i + 1} / {len(chunks)} to memory") | |
memory_to_add = f"Source: {url}\n" f"Raw content part#{i + 1}: {chunk}" | |
MEMORY.add(memory_to_add) | |
print(f"Summarizing chunk {i + 1} / {len(chunks)}") | |
messages = [create_message(chunk, question)] | |
summary = create_chat_completion( | |
model=CFG.fast_llm_model, | |
messages=messages, | |
) | |
summaries.append(summary) | |
print(f"Added chunk {i + 1} summary to memory") | |
memory_to_add = f"Source: {url}\n" f"Content summary part#{i + 1}: {summary}" | |
MEMORY.add(memory_to_add) | |
print(f"Summarized {len(chunks)} chunks.") | |
combined_summary = "\n".join(summaries) | |
messages = [create_message(combined_summary, question)] | |
return create_chat_completion( | |
model=CFG.fast_llm_model, | |
messages=messages, | |
) | |
def scroll_to_percentage(driver: WebDriver, ratio: float) -> None: | |
"""Scroll to a percentage of the page | |
Args: | |
driver (WebDriver): The webdriver to use | |
ratio (float): The percentage to scroll to | |
Raises: | |
ValueError: If the ratio is not between 0 and 1 | |
""" | |
if ratio < 0 or ratio > 1: | |
raise ValueError("Percentage should be between 0 and 1") | |
driver.execute_script(f"window.scrollTo(0, document.body.scrollHeight * {ratio});") | |
def create_message(chunk: str, question: str) -> Dict[str, str]: | |
"""Create a message for the chat completion | |
Args: | |
chunk (str): The chunk of text to summarize | |
question (str): The question to answer | |
Returns: | |
Dict[str, str]: The message to send to the chat completion | |
""" | |
return { | |
"role": "user", | |
"content": f'"""{chunk}""" Using the above text, answer the following' | |
f' question: "{question}" -- if the question cannot be answered using the text,' | |
" summarize the text.", | |
} | |