import os import logging from typing import Optional from datetime import datetime import chromadb from llama_index.core.tools import QueryEngineTool, FunctionTool, ToolMetadata from llama_index.agent.openai import OpenAIAgent from llama_index.vector_stores.chroma import ChromaVectorStore from llama_index.core import VectorStoreIndex from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.llms.openai import OpenAI from llama_index.core.vector_stores import ( MetadataFilters, MetadataFilter, FilterCondition, ) import gradio as gr from gradio.themes.utils import ( fonts, ) from utils import init_mongo_db from tutor_prompts import ( TEXT_QA_TEMPLATE, QueryValidation, system_message_validation, system_message_openai_agent, ) from call_openai import api_function_call logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) logging.getLogger("httpx").setLevel(logging.WARNING) # # This variables are used to intercept API calls # # launch mitmweb # cert_file = "/Users/omar/Downloads/mitmproxy-ca-cert.pem" # os.environ["REQUESTS_CA_BUNDLE"] = cert_file # os.environ["SSL_CERT_FILE"] = cert_file # os.environ["HTTPS_PROXY"] = "http://127.0.0.1:8080" CONCURRENCY_COUNT = int(os.getenv("CONCURRENCY_COUNT", 64)) MONGODB_URI = os.getenv("MONGODB_URI") AVAILABLE_SOURCES_UI = [ "Gen AI 360: LLMs", "Gen AI 360: LangChain", "Gen AI 360: Advanced RAG", "Towards AI Blog", "Activeloop Docs", "HF Transformers Docs", "Wikipedia", "OpenAI Docs", "LangChain Docs", ] AVAILABLE_SOURCES = [ "llm_course", "langchain_course", "advanced_rag_course", "towards_ai", "activeloop", "hf_transformers", "wikipedia", "openai", "langchain_docs", ] # Initialize MongoDB mongo_db = ( init_mongo_db(uri=MONGODB_URI, db_name="towardsai-buster") if MONGODB_URI else logger.warning("No mongodb uri found, you will not be able to save data.") ) # Initialize vector store and index db2 = chromadb.PersistentClient(path="scripts/ai-tutor-db") chroma_collection = db2.get_or_create_collection("ai-tutor-db") vector_store = ChromaVectorStore(chroma_collection=chroma_collection) index = VectorStoreIndex.from_vector_store(vector_store=vector_store) # Initialize OpenAI models llm = OpenAI(temperature=0, model="gpt-3.5-turbo-0125", max_tokens=None) # embeds = OpenAIEmbedding(model="text-embedding-3-large", mode="text_search") embeds = OpenAIEmbedding(model="text-embedding-3-large", mode="similarity") query_engine = index.as_query_engine( llm=llm, similarity_top_k=5, embed_model=embeds, streaming=True, text_qa_template=TEXT_QA_TEMPLATE, ) query_engine_tools = [ QueryEngineTool( query_engine=query_engine, metadata=ToolMetadata( name="AI_information", description="""The 'AI_information' tool serves as a comprehensive repository for insights into the field of artificial intelligence. When utilizing this tool, the input should be the user's complete question. The input can also be adapted to focus on specific aspects or further details of the current topic under discussion. This dynamic input approach allows for a tailored exploration of AI subjects, ensuring that responses are relevant and informative. Employ this tool to fetch nuanced information on topics such as model training, fine-tuning, LLM augmentation, and more, thereby facilitating a rich, context-aware dialogue.""", ), ) ] def initialize_agent(): agent = OpenAIAgent.from_tools( query_engine_tools, llm=llm, verbose=True, system_prompt=system_message_openai_agent, ) return agent def reset_agent(agent_state): agent_state = initialize_agent() # Reset the agent by reassigning a new instance chatbot = [[None, None]] return "Agent has been reset.", chatbot def log_emails(email: gr.Textbox): collection = "email_data-test" logger.info(f"User reported {email=}") email_document = {"email": email} try: mongo_db[collection].insert_one(email_document) logger.info("") except: logger.info("Something went wrong logging") return "" def format_sources(completion) -> str: if len(completion.source_nodes) == 0: return "" # Mapping of source system names to user-friendly names display_source_to_ui = { src: ui for src, ui in zip(AVAILABLE_SOURCES, AVAILABLE_SOURCES_UI) } documents_answer_template: str = ( "📝 Here are the sources I used to answer your question:\n\n{documents}" ) document_template: str = "[🔗 {source}: {title}]({url}), relevance: {score:2.2f}" documents = "\n".join( [ document_template.format( title=src.metadata["title"], score=src.score, source=display_source_to_ui.get( src.metadata["source"], src.metadata["source"] ), url=src.metadata["url"], ) for src in completion.source_nodes ] ) return documents_answer_template.format(documents=documents) def add_sources(history, completion): if completion is None: yield history formatted_sources = format_sources(completion) if formatted_sources == "": yield history history[-1][1] += "\n\n" + formatted_sources yield history def user(user_input, history, agent_state): agent = agent_state return "", history + [[user_input, None]] def get_answer(history, agent_state): user_input = history[-1][0] history[-1][1] = "" completion = agent_state.stream_chat(user_input) for token in completion.response_gen: history[-1][1] += token yield history, completion logger.info(f"completion: {history[-1][1]=}") example_questions = [ "What is the LLama model?", "What is a Large Language Model?", "What is an embedding?", ] theme = gr.themes.Soft() with gr.Blocks( theme=gr.themes.Soft( primary_hue="blue", secondary_hue="blue", font=fonts.GoogleFont("Source Sans Pro"), font_mono=fonts.GoogleFont("IBM Plex Mono"), ), fill_height=True, ) as demo: agent_state = gr.State(initialize_agent()) with gr.Row(): gr.HTML( "