File size: 9,876 Bytes
9023ac5 916b2a2 9023ac5 916b2a2 9023ac5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 | from llama_index.core.tools import FunctionTool
from src.setup.utils import retry
from src.controller.customlogger import logging
from src.llm.source_llm import LLMCall
class ChatBotFunctionTools:
def __init__(self, llm_type="google"):
self.generator = LLMCall(llm_type).get_llm()
@retry(max_retries=5, delay=1)
def machine_learning_concept(self, query):
logging.info(f"Tool Call: machine_learning_concept('{query}')")
prompt = (
"You are a Machine Learning teacher.\n"
"Explain the following ML concept in:\n"
"The explaination should include geometrical or mathematical intuition"
"Try to keep answer crisp and compact"
f"User Query: {query}"
)
return self.generator.complete(prompt)
@retry(max_retries=5, delay=1)
def math_concept(self, query):
logging.info(f"Tool Call: math_concept('{query}')")
prompt = (
"You are a Math teacher.\n"
"Explain the following mathematics behind the Machine Learning algorithm in details with each step by step :\n"
"Try to keep answer crisp and compact"
f"User Query: {query}"
)
return self.generator.complete(prompt)
@retry(max_retries=5, delay=1)
def deep_learning_architecture(self, arch):
logging.info(f"Tool Call: deep_learning_architecture('{arch}')")
prompt = f"Explain the {arch} neural network architecture with diagram description"
return self.generator.complete(prompt)
@retry(max_retries=5, delay=1)
def visualize_algorithm(self, algo):
logging.info(f"Tool Call: visualize_algorithm('{algo}')")
prompt = f"Create visualization code that demonstrates how {algo} works"
return self.generator.complete(prompt)
@retry(max_retries=5, delay=1)
def concept_combiner(self, concepts):
logging.info(f"Tool Call: concept_combiner('{concepts}')")
prompt = f"Explain the relationship between these concepts: {', '.join(concepts)}"
return self.generator.complete(prompt)
@retry(max_retries=5, delay=1)
def code_generator(self, query):
logging.info(f"Tool Call: copilot('{query}')")
prompt = (
"You are a Python Expert.\n"
"Give the python source code as asked like copilot or help to debug a particular code block:\n"
f"{query}\n"
"Keep it compact and dont give much theory. Explain code blocks only."
)
return self.generator.complete(prompt)
@retry(max_retries=5, delay=1)
def llm_query(self, concepts):
logging.info(f"Tool Call: llm_query('{concepts}')")
prompt = f"You are an Expert to Answer the following question respond to the best of your knowledge: {', '.join(concepts)}"
return self.generator.complete(prompt)
@retry(max_retries=5, delay=1)
def get_tools(self):
return {
"ml_concept": FunctionTool.from_defaults(fn=self.machine_learning_concept),
"dl_architecture": FunctionTool.from_defaults(fn=self.deep_learning_architecture),
"algo_visualizer": FunctionTool.from_defaults(fn=self.visualize_algorithm),
"concept_combiner": FunctionTool.from_defaults(fn=self.concept_combiner),
"math_concept": FunctionTool.from_defaults(fn=self.math_concept),
"llm_query": FunctionTool.from_defaults(fn=self.llm_query),
"code_generator": FunctionTool.from_defaults(fn=self.code_generator)
}
# class LearningAgent:
# """Core agent that orchestrates tool usage for learning system"""
#
# def __init__(self, llm_value):
# self.llm = LLMCall(llm_type=llm_value).get_llm()
# self.tools = self._setup_tools()
# self.chat_history: List[Dict[str, str]] = [] # Stores properly formatted messages
# self.max_history = 20
#
# def _setup_tools(self) -> Dict[str, FunctionTool]:
# """Initialize all learning tools"""
# return {
# **self._setup_ml_tools(),
# **self._setup_dl_tools(),
# **self._setup_graph_tools(),
# **self._setup_utility_tools()
# }
#
# def _setup_ml_tools(self) -> Dict[str, FunctionTool]:
# """Machine Learning tools"""
#
# def ml_concept_explainer(query: str) -> str:
# prompt = f"Explain this ML concept in simple terms with examples: {query}"
# return self.llm.complete(prompt).text
#
# return {
# "ml_concept": FunctionTool.from_defaults(fn=ml_concept_explainer)
# }
#
# def _setup_dl_tools(self) -> Dict[str, FunctionTool]:
# """Deep Learning tools"""
#
# def dl_architecture(arch: str) -> str:
# prompt = f"Explain the {arch} neural network architecture with diagram description"
# return self.llm.complete(prompt).text
#
# return {
# "dl_architecture": FunctionTool.from_defaults(fn=dl_architecture)
# }
#
# def _setup_graph_tools(self) -> Dict[str, FunctionTool]:
# """Graph/Visualization tools"""
#
# def visualize_algorithm(algo: str) -> str:
# prompt = f"Create visualization code that demonstrates how {algo} works"
# return self.llm.complete(prompt).text
#
# return {
# "algo_visualizer": FunctionTool.from_defaults(fn=visualize_algorithm)
# }
#
# def _setup_utility_tools(self) -> Dict[str, FunctionTool]:
# """Utility tools"""
#
# def concept_combiner(concepts: List[str]) -> str:
# prompt = f"Explain the relationship between these concepts: {', '.join(concepts)}"
# return self.llm.complete(prompt).text
#
# return {
# "concept_combiner": FunctionTool.from_defaults(fn=concept_combiner)
# }
#
# def extract_json_from_markdown(self, markdown_text):
# """
# Extract and parse JSON content from a markdown-style code block.
# Handles formats like ```json ... ```
# """
# try:
# # Extract JSON block using regex
# match = re.search(r"```json\s*(\{.*?\})\s*```", markdown_text, re.DOTALL)
# if not match:
# raise ValueError("No JSON block found in markdown")
#
# json_str = match.group(1)
# return json.loads(json_str)
# except Exception as e:
# print(f"[ERROR] Could not parse JSON: {e}")
# return None
#
# def determine_tools(self, query: str):
# """Decide which tools to use based on query"""
# previous_questions = ""
# if self.chat_history:
# previous_questions = "\n".join([msg["content"] for msg in self.chat_history if msg["role"] == "user"][:-1]) \
# if self.chat_history else "No previous questions"
# prompt = f"""Analyze this learning query and select appropriate tools also form the condensed query based on
# previous_questions and Query:
# Query: {query}
# Previous Query: {previous_questions}
# Available Tools: {list(self.tools.keys())}
# Return dictionary of tool names and condensed query as dictionary in the below format:
# ```json
# {{
# "condensed_query": condensed query considering chat history and user query as string,
# "tool_names": tool names as comma-separated list
# }}
# ```
# Do Not add additional text"""
#
# response = self.llm.complete(prompt).text
# response = self.extract_json_from_markdown(response)
# return [t.strip() for t in response['tool_names'].split(",") if t.strip() in self.tools], response["condensed_query"]
#
# def execute_tools(self, tools: List[str], query: str) -> tuple[str, str]:
# """Execute multiple tools and combine results"""
# tool_results = []
# content_results = []
#
# for tool in tools:
# try:
# tool_output = self.tools[tool](query)
# content = tool_output.content if isinstance(tool_output, ToolOutput) else str(tool_output)
# tool_results.append(tool)
# content_results.append(content)
# except Exception as e:
# logging.error(f"Tool {tool} failed: {str(e)}")
# tool_results.append(tool)
# content_results.append(f"Error: {str(e)}")
#
# if len(tool_results) > 1:
# combined = "\n\n".join(f"**{t}**:\n{c}" for t, c in zip(tool_results, content_results))
# explanation = self.tools["concept_combiner"](content_results)
# return "multiple", f"{combined}\n\n**Combined Analysis**:\n{explanation}"
# return tool_results[0], content_results[0]
#
# def process_query(self, query: str) -> tuple[List[Dict[str, str]], str, str]:
# """Process query and return properly formatted messages"""
# tools, condensed_query = self.determine_tools(query)
# logging.info(f"Selected tools: {tools}")
#
# tool_used, response = self.execute_tools(tools, condensed_query)
#
# # Format messages for Gradio Chatbot
# user_msg = {"role": "user", "content": query.title()}
# assistant_msg = {"role": "assistant", "content": response}
#
# self.chat_history.extend([user_msg, assistant_msg])
# if len(self.chat_history) > self.max_history * 2: # *2 for user+assistant pairs
# self.chat_history = self.chat_history[-(self.max_history * 2):]
#
# return self.chat_history, tool_used, response
|