| from langchain.tools import tool |
| |
| from langchain_community.tools import TavilySearchResults |
| from langchain_community.document_loaders import WebBaseLoader |
| from langchain_community.document_loaders import YoutubeLoader |
| from langchain_community.document_loaders import WikipediaLoader |
| from langchain_community.document_loaders import ArxivLoader |
| import requests |
| import os |
| from langchain_core.messages import HumanMessage |
| from langchain_openai import ChatOpenAI |
|
|
| @tool |
| def add(a: float, b: float) -> float: |
| """Add two integers and return the result.""" |
| return a + b |
|
|
| @tool |
| def subtract(a: float, b: float) -> float: |
| """Subtract two integers and return the result.""" |
| return a - b |
|
|
| @tool |
| def multiply(a: float, b: float) -> float: |
| """Multiply two integers and return the result.""" |
| return a * b |
|
|
| @tool |
| def divide(a: float, b: float) -> float: |
| """Divide two integers and return the result.""" |
| if b == 0: |
| raise ValueError("Cannot divide by zero.") |
| return a / b |
|
|
| @tool |
| def exponentiate(base: float, exponent: float) -> float: |
| """Raise a number to the power of another number and return the result.""" |
| return base ** exponent |
|
|
| @tool |
| def modulus(a: float, b: float) -> float: |
| """Return the modulus of two integers.""" |
| return a % b |
|
|
| @tool |
| def wiki_search(query: str) -> str: |
| """Search Wikipedia and returns only 2 results. |
| |
| Args: |
| query: The search query.""" |
| docs = WikipediaLoader(query=query, load_max_docs=2).load() |
| res = "\n#######\n".join( |
| [ |
| f"Document {i+1}:\nSource: {doc.metadata.get('source', '')}\nPage: {doc.metadata.get('page', '')}\nContent:\n{doc.page_content}\n" |
| for i, doc in enumerate(docs) |
| ]) |
| print(f"load wiki page : {res}") |
| return {"results": res} |
|
|
| @tool |
| def load_web_page(url: str) -> str: |
| """Load a web page and return its content. |
| |
| Args: |
| url: The URL of the web page to load. |
| """ |
| loader = WebBaseLoader(url) |
| docs = loader.load() |
| res = "\n#######\n".join( |
| [ |
| f"Document {i+1}:\nSource: {doc.metadata.get('source', '')}\nPage: {doc.metadata.get('page', '')}\nContent:\n{doc.page_content}\n" |
| for i, doc in enumerate(docs) |
| ]) |
| print(f"load web page : {res}") |
| return {"results": res} |
| |
| @tool |
| def paper_search(query: str) -> str: |
| """Search Arxiv for a query and return maximum 3 result. |
| |
| Args: |
| query: The search query.""" |
| docs = ArxivLoader(query=query, load_max_docs=3).load() |
| res = "\n#######\n".join( |
| [ |
| f"Document {i+1}:\nSource: {doc.metadata.get('source', '')}\nPage: {doc.metadata.get('page', '')}\nContent:\n{doc.page_content}\n" |
| for i, doc in enumerate(docs) |
| ]) |
| print(f"load paper page : {res}") |
| return {"results": res} |
|
|
| @tool |
| def understand_image(text: str, image_url: str): |
| """ |
| Sends a text prompt and an image URL to OpenAI's API using the ChatOpenAI model. |
| Returns the model's response. |
| |
| Args: |
| text (str): The text prompt to send. |
| image_url (str): URL to the image to send. |
| |
| Returns: |
| str: The response from the model. |
| """ |
|
|
| |
| |
| |
| |
|
|
| |
| message = HumanMessage( |
| content=[ |
| {"type": "text", "text": text}, |
| |
| {"type": "image_url", "image_url": {"url": image_url}} |
| ] |
| ) |
| model = ChatOpenAI(model="gpt-4o", temperature=0) |
| response = model.invoke([message]) |
| return response.content |
|
|
| @tool |
| def load_youtube_video(url: str) -> str: |
| """Load a YouTube video and return its content.""" |
| loader = YoutubeLoader.from_youtube_url(url, add_video_info=True) |
| documents = loader.load() |
| return documents[0].page_content if documents else "No content found" |
|
|
| @tool |
| def web_search(query: str) -> str: |
| """Search Tavily for a query and return maximum 5 results. |
| |
| Args: |
| query: The search query.""" |
| documents = TavilySearchResults(max_results=5).invoke(input=query) |
| res = "\n#######\n".join( |
| [ |
| f"Document {i+1}:\nContent: {doc['content']}\n" |
| for i, doc in enumerate(documents) |
| ]) |
| print(f"load tavily search : {res}") |
| return {"results": res} |
|
|
| @tool |
| def transcribe_audio(audio_url: str) -> str: |
| """Transcribe audio from a URL and return the text. |
| |
| Args: |
| audio_url: The URL of the audio file to transcribe. |
| """ |
| |
| response = requests.get(audio_url) |
| audio_file = "audio.mp3" |
| with open(audio_file, "wb") as f: |
| f.write(response.content) |
|
|
| |
| api_key = os.environ.get("OPENAI_API_KEY") |
| headers = { |
| "Authorization": f"Bearer {api_key}" |
| } |
| files = { |
| 'file': (audio_file, open(audio_file, 'rb')), |
| 'model': (None, 'whisper-1') |
| } |
|
|
| transcribe_response = requests.post( |
| "https://api.openai.com/v1/audio/transcriptions", |
| headers=headers, |
| files=files |
| ) |
| print(f"Transcription response: {transcribe_response.json()}") |
| return {"results": transcribe_response.json().get("text", "Transcription failed.")} |
