tools/new_tools.py

from langchain_core.messages import HumanMessage
from langchain_core.tools import tool, Tool
from langchain_together import ChatTogether
from langgraph.prebuilt import create_react_agent
from langchain_community.retrievers import WikipediaRetriever
from langchain_community.tools import BraveSearch
from langchain_experimental.utilities import PythonREPL
from langchain_community.agent_toolkits.load_tools import load_tools
import requests
from langgraph_supervisor import create_supervisor
from youtube_transcript_api import YouTubeTranscriptApi
from pytubefix import extract, YouTube
import whisper
from qwen_vl_utils import process_vision_info
from transformers import Qwen2_5_VLForConditionalGeneration, AutoTokenizer, AutoProcessor

@tool
def wiki_search(query: str) -> str:
    """Search Wikipedia for query and return maximum 3 results
    
    Args:
        query (str): query to search on Wikipedia
    Returns:
        wiki_result (str): result of search
    """
    try:
        retriever = WikipediaRetriever()
        wiki_result = retriever.invoke(query)
        return wiki_result
    except Exception as e:
        return f"wiki_search failed {e}"

@tool
def query_image(query: str, image_url: str):
    """Analyze the query on an image using a VLM
    
    Args:
        query (str): query about the image
        image_url (str): link to the image
    Returns:
        response (str): response to the query on image
    """
    try:
        model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
            "Qwen/Qwen2.5-VL-3B-Instruct", torch_dtype="auto", device_map="auto"
        )

        processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-3B-Instruct")

        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "image": image_url,
                        "max_pixels": 360 * 420,
                    },
                    {"type": "text", "text": query},
                ],
            }
        ]

        text = processor.apply_chat_template(
            messages, tokenize=False, add_generation_prompt=True
        )
        image_inputs, video_inputs = process_vision_info(messages)
        inputs = processor(
            text=[text],
            images=image_inputs,
            videos=video_inputs,
            padding=True,
            return_tensors="pt",
        )
        inputs = inputs.to("cuda")

        # Inference
        generated_ids = model.generate(**inputs, max_new_tokens=128)
        generated_ids_trimmed = [
            out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
        ]
        output_text = processor.batch_decode(
            generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
        )
        print(output_text)
        return output_text

    except Exception as e:
        return f"query_image failed {e}"

@tool
def reverse_string(input_string: str) -> str:
    """Reverse the character order of input string.
    
    Args:
        input_string (str): string to reverse
    Returns:
        reversed_string (str): reversed string
    """
    try:
        reversed_string = input_string[::-1]
        reversed_string = f"The reversed string returned from reverse_string function is: {reversed_string}"
        return reversed_string
    except Exception as e:
        return f"reverse_string failed {e}"

repl = PythonREPL()
python_repl_tool = Tool(
    name="python_repl",
    description="""A Python shell. Use this to execute python commands. 
        Input should be a valid python command. 
        Input should be a valid Python expression or script.
        If you want to see the output of a value, you should print it out with `print(...)`.
        Always return the printed code output.
        Example: print(2 + 2) → will return 4
        Do NOT execute code that could be harmful to the host system.
        You are allowed to download files from URLs.""",
    func=repl.run
)


class langgraph_agent:
    def __init__(self):
        llm = ChatTogether(
            model="Qwen/Qwen3-235B-A22B-fp8-tput",
            temperature=0
        )

        helper_llm = ChatTogether(
            model="meta-llama/Llama-3.3-70B-Instruct-Turbo-Free",
            temperature=0
        )

        research_agent = create_react_agent(
            llm,
            tools=research_tools,
            name="research_agent",
            prompt=(
                    "You are a research agent. You have access to web_search tool to search the web, wiki_search tool to search wikipedia\n\n"
                    "INSTRUCTIONS:\n"
                    "- Assist ONLY with research tasks\n"
                    "- After you're done with your tasks, respond to the supervisor directly\n"
                    "- Respond ONLY with the results of your work, do NOT include ANY other text."
            ),
        )

        vision_agent = create_react_agent(
            helper_llm,
            tools=vision_tools,
            name="vision_agent",
            prompt=(
                    "You are a vision agent. You have access to the following tools: \n"
                    "   query_image(query: str, image_url: str): \n"
                    "       Args:\n"
                    "           query (str): query on the image \n"
                    "           image_url (str): link to the image \n"
                    "       Returns:\n"
                    "           response (str): response to the query after analyzing image \n\n"
                    "   query_video(query: str, video_url: str): \n"
                    "       Args:\n"
                    "           query (str): query on the video\n"
                    "           video_url (str): link to the video \n"
                    "       Returns: \n"
                    "           response (str): response to the query after analyzing video \n\n"
                    "INSTRUCTIONS:\n"
                    "- Assist ONLY with vision related tasks\n"
                    "- After you're done with your tasks, respond to the supervisor directly\n"
                    "- Respond ONLY with the results of your work, do NOT include ANY other text."        
            ),
        )

        python_agent = create_react_agent(
            helper_llm,
            tools=[python_repl_tool],
            name="python_agent",
            prompt=(
                    "You are a python coding agent with access to a python REPL. You will be given a query and a link to a piece of python code. Retrieve and execute the linked code with python_repl tool to answer the query. \n\n"
                    "INSTRUCTIONS:\n"
                    "- Assist ONLY with python coding tasks\n"
                    "- You are allowed to download files from given URLs \n"
                    "- Do not execute code that can be harmful to host system \n"
                    "- If there is Exception thrown during execution, try to debug your code, then execute again. \n"
                    "- Always transfer any printed output from executed code to supervisor \n"
                    "- After you're done with your tasks, respond to the supervisor directly\n"
                    "- Respond ONLY with the results of your work, do NOT include ANY other text."
            ),       
        )