tav.py•2.95 kB
import os
from tavily import TavilyClient
from typing import Annotated
from config import TAVILY_API_KEY, OPENAI_API_KEY
# from langchain.adapters.openai import convert_openai_messages
# New
from langchain_openai import ChatOpenAI
# from apiclient.errors import HttpError
from googleapiclient.errors import HttpError
from .web_scrap import extract_article
# from apiclient import errors
# from langchain.adapters.openai import convert_openai_messages
import warnings
# warnings.filterwarnings("ignore", category=DeprecationWarning)
from langchain_community.adapters.openai import convert_openai_messages
warnings.simplefilter("ignore")
os.environ["LANGCHAIN_IMPORT_WARNING_DISABLED"] = "1"
client = TavilyClient(api_key=TAVILY_API_KEY)
def search_tool(
query: Annotated[str, "The search query"]
) -> Annotated[str, "The search results"]:
try:
# print("query = ",query)
content = client.search(
query=query,
search_depth="advanced",
use_cache=True,
max_results=2,
)
# print("content = ",content)
content = content["results"]
content = get_content(content)
prompt = [
{
"role": "system",
"content": f"You are an AI critical thinker research assistant. "
f"Your sole purpose is to write well written, critically acclaimed,"
f"It must include the thoughts of the industry leaders.It is a mandatory in the report"
f"objective and structured reports on given text."
f"The report should contain the economic and market growth impact."
f"The report should be detailed and it should mention the source of the information."
f"The report should be extremely informative."
f"Always mention the examples related to the facts.",
},
{
"role": "user",
"content": f'Information: """{content}"""\n\n'
f"Using the above information, answer the following"
f'query: "{query}" in a detailed report do not include conclusion and references and summary --',
},
]
lc_messages = convert_openai_messages(prompt)
result = str(
ChatOpenAI(model="gpt-4o", openai_api_key=OPENAI_API_KEY)
.invoke(lc_messages)
.content
)
return result.encode("utf-8")
except HttpError as e:
print(f"An HTTP error {e.resp.status} occurred:\n{e.content}")
def get_content(results):
data = []
for result in results:
title = result.get("title")
url = result.get("url")
if title and url:
content = extract_article(url)
data.append({"title": title, "content": content})
return data