slipstream/discordbot
slipstream/discordbot
1
1
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

fix: Refactor tool call processing in TetoCog to improve error handling and streamline execution logic

Browse Source
This commit is contained in:
Slipstream 2025-05-28 23:46:27 -06:00
parent 72b0ce1a4d
commit c55113a4ca
Signed by: slipstream
GPG Key ID: 13E498CE010AC6FD
1 changed files with 46 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
cogs/teto_cog.py
View File

@ -391,7 +391,7 @@ class TetoCog(commands.Cog):
tools=vertex_tools, # Add tools here
tool_config=types.ToolConfig( # Add tool_config here
function_calling_config=types.FunctionCallingConfig(
mode=types.FunctionCallingConfigMode.ANY if vertex_tools else types.FunctionCallingConfigMode.NONE
mode=types.FunctionCallingConfigMode.AUTO if vertex_tools else types.FunctionCallingConfigMode.NONE
)
) if vertex_tools else None,
)
@ -419,66 +419,52 @@ class TetoCog(commands.Cog):
# Check for function calls
if candidate.finish_reason == types.FinishReason.FUNCTION_CALL and candidate.content and candidate.content.parts:
has_tool_call = False
try: # Add try-except around tool call processing
for part in candidate.content.parts:
if part.function_call:
has_tool_call = True
function_call = part.function_call
tool_name = function_call.name
tool_args = dict(function_call.args) if function_call.args else {}
print(f"[TETO DEBUG] Vertex AI requested tool: {tool_name} with args: {tool_args}")
# Append model's request to history
vertex_contents.append(candidate.content)
tool_result_str = ""
try: # Inner try-except for tool execution
if tool_name == "execute_shell_command":
command_to_run = tool_args.get("command", "")
if self._is_dangerous_command(command_to_run):
tool_result_str = "❌ Error: Execution was blocked due to a potentially dangerous command."
else:
tool_result_str = await self._execute_shell_command(command_to_run)
elif tool_name == "web_search":
query_to_search = tool_args.get("query", "")
search_api_results = await self.web_search(query=query_to_search)
if "error" in search_api_results:
tool_result_str = f"❌ Error: Web search failed - {search_api_results['error']}"
else:
results_text_parts = []
for i, res_item in enumerate(search_api_results.get("results", [])[:3], 1): # Limit to 3 results for brevity
results_text_parts.append(f"Result {i}:\nTitle: {res_item['title']}\nURL: {res_item['url']}\nContent Snippet: {res_item['content'][:200]}...\n")
if search_api_results.get("answer"):
results_text_parts.append(f"Summary Answer: {search_api_results['answer']}")
tool_result_str = "\n\n".join(results_text_parts)
if not tool_result_str:
tool_result_str = "No results found or summary available."
else:
tool_result_str = f"Error: Unknown tool '{tool_name}' requested."
except Exception as tool_exec_e:
tool_result_str = f"❌ Error during tool execution ({tool_name}): {tool_exec_e}"
print(f"[TETO DEBUG] Exception during tool execution: {tool_exec_e}")
for part in candidate.content.parts:
if part.function_call:
has_tool_call = True
function_call = part.function_call
tool_name = function_call.name
tool_args = dict(function_call.args) if function_call.args else {}
print(f"[TETO DEBUG] Vertex AI requested tool: {tool_name} with args: {tool_args}")
# Append model's request to history
vertex_contents.append(candidate.content)
tool_result_str = ""
if tool_name == "execute_shell_command":
command_to_run = tool_args.get("command", "")
if self._is_dangerous_command(command_to_run):
tool_result_str = "❌ Error: Execution was blocked due to a potentially dangerous command."
else:
tool_result_str = await self._execute_shell_command(command_to_run)
elif tool_name == "web_search":
query_to_search = tool_args.get("query", "")
search_api_results = await self.web_search(query=query_to_search)
if "error" in search_api_results:
tool_result_str = f"❌ Error: Web search failed - {search_api_results['error']}"
else:
results_text_parts = []
for i, res_item in enumerate(search_api_results.get("results", [])[:3], 1): # Limit to 3 results for brevity
results_text_parts.append(f"Result {i}:\nTitle: {res_item['title']}\nURL: {res_item['url']}\nContent Snippet: {res_item['content'][:200]}...\n")
if search_api_results.get("answer"):
results_text_parts.append(f"Summary Answer: {search_api_results['answer']}")
tool_result_str = "\n\n".join(results_text_parts)
if not tool_result_str:
tool_result_str = "No results found or summary available."
else:
tool_result_str = f"Error: Unknown tool '{tool_name}' requested."
# Append tool response to history
vertex_contents.append(types.Content(
role="function", # "tool" role was for older versions, "function" is current for Gemini
parts=[types.Part.from_function_response(name=tool_name, response={"result": tool_result_str})]
))
tool_calls_made += 1
break # Re-evaluate with new history
if has_tool_call:
continue # Continue the while loop for next API call
else:
# Model indicated FUNCTION_CALL but no valid function_call part was found
print("[TETO DEBUG] Model indicated FUNCTION_CALL but no valid function_call part was found.")
return "(Teto AI tried to call a tool but the tool call was malformed.)"
except Exception as tool_parse_e:
print(f"[TETO DEBUG] Error parsing tool call from Vertex AI response: {tool_parse_e}")
return f"(Teto AI encountered an error processing a tool call: {tool_parse_e})"
# Append tool response to history
vertex_contents.append(types.Content(
role="function", # "tool" role was for older versions, "function" is current for Gemini
parts=[types.Part.from_function_response(name=tool_name, response={"result": tool_result_str})]
))
tool_calls_made += 1
break # Re-evaluate with new history
if has_tool_call:
continue # Continue the while loop for next API call
# If no function call or loop finished
final_ai_text_response = _get_response_text(response)