import json from multiprocessing import Pool, cpu_count # import requests # from tenacity import RetryError import re import logging import chainlit as cl from termcolor import colored from typing import Any, Dict, Union, List from typing import TypedDict, Annotated from langgraph.graph.message import add_messages from agents.base_agent import BaseAgent from utils.read_markdown import read_markdown_file from tools.google_serper import serper_search, serper_shopping_search from utils.logging import log_function, setup_logging from tools.offline_graph_rag_tool import run_rag from prompt_engineering.guided_json_lib import ( guided_json_search_query, guided_json_best_url_two, guided_json_router_decision, guided_json_parse_expert, guided_json_search_query_two ) setup_logging(level=logging.DEBUG) logger = logging.getLogger(__name__) class MessageDict(TypedDict): role: str content: str class State(TypedDict): meta_prompt: Annotated[List[dict], add_messages] conversation_history: Annotated[List[dict], add_messages] requirements_gathering: Annotated[List[str], add_messages] expert_plan: str expert_research: Annotated[List[str], add_messages] expert_research_shopping: Annotated[List[str], add_messages] expert_writing: str user_input: Annotated[List[str], add_messages] previous_search_queries: Annotated[List[dict], add_messages] router_decision: str chat_limit: int chat_finished: bool recursion_limit: int final_answer: str previous_type2_work: Annotated[List[str], add_messages] state: State = { "meta_prompt": [], "conversation_history": [], "requirements_gathering": [], "expert_plan": [], "expert_research": [], "expert_research_shopping": [], "expert_writing": [], "user_input": [], "previous_search_queries": [], "router_decision": None, "chat_limit": None, "chat_finished": False, "recursion_limit": None, "final_answer": None, "previous_type2_work": [] } def chat_counter(state: State) -> State: chat_limit = state.get("chat_limit") if chat_limit is None: chat_limit = 0 chat_limit += 1 state["chat_limit"] = chat_limit return chat_limit def routing_function(state: State) -> str: decision = state["router_decision"] print(colored(f"\n\n Routing function called. Decision: {decision}\n\n", 'green')) return decision def set_chat_finished(state: State) -> bool: state["chat_finished"] = True final_response = state["meta_prompt"][-1].content print(colored(f"\n\n DEBUG FINAL RESPONSE: {final_response}\n\n", 'green')) # Split the response at ">> FINAL ANSWER:" parts = final_response.split(">> FINAL ANSWER:") if len(parts) > 1: answer_part = parts[1].strip() # Remove any triple quotes final_response_formatted = answer_part.strip('"""') # Remove leading whitespace final_response_formatted = final_response_formatted.lstrip() # Remove the CoR dictionary at the end cor_pattern = r'\nCoR\s*=\s*\{[\s\S]*\}\s*$' final_response_formatted = re.sub(cor_pattern, '', final_response_formatted) # Remove any trailing whitespace final_response_formatted = final_response_formatted.rstrip() # print(colored(f"\n\n DEBUG: {final_response_formatted}\n\n", 'green')) print(colored(f"\n\n Jarvis👩‍💻: {final_response_formatted}", 'cyan')) state["final_answer"] = f'''{final_response_formatted}''' else: print(colored("Error: Could not find '>> FINAL ANSWER:' in the response", 'red')) state["final_answer"] = "Error: No final answer found" return state class Jar3d(BaseAgent[State]): def __init__(self, model: str = None, server: str = None, temperature: float = 0, model_endpoint: str = None, stop: str = None): super().__init__(model, server, temperature, model_endpoint, stop) self.llm = self.get_llm(json_model=False) def get_prompt(self, state: State = None) -> str: system_prompt = read_markdown_file('prompt_engineering/jar3d_requirements_prompt.md') return system_prompt def process_response(self, response: Any, user_input: str, state: State = None) -> Dict[str, List[Dict[str, str]]]: updates_conversation_history = { "requirements_gathering": [ {"role": "user", "content": f"{user_input}"}, {"role": "assistant", "content": str(response)} ] } return updates_conversation_history def get_conv_history(self, state: State) -> str: conversation_history = state.get('requirements_gathering', []) return "\n".join([f"{msg['role']}: {msg['content']}" for msg in conversation_history]) def get_user_input(self) -> str: pass def get_guided_json(self, state: State) -> Dict[str, Any]: pass def use_tool(self) -> Any: pass def run_chainlit(self, state: State, message: cl.Message) -> State: user_message = message.content system_prompt = self.get_prompt() user_input = f"{system_prompt}\n cogor {user_message}" state = self.invoke(state=state, user_input=user_input) response = state['requirements_gathering'][-1]["content"] response = re.sub(r'^```python[\s\S]*?```\s*', '', response, flags=re.MULTILINE) response = response.lstrip() return state, response class MetaExpert(BaseAgent[State]): def __init__(self, model: str = None, server: str = None, temperature: float = 0, model_endpoint: str = None, stop: str = None): super().__init__(model, server, temperature, model_endpoint, stop) self.llm = self.get_llm(json_model=False) def get_prompt(self, state:None) -> str: system_prompt = read_markdown_file('prompt_engineering/jar3d_meta_prompt.md') return system_prompt def process_response(self, response: Any, user_input: str, state: State = None) -> Dict[str, List[MessageDict]]: user_input = None updates_conversation_history = { "meta_prompt": [ {"role": "user", "content": f"{user_input}"}, {"role": "assistant", "content": str(response)} ], "conversation_history": [ {"role": "user", "content": f"{user_input}"}, {"role": "assistant", "content": str(response)} ], } return updates_conversation_history # @log_function(logger) def get_conv_history(self, state: State) -> str: all_expert_research = [] if state["expert_research"]: expert_research = state["expert_research"] all_expert_research.extend(expert_research) else: all_expert_research = [] expert_message_history = f""" ## Your Expert Plan:\n{state.get("expert_plan", [])}\n ## Your Expert Writing:\n{state.get("expert_writing", [])}\n ## Your Expert Shopping List:\n{state.get("expert_research_shopping", [])}\n ## Your Expert Research:{all_expert_research}\n """ return expert_message_history def get_user_input(self) -> str: user_input = input("Enter your query: ") return user_input def get_guided_json(self, state: State) -> Dict[str, Any]: pass def use_tool(self) -> Any: pass @log_function(logger) def run(self, state: State) -> State: counter = chat_counter(state) # Counts every time we invoke the Meta Agent recursion_limit = state.get("recursion_limit") recursions = 3*counter - 2 print(colored(f"\n\n * We have envoked the Meta-Agent {counter} times.\n * we have run {recursions} max total iterations: {recursion_limit}\n\n", "green")) upper_limit_recursions = recursion_limit lower_limit_recursions = recursion_limit - 2 if recursions >= lower_limit_recursions and recursions <= upper_limit_recursions: final_answer = "**You are being explicitly told to produce your [Type 2] work now!**" elif recursions > upper_limit_recursions: extra_recursions = recursions - upper_limit_recursions base_message = "**You are being explicitly told to produce your [Type 2] work now!**" final_answer = (base_message + "\n") * (extra_recursions + 1) else: final_answer = None try: requirements = state['requirements_gathering'][-1]["content"] except: requirements = state['requirements_gathering'][-1].content formatted_requirements = '\n\n'.join(re.findall(r'```python\s*([\s\S]*?)\s*```', requirements, re.MULTILINE)) print(colored(f"\n\n User Requirements: {formatted_requirements}\n\n", 'green')) if state.get("meta_prompt"): try: meta_prompt = state['meta_prompt'][-1]["content"] except: meta_prompt = state['meta_prompt'][-1].content # print(colored(f"\n\n DEBUG Meta-Prompt: {meta_prompt}\n\n", 'yellow')) cor_match = '\n\n'.join(re.findall(r'```python\s*([\s\S]*?)\s*```', meta_prompt, re.MULTILINE)) # print(colored(f"\n\n DEBUG CoR Match: {cor_match}\n\n", 'yellow')) user_input = f"{formatted_requirements} \n\n Here is your last CoR {cor_match} update your next CoR from here." else: user_input = formatted_requirements state = self.invoke(state=state, user_input=user_input, final_answer=final_answer) meta_prompt_cor = state['meta_prompt'][-1]["content"] print(colored(f"\n\n Meta-Prompt Chain of Reasoning: {meta_prompt_cor}\n\n", 'green')) return state class NoToolExpert(BaseAgent[State]): print(colored(f"\n\n DEBUG: We are running the NoToolExpert tool\n\n", 'red')) def __init__(self, model: str = None, server: str = None, temperature: float = 0, model_endpoint: str = None, stop: str = None): super().__init__(model, server, temperature, model_endpoint, stop) self.llm = self.get_llm(json_model=False) def get_prompt(self, state) -> str: # print(f"\nn{state}\n") system_prompt = state["meta_prompt"][-1].content return system_prompt def process_response(self, response: Any, user_input: str = None, state: State = None) -> Dict[str, Union[str, dict]]: # meta_prompts = state.get("meta_prompt", []) associated_meta_prompt = state["meta_prompt"][-1].content parse_expert = self.get_llm(json_model=True) parse_expert_prompt = """ You must parse the expert from the text. The expert will be one of the following. 1. Expert Planner 2. Expert Writer Return your response as the following JSON {{"expert": "Expert Planner" or "Expert Writer"}} """ input = [ {"role": "user", "content": associated_meta_prompt}, {"role": "assistant", "content": f"system_prompt:{parse_expert_prompt}"} ] retries = 0 associated_expert = None while retries < 4 and associated_expert is None: retries += 1 if self.server == 'vllm': guided_json = guided_json_parse_expert parse_expert_response = parse_expert.invoke(input, guided_json) else: parse_expert_response = parse_expert.invoke(input) associated_expert_json = json.loads(parse_expert_response) associated_expert = associated_expert_json.get("expert") # associated_expert = parse_expert_text(associated_meta_prompt) print(colored(f"\n\n Expert: {associated_expert}\n\n", 'green')) if associated_expert == "Expert Planner": expert_update_key = "expert_plan" if associated_expert == "Expert Writer": expert_update_key = "expert_writing" updates_conversation_history = { "conversation_history": [ {"role": "user", "content": user_input}, {"role": "assistant", "content": f"{str(response)}"} ], expert_update_key: {"role": "assistant", "content": f"{str(response)}"} } return updates_conversation_history def get_conv_history(self, state: State) -> str: pass def get_user_input(self) -> str: pass def get_guided_json(self, state: State) -> Dict[str, Any]: pass def use_tool(self) -> Any: pass # @log_function(logger) def run(self, state: State) -> State: # chat_counter(state) all_expert_research = [] meta_prompt = state["meta_prompt"][1].content if state.get("expert_research"): expert_research = state["expert_research"] all_expert_research.extend(expert_research) research_prompt = f"\n Your response must be delivered considering following research.\n ## Research\n {all_expert_research} " user_input = f"{meta_prompt}\n{research_prompt}" else: user_input = meta_prompt print(colored(f"\n\n DEBUG: We are running the NoToolExpert tool\n\n", 'red')) state = self.invoke(state=state, user_input=user_input) return state class ToolExpert(BaseAgent[State]): print(colored(f"\n\n DEBUG: We are running the ToolExpert tool\n\n", 'red')) def __init__(self, model: str = None, server: str = None, temperature: float = 0, model_endpoint: str = None, stop: str = None, location: str = None, hybrid: bool = False): super().__init__(model, server, temperature, model_endpoint, stop, location, hybrid) print(f"\n\n DEBUG LOCATION: {self.location}") self.llm = self.get_llm(json_model=False) def get_prompt(self, state) -> str: system_prompt = state["meta_prompt"][-1].content return system_prompt def process_response(self, response: Any, user_input: str = None, state: State = None) -> Dict[str, Union[str, dict]]: updates_conversation_history = { "conversation_history": [ {"role": "user", "content": user_input}, {"role": "assistant", "content": f"{str(response)}"} ], "expert_research": {"role": "assistant", "content": f"{str(response)}"} } return updates_conversation_history def get_conv_history(self, state: State) -> str: pass def get_user_input(self) -> str: pass def get_guided_json(self, state: State) -> Dict[str, Any]: pass # Change added query list to RAG def use_tool(self, mode: str, engine: str, tool_input: str, meta_prompt: str = None, query: list[str] = None, hybrid: bool = False) -> Any: if mode == "serper": if engine == "search": results = serper_search(tool_input, self.location) return {"results": results, "is_shopping": False} elif engine == "shopping": results = serper_shopping_search(tool_input, self.location) return {"results": results, "is_shopping": True} elif mode == "rag": print(colored(f"\n\n DEBUG: We are running the Graph RAG TOOL!!\n\n", 'red')) # if hybrid: # nodes, relationships = self.get_graph_elements(meta_prompt) # print(colored(f"\n\n DEBUG: Nodes: {nodes}\n\n", 'green')) # print(colored(f"\n\n DEBUG: Relationships: {relationships}\n\n", 'green')) # else: nodes = None relationships = None print(colored(f"\n\n DEBUG Retreival Mode: {hybrid}\n\n", 'green')) results = run_rag(urls=tool_input, allowed_nodes=nodes, allowed_relationships=relationships, query=query, hybrid=self.hybrid) return {"results": results, "is_shopping": False} def generate_search_queries(self, meta_prompt: str, num_queries: int = 5) -> List[str]: refine_query_template = """ # Objective Your mission is to systematically address your manager's instructions by determining the most appropriate search queries to use **AND** to determine the best engine to use for each query. Your engine choice is either search, or shopping. You must return either the search or shopping engine for each query. You will generate {num_queries} different search queries. # Manager's Instructions {manager_instructions} # Flexible Search Algorithm for Simple and Complex Questions 1. Initial search: - For a simple question: "[Question keywords]" - For a complex topic: "[Main topic] overview" 2. For each subsequent search: - Choose one of these strategies: a. Specify: Add a more specific term or aspect related to the topic. b. Broaden: Remove a specific term or add "general" or "overview" to the query. c. Pivot: Choose a different but related term from the topic. d. Compare: Add "vs" or "compared to" along with a related term. e. Question: Rephrase the query as a question by adding "what", "how", "why", etc. # Response Format **Return the following JSON:** {{ "search_queries": [ {{"engine": "search", "query": "Query 1"}}, {{"engine": "shopping", "query": "Query 2"}}, ... {{"engine": "search", "query": "Query {num_queries}"}} ] }} Remember: - Generate {num_queries} unique and diverse search queries. - Each query should explore a different aspect or approach to the topic. - Ensure the queries cover various aspects of the manager's instructions. - The "engine" field should be either "search" or "shopping" for each query. """ refine_query = self.get_llm(json_model=True) refine_prompt = refine_query_template.format(manager_instructions=meta_prompt, num_queries=num_queries) input = [ {"role": "user", "content": "Generate search queries"}, {"role": "assistant", "content": f"system_prompt:{refine_prompt}"} ] guided_json = guided_json_search_query_two if self.server == 'vllm': refined_queries = refine_query.invoke(input, guided_json) else: print(colored(f"\n\n DEBUG: We are running the refine_query tool without vllm\n\n", 'red')) refined_queries = refine_query.invoke(input) refined_queries_json = json.loads(refined_queries) return refined_queries_json.get("search_queries", []) def process_serper_result(self, query, serper_response ): # Add to other Jar3d Script best_url_template = """ Given the serper results, and the search query, select the best URL # Search Query {search_query} # Serper Results {serper_results} **Return the following JSON:** {{"best_url": The URL of the serper results that aligns most with the search query.}} """ best_url = self.get_llm(json_model=True) best_url_prompt = best_url_template.format(search_query=query["query"], serper_results=serper_response) input = [ {"role": "user", "content": serper_response}, {"role": "assistant", "content": f"system_prompt:{best_url_prompt}"} ] guided_json = guided_json_best_url_two if self.server == 'vllm': best_url = best_url.invoke(input, guided_json) else: print(colored(f"\n\n DEBUG: We are running the best_url tool without vllm\n\n", 'red')) best_url = best_url.invoke(input) best_url_json = json.loads(best_url) return {"query": query, "url": best_url_json.get("best_url")} # return best_url_json.get("best_url") def get_graph_elements(self, meta_prompt: str): graph_elements_template = """ You are an intelligent assistant helping to construct elements for a knowledge graph in Neo4j. Your objectove is to create two lists. # Lists list 1: A list of nodes. list 2: A list of relationships. # Instructions Constructing Lists 1. You must construct lists based on what would be most useful for exploring data to fulfil the [Manager's Instructions]. 2. Each item in your list must follow Neo4j's formattng standards. 3. Limit lists to a maximum of 8 items each. # Neo4j Formatting Standards 1. Each element in the list must be in capital letters. 2. Spaces between words must be replaced with underscores. 3. There should be no apostrophes or special characters. # [Manager's Instructions] {manager_instructions} Return the following JSON: {{ "nodes": [list of nodes], "relationships": [list of relationships]}} """ get_graph_elements = self.get_llm(json_model=True) graph_elements_prompt = graph_elements_template.format(manager_instructions=meta_prompt) input = [ {"role": "user", "content": "Construct Neo4j Graph Elements"}, {"role": "assistant", "content": f"system_prompt:{graph_elements_prompt}"} ] guided_json = guided_json_best_url_two if self.server == 'vllm': graph_elements = get_graph_elements.invoke(input, guided_json) else: print(colored(f"\n\n DEBUG: We are running the graph_elements tool without vllm\n\n", 'red')) graph_elements = get_graph_elements.invoke(input) graph_elements_json = json.loads(graph_elements) nodes = graph_elements_json.get("nodes") relationships = graph_elements_json.get("relationships") return nodes, relationships def run(self, state: State) -> State: meta_prompt = state["meta_prompt"][-1].content print(colored(f"\n\n Meta-Prompt: {meta_prompt}\n\n", 'green')) # Generate multiple search queries num_queries = 10 search_queries = self.generate_search_queries(meta_prompt, num_queries=num_queries) print(colored(f"\n\n Generated Search Queries: {search_queries}\n\n", 'green')) try: # Use multiprocessing to call Serper tool for each query in parallel with Pool(processes=min(cpu_count(), len(search_queries))) as pool: serper_results = pool.starmap( self.use_tool, [("serper", query["engine"], query["query"], None) for query in search_queries] ) # Collect shopping results separately shopping_results = [result["results"] for result in serper_results if result["is_shopping"]] if shopping_results: state["expert_research_shopping"] = shopping_results # Process Serper results to get best URLs with Pool(processes=min(cpu_count(), len(serper_results))) as pool: best_urls = pool.starmap( self.process_serper_result, [(query, result["results"]) for query, result in zip(search_queries, serper_results)] # zip(search_queries, serper_results) ) except Exception as e: print(colored(f"Error in multithreaded processing: {str(e)}. Falling back to non-multithreaded approach.", "yellow")) # Fallback to non-multithreaded approach serper_results = [self.use_tool("serper", query["engine"], query["query"], None) for query in search_queries] shopping_results = [result["results"] for result in serper_results if result["is_shopping"]] if shopping_results: state["expert_research_shopping"] = shopping_results best_urls = [self.process_serper_result(query, result) for query, result in zip(search_queries, serper_results)] # Remove duplicates from the list of URLs # Additional Line deduplicated_urls = {item['url']: item for item in best_urls}.values() deduplicated_urls_list = list(deduplicated_urls) print(colored(f"\n\n DEBUG DEBUG Best URLs: {deduplicated_urls_list}\n\n", 'red')) # DEBUG LINE unique_urls = list(dict.fromkeys(result["url"] for result in best_urls if result["url"] and result["query"]["engine"] == "search")) unique_queries = list(dict.fromkeys(result["query"]["query"] for result in deduplicated_urls_list if result["query"] and result["query"]["engine"] == "search")) # unique_urls = list(dict.fromkeys(url for url in best_urls if url)) print(colored("\n\n Sourced data from {} sources:".format(len(unique_urls)), 'yellow')) print(colored(f"\n\n Search Queries {unique_queries}", 'yellow')) for i, url in enumerate(unique_urls, 1): print(colored(" {}. {}".format(i, url), 'green')) print() try: scraper_response = self.use_tool(mode="rag", engine=None, tool_input=unique_urls, meta_prompt=meta_prompt, query=unique_queries) except Exception as e: scraper_response = {"results": f"Error {e}: Failed to scrape results", "is_shopping": False} updates = self.process_response(scraper_response, user_input="Research") for key, value in updates.items(): state = self.update_state(key, value, state) return state class Router(BaseAgent[State]): def __init__(self, model: str = None, server: str = None, temperature: float = 0, model_endpoint: str = None, stop: str = None): super().__init__(model, server, temperature, model_endpoint, stop) self.llm = self.get_llm(json_model=True) def get_prompt(self, state) -> str: system_prompt = state["meta_prompt"][-1].content return system_prompt def process_response(self, response: Any, user_input: str = None, state: State = None) -> Dict[str, Union[str, dict]]: updates_conversation_history = { "router_decision": [ {"role": "user", "content": user_input}, {"role": "assistant", "content": f"{str(response)}"} ] } return updates_conversation_history def get_conv_history(self, state: State) -> str: pass def get_user_input(self) -> str: pass def get_guided_json(self, state: State) -> Dict[str, Any]: pass def use_tool(self, tool_input: str, mode: str) -> Any: pass # @log_function(logger) def run(self, state: State) -> State: router_template = """ Given these instructions from your manager. # Response from Manager {manager_response} **Return the following JSON:** {{""router_decision: Return the next agent to pass control to.}} **strictly** adhere to these **guidelines** for routing. If your maneger's response contains "Expert Internet Researcher", return "tool_expert". If your manager's response contains "Expert Planner" or "Expert Writer", return "no_tool_expert". If your manager's response contains '>> FINAL ANSWER:', return "end_chat". """ system_prompt = router_template.format(manager_response=state["meta_prompt"][-1].content) input = [ {"role": "user", "content": ""}, {"role": "assistant", "content": f"system_prompt:{system_prompt}"} ] router = self.get_llm(json_model=True) if self.server == 'vllm': guided_json = guided_json_router_decision router_response = router.invoke(input, guided_json) else: router_response = router.invoke(input) router_response = json.loads(router_response) router_response = router_response.get("router_decision") state = self.update_state("router_decision", router_response, state) return state