|
|
from sophos_models import Sophos |
|
|
from sophos_tools import * |
|
|
|
|
|
class Agent: |
|
|
def __init__(self, name: str, instructions: str, model: str, tools: list): |
|
|
self.name = name |
|
|
self.instructions = instructions |
|
|
self.model = model |
|
|
self.tools = {tool.__name__: tool for tool in tools if hasattr(tool, "_is_tool")} |
|
|
self.sophos = Sophos(model_path=model) |
|
|
|
|
|
def process_tool_calls(self, response_content: str): |
|
|
"""Process tool calls from response content and return results""" |
|
|
if "```tool" not in response_content: |
|
|
return None |
|
|
|
|
|
print("🔧 TOOL DETECTED - Processing...") |
|
|
try: |
|
|
tool_block = response_content.split("```tool")[1].split("```")[0].strip() |
|
|
print(f"🔧 Tool block found: {tool_block}") |
|
|
|
|
|
tool_results = [] |
|
|
tool_lines = [line.strip() for line in tool_block.split('\n') if line.strip()] |
|
|
|
|
|
for tool_line in tool_lines: |
|
|
if "(" in tool_line and ")" in tool_line: |
|
|
tool_name = tool_line.split("(")[0].strip() |
|
|
arg_part = tool_line.split("(", 1)[1].rsplit(")", 1)[0].strip() |
|
|
arg = arg_part.strip('"').strip("'") |
|
|
|
|
|
if tool_name in self.tools: |
|
|
print(f"🔧 CALLING TOOL: {tool_name}({arg})") |
|
|
if arg: |
|
|
tool_result = self.tools[tool_name](arg) |
|
|
else: |
|
|
tool_result = self.tools[tool_name]() |
|
|
print(f"🔧 TOOL RESULT: {tool_result}") |
|
|
tool_results.append(f"{tool_name}: {tool_result}") |
|
|
else: |
|
|
print(f"🔧 ERROR: Tool {tool_name} not found") |
|
|
tool_results.append(f"Error: Tool {tool_name} not found") |
|
|
|
|
|
return tool_results if tool_results else None |
|
|
|
|
|
except Exception as e: |
|
|
print(f"Error parsing tool: {e}") |
|
|
return None |
|
|
|
|
|
def get_tools_names_and_descriptions(self): |
|
|
return {name: func.__doc__ for name, func in self.tools.items()} |
|
|
|
|
|
def run(self, prompt: str): |
|
|
system_prompt = f"""You are {self.name}. {self.instructions} |
|
|
|
|
|
Available tools: {self.get_tools_names_and_descriptions()} |
|
|
|
|
|
CRITICAL: To use a tool, respond with exactly this format: |
|
|
```tool |
|
|
get_weather(Tokyo) |
|
|
``` |
|
|
|
|
|
You can use multiple tools: |
|
|
```tool |
|
|
get_weather(Tokyo) |
|
|
get_time() |
|
|
``` |
|
|
|
|
|
YOU MUST USE A BLOCK FOR TOOLS, or the system will not understand you are calling a tool. |
|
|
CRITICAL: |
|
|
```tool |
|
|
<tool_name>(<arg>) |
|
|
``` |
|
|
REMEMBER ALL TOOLS MUST BE INSIDE THE SAME BLOCK |
|
|
|
|
|
IMPORTANT WORKFLOW: |
|
|
1. If you need live information, use the appropriate tool(s) first |
|
|
2. Once you have the information you need, provide your final response WITHOUT any tool calls |
|
|
3. Do NOT call the same tool repeatedly unless the user asks for updated information |
|
|
4. After calling a tool once and getting results, provide your answer to the user |
|
|
|
|
|
user prompt: {prompt} |
|
|
""" |
|
|
print(f"--- System Prompt ---\n{system_prompt}\n---------------------") |
|
|
|
|
|
messages = [ |
|
|
{"role": "system", "content": system_prompt}, |
|
|
{"role": "user", "content": prompt} |
|
|
] |
|
|
|
|
|
max_iterations = 99 |
|
|
iteration = 0 |
|
|
last_tool_calls = [] |
|
|
|
|
|
while iteration < max_iterations: |
|
|
iteration += 1 |
|
|
print(f"\n--- Iteration {iteration} ---") |
|
|
|
|
|
response = self.sophos.ask(str(messages)) |
|
|
|
|
|
current_response = response |
|
|
print(f"AI Response: {current_response}") |
|
|
|
|
|
tool_results = self.process_tool_calls(current_response) |
|
|
|
|
|
if tool_results is None: |
|
|
print("🏁 No tool calls detected - Agent finished!") |
|
|
return current_response |
|
|
|
|
|
all_results = "\n".join(tool_results) |
|
|
|
|
|
|
|
|
current_tool_calls = [result.split(":")[0] for result in tool_results] |
|
|
if current_tool_calls == last_tool_calls and iteration > 1: |
|
|
print("⚠️ Detected repeated tool calls - forcing final response") |
|
|
messages.append({"role": "assistant", "content": current_response}) |
|
|
messages.append({"role": "user", "content": f"Tool results:\n{all_results}\n\nYou have already called these tools. Now provide your final answer to the user without calling any more tools."}) |
|
|
else: |
|
|
messages.append({"role": "assistant", "content": current_response}) |
|
|
messages.append({"role": "user", "content": f"Tool results:\n{all_results}\n\nNow provide your final response to the user. Do NOT call any more tools unless absolutely necessary."}) |
|
|
|
|
|
last_tool_calls = current_tool_calls |
|
|
|
|
|
print(f"⚠️ Reached maximum iterations ({max_iterations})") |
|
|
return current_response |
