File size: 4,122 Bytes
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 | import re, ast
import logging
import re
import json
import logging
from typing import List, Dict
from llama_index.core.tools.types import ToolOutput
from src.agent.agenttools import ChatBotFunctionTools
class LearningAgent:
def __init__(self, llm_value: str):
tool_builder = ChatBotFunctionTools(llm_type=llm_value)
self.llm = tool_builder.generator
self.tools = tool_builder.get_tools()
self.chat_history: List[Dict[str, str]] = []
self.max_history = 20
def extract_json_from_markdown(self, markdown_text: str):
try:
match = re.search(r"```json\s*(\{.*?\})\s*```", markdown_text, re.DOTALL)
if not match:
raise ValueError("No JSON block found in markdown")
return json.loads(match.group(1))
except Exception as e:
logging.error(f"Could not parse JSON: {e}")
return None
def determine_tools(self, query: str):
previous_questions = "\n".join(
[msg["content"] for msg in self.chat_history if msg["role"] == "user"]
) 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:
For Choosing tool properly analyze the condensed query and then decide. Choose multiple if its necessary based on condensed query
Its mandatory to select to atleast 1 tool.
If asked for any query related to code generation dont include theory rather give code and its explaination
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)
try:
response = response.text
except:
response = response
parsed = self.extract_json_from_markdown(response)
condensed_query = parsed["condensed_query"]
tools = [t.strip() for t in parsed['tool_names'].split(",") if t.strip() in self.tools]
return tools if tools else ['llm_query'], condensed_query
def execute_tools(self, tools: List[str], query: str) -> tuple[str, str]:
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]:
tools, condensed_query = self.determine_tools(query)
tool_used, response = self.execute_tools(tools, condensed_query)
self.chat_history += [{"role": "user", "content": query}, {"role": "assistant", "content": response}]
self.chat_history = self.chat_history[-(self.max_history * 2):]
return self.chat_history, tool_used, response
|