File size: 5,532 Bytes
3823795 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 |
from langchain.tools import tool
#from langchain_community.tools import DuckDuckGoSearchRun
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.
"""
# Fetch image from URL and encode as base64
#response = requests.get(image_url)
#image_bytes = response.content
#image_b64 = base64.b64encode(image_bytes).decode("utf-8")
# Prepare message with text and image
message = HumanMessage(
content=[
{"type": "text", "text": text},
#{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_b64}", "detail": "auto"}}
{"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)
# Step 2: Send it to OpenAI's transcription API
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.")} |