import gradio as gr
import warnings
# 抑制所有警告
warnings.filterwarnings('ignore')
import os
import sys
#sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from Smurfs.inference.smurfs_worker import smurfs_hotpot_worker, smurfs_worker, stream_smurfs_worker
# from Smurfs.tools.tool_env import HotpotToolEnv
from Smurfs.deploy import global_dict
from Smurfs.model.openai_model.openai_model import OpenAI_Model
from Smurfs.agents.answer_agent.answer import stream_answer_agent
from Smurfs.agents.executor_agent.executor import stream_executor_agent
from Smurfs.agents.planning_agent.planner import stream_hotpot_planning_agent
from Smurfs.agents.verifier_agent.verifier import stream_verifier_agent
from Smurfs.tools.docqa.api import tool_env as docqa_tool_env
from Smurfs.tools.hotpotQA.api import tool_env as hotpot_tool_env
from Smurfs.tools.math.api import tool_env as math_tool_env
from Smurfs.tools.shell.api import tool_env as shell_tool_env
from Smurfs.tools.websearch.api import tool_env as websearch_tool_env
import json
import threading
import joblib
from tqdm import tqdm
import time
from PyPDF2 import PdfReader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from datetime import datetime
current_datetime = datetime.now()
# user_input = gr.Textbox(show_label=False, placeholder="Input...", lines=10).style(container=False)
# inp = gr.Textbox(placeholder="Enter your task")
# css = """
# .btn {background-color: blue; color: white;}
# #bot {background-color: blue; color: white;}
# #e {display: inline-block; vertical-align: middle;}
# """
# def update(name):
# return f"Welcome to Gradio, {name}!"
tool_env_map = {
"shell": shell_tool_env,
"math": math_tool_env,
"docqa": docqa_tool_env,
"hotpotQA": hotpot_tool_env,
"websearch": websearch_tool_env
}
total_env, env_name_list = {}, []
def loading():
return "Loading..."
def load_text_from_pdf(up, key=None):
global global_dict
if key == None:
key = os.environ.get("OPENAI_API_KEY")
pdf_path = up.name
pdf_reader = PdfReader(pdf_path)
text = ""
for page in pdf_reader.pages:
text += page.extract_text()
# split into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
add_start_index=True
)
chunks = text_splitter.split_text(text)
# create embeddings
# embeddings = OpenAIEmbeddings()
embeddings = OpenAIEmbeddings(openai_api_key=key)
global_dict["knowledge_base"] = FAISS.from_texts(chunks, embeddings)
return "upload success!"
#return knowledge_base
def update(query, OPENAI_API_KEY, BING_SUBSCRIPT_KEY, WOLFRAMALPH_APP_ID, WEATHER_API_KEYS):
global total_env, env_name_list
# print(total_env)
# print(BING_SUBSCRIPT_KEY)
# print(WOLFRAMALPH_APP_ID)
# print(WEATHER_API_KEYS)
# model_name = "mistralai/mistral-7b-instruct-v0.2"
model_name = "gpt-4"
method_name = "cli_inference"
tool_doc_path = "Smurfs/tools/tool_doc.json"
if OPENAI_API_KEY == None or OPENAI_API_KEY == '':
yield [(query, "No OPENAI KEY provided!")]
raise KeyError
if (BING_SUBSCRIPT_KEY == None or BING_SUBSCRIPT_KEY == ''):
yield [(query, "No BING_SUBSCRIPT_KEY provided! Please register one from https://www.microsoft.com/en-us/bing/apis/bing-web-search-api and add it to your keys")]
raise KeyError
if WOLFRAMALPH_APP_ID == None or WOLFRAMALPH_APP_ID == '':
yield [(query, "No WOLFRAMALPH_APP_ID provided! please register one from https://products.wolframalpha.com/api/ and add it to your keys")]
raise KeyError
if WEATHER_API_KEYS == None or WEATHER_API_KEYS == '':
yield [(query, "No WEATHER_API_KEYS provided! Please register one from https://www.weatherapi.com/ and add it to")]
raise KeyError
llm = OpenAI_Model(model_name=model_name, api_key=OPENAI_API_KEY)
#llm = OpenRouter_Model(model_name=model_name)
if "docqa" in total_env:
sys_prompt = llm.sys_prompt + "You already have access to the file uploaded by the user. So just answer the question from the user, you don't need to find the file first."
llm.change_sys_prompt(sys_prompt)
else:
llm.set_default_sys_prompt()
# parser_llm = OpenAI_Model(model_name="gpt-4")
with open(tool_doc_path, "r") as f:
tool_doc = json.load(f)
tool_doc["bing_search"]["api_description"] += f"Today is {current_datetime.year}.{current_datetime.month}.{current_datetime.day}"
available_tools = []
for env_name in env_name_list:
available_tools.append(tool_doc[env_name])
test_set = "cli"
output_dir = f"data/{method_name}/{test_set}/answer"
results_dir = f"data/{method_name}/{test_set}/results.json"
if not os.path.exists(f"data/{method_name}/{test_set}/parser_log"):
os.makedirs(f"data/{method_name}/{test_set}/parser_log")
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# HP_answer_agent = answer_agent(llm=parser_llm, logger_dir=f"data/{method_name}/{test_set}/parser_log")
# worker = smurfs_hotpot_worker(available_tools, HotpotToolEnv, llm, method_name, test_set, answer_agent, executor_agent,hotpot_planning_agent, verifier_agent)
#print(total_env)
worker = stream_smurfs_worker(available_tools, total_env, llm, method_name, test_set, stream_answer_agent, stream_executor_agent, stream_hotpot_planning_agent, stream_verifier_agent, OPENAI_API_KEY, BING_SUBSCRIPT_KEY, WOLFRAMALPH_APP_ID, WEATHER_API_KEYS)
stream_generator = worker.run(query, 0)
# messages = []
while True:
try:
response = next(stream_generator)
messages = [(query, response)]
yield messages
except StopIteration:
break
# query = input("Please Enter Your Task: ")
# cli_run(query, worker)
# def rs_to_tool_env(rs):
# total_env = {}
# env_name_list = []
# for tool_system in rs:
# tool = tool_system.split(": ")[0]
# env = tool_env_map[tool]
# print(f"env: {env}")
# for e in env:
# if e not in env_name_list:
# total_env[e] = env[e]
# env_name_list.append(e)
# print(total_env)
# return total_env, env_name_list
# def merge(rs):
# total_env, env_name_list = rs_to_tool_env(rs)
# return total_env, env_name_list
def update_tools(rs):
global total_env, env_name_list
total_env = {}
env_name_list = []
for tool_system in rs:
tool = tool_system.split(": ")[0]
env = tool_env_map[tool]
print(f"env: {env}")
for e in env:
if e not in env_name_list:
total_env[e] = env[e]
env_name_list.append(e)
print(total_env)
#return total_env, env_name_list
def user(user_msg):
return user_msg
tools = ["math: Tool that can handle mathematical problems",
"docqa: Tool that can answer questions about your uploaded file",
"hotpotQA: Tool that can do multi-hop commonsense reasoning",
"websearch: Tool that can do web search to answer your question"]
websearch_example = ["请根据深圳明天的天气推荐给我推荐一套穿搭方案，结果用中文输出。", "今年的中秋节是哪天？用中文输出"]
math_example = ["Calc integral of sin(x)+2x^2+3x+1 from 0 to 1", "When both sides of a right triangle are 6 and 8, what is the length of the other side?"]
inp = gr.Textbox(placeholder="Please input your task", label="Task")
with gr.Blocks() as demo:
gr.HTML("""
