import json
import logging
from termcolor import colored
from datetime import datetime
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.advanced_scraper import scraper
from tools.google_serper import serper_search
from utils.logging import log_function, setup_logging
from utils.message_handling import get_ai_message_contents
from prompt_engineering.guided_json_lib import guided_json_search_query, guided_json_best_url, guided_json_router_decision
setup_logging(level=logging.DEBUG)
logger = logging.getLogger(__name__)
class MessageDict(TypedDict):
role: str
content: str
class State(TypedDict):
meta_prompt: Annotated[List[MessageDict], add_messages]
conversation_history: Annotated[List[dict], add_messages]
user_input: Annotated[List[str], add_messages]
router_decision: bool
chat_limit: int
chat_finished: bool
recursion_limit: int
state: State = {
"meta_prompt": [],
"conversation_history": [],
"user_input": [],
"router_decision": None,
"chat_limit": None,
"chat_finished": False,
"recursion_limit": None
}
# class State(TypedDict):
# meta_prompt: Annotated[List[MessageDict], add_messages]
# conversation_history: Annotated[List[dict], add_messages]
# user_input: Annotated[List[str], add_messages]
# router_decision: bool
# chat_limit: int
# chat_finished: bool
# state: State = {
# "meta_prompt": [],
# "conversation_history": [],
# "user_input": [],
# "router_decision": None,
# "chat_limit": None,
# "chat_finished": False
# }
# 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 state
# 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:
if state["router_decision"]:
return "no_tool_expert"
else:
return "tool_expert"
def set_chat_finished(state: State) -> bool:
state["chat_finished"] = True
final_response = state["meta_prompt"][-1].content
print(colored(f"\n\n Meta Agent 🧙♂️: {final_response}", 'cyan'))
return state
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/meta_prompt.md')
return system_prompt
def process_response(self, response: Any, user_input: str = None, 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)}
]
}
return updates_conversation_history
@log_function(logger)
def get_conv_history(self, state: State) -> str:
conversation_history = state.get("conversation_history", [])
expert_message_history = get_ai_message_contents(conversation_history)
print(f"Expert Data Collected: {expert_message_history}")
expert_message_history = f"Expert Data Collected: {expert_message_history}"
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)
user_input = state.get("user_input")
state = self.invoke(state=state, user_input=user_input)
return state
class NoToolExpert(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) -> 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]]:
updates_conversation_history = {
"conversation_history": [
{"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) -> Any:
pass
# @log_function(logger)
def run(self, state: State) -> State:
# chat_counter(state)
user_input = state["meta_prompt"][1].content
state = self.invoke(state=state, user_input=user_input)
return state
class ToolExpert(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) -> 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)}"}
]
}
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
# Use Serper to search
def use_tool(self, mode: str, tool_input: str, doc_type: str = None) -> Any:
if mode == "serper":
results = serper_search(tool_input, self.location)
return results
elif mode == "scraper":
results = scraper(tool_input, doc_type)
return results
# @log_function(logger)
def run(self, state: State) -> State:
# counter = chat_counter(state)
refine_query_template = """
Given the response from your manager.
# Response from Manager
{manager_response}
**Return the following JSON:**
{{"search_query": The refined google search engine query that aligns with the response from your managers.}}
"""
best_url_template = """
Given the serper results, and the instructions from your manager. Select the best URL
# Manger Instructions
{manager_response}
# Serper Results
{serper_results}
**Return the following JSON:**
{{"best_url": The URL of the serper results that aligns most with the instructions from your manager.,
"pdf": A boolean value indicating whether the URL is a PDF or not. This should be True if the URL is a PDF, and False otherwise.}}
"""
user_input = state["meta_prompt"][-1].content
state = self.invoke(state=state, user_input=user_input)
full_query = state["conversation_history"][-1].get("content")
refine_query = self.get_llm(json_model=True)
refine_prompt = refine_query_template.format(manager_response=full_query)
input = [
{"role": "user", "content": full_query},
{"role": "assistant", "content": f"system_prompt:{refine_prompt}"}
]
if self.server == 'vllm':
guided_json = guided_json_search_query
refined_query = refine_query.invoke(input, guided_json)
else:
refined_query = refine_query.invoke(input)
refined_query_json = json.loads(refined_query)
refined_query = refined_query_json.get("search_query")
serper_response = self.use_tool("serper", refined_query)
best_url = self.get_llm(json_model=True)
best_url_prompt = best_url_template.format(manager_response=full_query, serper_results=serper_response)
input = [
{"role": "user", "content": serper_response},
{"role": "assistant", "content": f"system_prompt:{best_url_prompt}"}
]
if self.server == 'vllm':
guided_json = guided_json_best_url
best_url = best_url.invoke(input, guided_json)
else:
best_url = best_url.invoke(input)
best_url_json = json.loads(best_url)
best_url = best_url_json.get("best_url")
doc_type = best_url_json.get("pdf")
if doc_type == "True" or doc_type == True:
doc_type = "pdf"
else:
doc_type = "html"
scraper_response = self.use_tool("scraper", best_url, doc_type)
updates = self.process_response(scraper_response, user_input)
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)} Todays date is {datetime.now()}"}
]
}
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 manager's response suggests a tool might be required to answer the query, return "tool_expert".
# If your manager's response suggests no tool is required to answer the query, return "no_tool_expert".
# If your manager's response suggest they have provided a final answer, return "end_chat".
# """
# chat_counter(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 manager's response suggests the Expert Internet Researcher or the suggests the internet might be required, return "tool_expert".
If your manager's response suggests that the internet is not required, return "no_tool_expert".
If your manager's response suggest they have provided a 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
# Example usage
if __name__ == "__main__":
from langgraph.graph import StateGraph
# For Claude
agent_kwargs = {
"model": "claude-3-5-sonnet-20240620",
"server": "claude",
"temperature": 0.5
}
agent_kwargs = {
"model": "gpt-4o-mini",
"server": "openai",
"temperature": 0.1
}
# Ollama
# agent_kwargs = {
# "model": "phi3:instruct",
# "server": "ollama",
# "temperature": 0.5
# }
# Groq
# agent_kwargs = {
# "model": "mixtral-8x7b-32768",
# "server": "groq",
# "temperature": 0.5
# }
# # Gemnin - Not currently working, I will be debugging this soon.
# agent_kwargs = {
# "model": "gemini-1.5-pro",
# "server": "gemini",
# "temperature": 0.5
# }
# # Vllm
# agent_kwargs = {
# "model": "meta-llama/Meta-Llama-3-70B-Instruct",
# "server": "vllm",
# "temperature": 0.5,
# "model_endpoint": "https://vpzatdgopr2pmx-8000.proxy.runpod.net/",
# }
tools_router_agent_kwargs = agent_kwargs.copy()
tools_router_agent_kwargs["temperature"] = 0
def routing_function(state: State) -> str:
decision = state["router_decision"]
print(colored(f"\n\n Routing function called. Decision: {decision}", 'red'))
return decision
graph = StateGraph(State)
graph.add_node("meta_expert", lambda state: MetaExpert(**agent_kwargs).run(state=state))
graph.add_node("router", lambda state: Router(**tools_router_agent_kwargs).run(state=state))
graph.add_node("no_tool_expert", lambda state: NoToolExpert(**agent_kwargs).run(state=state))
graph.add_node("tool_expert", lambda state: ToolExpert(**tools_router_agent_kwargs).run(state=state))
graph.add_node("end_chat", lambda state: set_chat_finished(state))
graph.set_entry_point("meta_expert")
graph.set_finish_point("end_chat")
graph.add_edge("meta_expert", "router")
graph.add_edge("tool_expert", "meta_expert")
graph.add_edge("no_tool_expert", "meta_expert")
graph.add_conditional_edges(
"router",
lambda state: routing_function(state),
)
workflow = graph.compile()
while True:
query = input("Ask me anything: ")
if query.lower() == "exit":
break
# current_time = datetime.now()
recursion_limit = 15
state["recursion_limit"] = recursion_limit
state["user_input"] = query
limit = {"recursion_limit": recursion_limit}
for event in workflow.stream(state, limit):
pass