Spaces:
Runtime error
Runtime error
| from smolagents import Tool, tool | |
| from langchain_community.tools.tavily_search import TavilySearchResults | |
| import requests | |
| import inspect | |
| import pandas as pd | |
| from PIL import Image | |
| from io import BytesIO | |
| import base64 | |
| from langchain_core.documents import Document | |
| from langchain_community.retrievers import BM25Retriever | |
| from src.final_assignment_template.models import videoLiteLLm,modelLiteLLm, summarizeModle, imageLiteLLm | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| travily_tool = Tool.from_langchain(TavilySearchResults(max_results=20)) | |
| from smolagents import Tool | |
| # class SearchTool(Tool): | |
| # name = "SearchTool" | |
| # description = """ | |
| # This is a tool that returns the most downloaded model of a given task on the Hugging Face Hub. | |
| # It returns the name of the checkpoint.""" | |
| # inputs = { | |
| # "task": { | |
| # "type": "string", | |
| # "description": "the task category (such as text-classification, depth-estimation, etc)", | |
| # } | |
| # } | |
| # output_type = "string" | |
| # def forward(self, task: str): | |
| # from huggingface_hub import list_models | |
| # model = next(iter(list_models(filter=task, sort="downloads", direction=-1))) | |
| # return model.id | |
| # model_downloads_tool = HFModelDownloadsTool() | |
| from langchain_core.documents import Document | |
| from langchain_community.retrievers import BM25Retriever | |
| def bm25_query(texts: list[str], query: str, top_k: int = 3) -> list[str]: | |
| """ | |
| Creates a BM25 retriever from a list of texts (e.g., web pages, articles), | |
| queries it, and returns the top relevant results. | |
| Args: | |
| texts (list[str]): List of text contents (e.g., web page texts, articles, notes). | |
| query (str): The search query string. | |
| top_k (int): Number of top results to return (default is 3). | |
| Returns: | |
| list[str]: List of top-k relevant page contents. | |
| """ | |
| documents = [Document(page_content=text) for text in texts] | |
| retriever = BM25Retriever.from_documents(documents) | |
| results = retriever.get_relevant_documents(query) | |
| print(results) | |
| return [doc.page_content for doc in results[:top_k]] | |
| class BM25Tool(Tool): | |
| name = "bm25" | |
| description = ( | |
| "Retrieves relevant information from a provided list of text strings " | |
| "based on a query using BM25." | |
| ) | |
| inputs = { | |
| "query": { | |
| "type": "string", | |
| "description": "The text query to search for relevant strings." | |
| } | |
| } | |
| output_type = "string" | |
| def __init__(self, texts: list[str]): | |
| """ | |
| Args: | |
| texts (list[str]): A list of text strings to index (e.g., guest bios, docs, notes). | |
| """ | |
| documents = [Document(page_content=text) for text in texts] | |
| self.retriever = BM25Retriever.from_documents(documents) | |
| def forward(self, query: str) -> str: | |
| """ | |
| Retrieves the top-3 most relevant strings matching the query. | |
| Args: | |
| query (str): Text query. | |
| Returns: | |
| str: Concatenated top-3 matching strings or a not-found message. | |
| """ | |
| results = self.retriever.get_relevant_documents(query) | |
| if not results: | |
| return "No relevant information found." | |
| top_texts = [doc.page_content for doc in results[:3]] | |
| return "\n\n".join(top_texts) | |
| def summarize_before_final_answer( | |
| context: str, | |
| question: str, | |
| ) -> str: | |
| """ | |
| Given a whole context(all logs) and question sends it to the LLM, and returns the paragraph overview for the answer. | |
| Args: | |
| context (str): The full context or background information. | |
| question (str): The user's specific question about that context. | |
| Returns: | |
| str: Summarization of whole process for generating final answer. | |
| """ | |
| # build a single user prompt | |
| prompt = ( | |
| context.strip() | |
| + "\n\n" | |
| + "Question: " | |
| + question.strip() | |
| + "\n\n" | |
| + "Give the summarize of all steps for generating final answer in next step:" | |
| ) | |
| # call the model | |
| response = summarizeModle( | |
| messages=[{"role": "user", "content": prompt}], | |
| ) | |
| # the .content attribute holds the generated text | |
| return response.content.strip() | |
| def Video_link_understanding_tool(query: str) -> str: | |
| """ | |
| A tool that processes a video link (e.g., YouTube) and returns a textual understanding of its content using an LLM. | |
| Args: | |
| query: A video URL along with an optional query for context or specific focus. | |
| Returns: | |
| A text-based summary or understanding of the video content. | |
| """ | |
| print("Processing video:", query) | |
| messages = [{"role": "user", "content": [{"type": "text", "text": query}]}] | |
| resp = videoLiteLLm(messages) | |
| return resp.content or 'No data' | |
| def get_task_file(task_id:str)->requests.models.Response: | |
| """ | |
| This tool is for get the task file using task_id. | |
| it will return the request response and then this response will be used for other tools. | |
| Args: | |
| task_id: Task ID | |
| """ | |
| url = f"{DEFAULT_API_URL}/files/{task_id}" | |
| print(url) | |
| headers = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36" | |
| } | |
| response = requests.get(url,headers=headers) | |
| return response | |
| def image_understanding_tool(query: str, response: requests.models.Response) -> str: | |
| """ | |
| A tool for analyzing and understanding the content of an image based on a given query. | |
| This tool processes the image provided in the response (from get_task_file), encodes it into base64, | |
| and queries a lightweight image LLM to generate insights or answers about the image. | |
| Args: | |
| query: The query or instruction related to the image content. | |
| response: The HTTP response object containing the image data. | |
| Returns: | |
| A text-based understanding or interpretation of the image. | |
| """ | |
| print("Processing image...") | |
| image = Image.open(BytesIO(response.content)).convert("RGB") | |
| buffered = BytesIO() | |
| image.save(buffered, format="PNG") | |
| img_bytes = buffered.getvalue() | |
| img_b64 = base64.b64encode(img_bytes).decode('utf-8') | |
| # print(img_b64) | |
| messages = [{ | |
| "role": "user", | |
| "content": [ | |
| {"type": "text", "text": query}, | |
| { | |
| "type": "image_url", | |
| "image_url": { | |
| "url": img_b64, | |
| "format": "image/png" | |
| } | |
| } | |
| ] | |
| }] | |
| resp = imageLiteLLm(messages) | |
| print(resp.content) | |
| return resp.content or 'No data' | |
| def extract_filter_textual_info_from_textual_context( | |
| context: str, | |
| question: str, | |
| ) -> str: | |
| """ | |
| Tool to pull out targeted details from a large body of text. | |
| Combines the context and an questoin into a single prompt, | |
| queries the llm, and returns the resulting extract. | |
| Args: | |
| context (str): The full background text (e.g., long document, webpage, notes). | |
| question (str): What you want to extract (e.g., “list all dates mentioned”). | |
| Returns: | |
| str: The extracted information, trimmed of whitespace. | |
| """ | |
| # Build the extraction prompt | |
| prompt = ( | |
| "Context:\n" + context.strip() + | |
| "\n\nQuestion: " + question.strip() + | |
| "\n\nExtracted Information:" | |
| ) | |
| # Call the model to perform extraction | |
| response = modelLiteLLm( | |
| messages=[{"role": "user", "content": prompt}], | |
| ) | |
| print(response) | |
| return response.content | |