Better modular execute_tool for TUI agent

This commit is contained in:
Dita Aji Pratama 2026-06-22 11:47:29 +07:00
parent 7aa1f56124
commit fb52623d3e
2 changed files with 28 additions and 38 deletions

View File

@ -4,25 +4,32 @@ from datetime import datetime
def _ts(): def _ts():
return datetime.now().strftime('%H:%M:%S') return datetime.now().strftime('%H:%M:%S')
def execute_tool(tool_call, TOOL_HANDLERS): def execute_tool(tool_call, TOOL_HANDLERS, app=None):
tname = tool_call['function']['name'] tname = tool_call['function']['name']
targs = json.loads(tool_call['function']['arguments']) targs = json.loads(tool_call['function']['arguments'])
handler = TOOL_HANDLERS.get(tname) handler = TOOL_HANDLERS.get(tname)
if not handler: if not handler:
return f'Tool {tname} not found' result = f'Tool {tname} not found'
try: else:
if tname == 'search_code': try:
return handler( if tname == 'search_code':
pattern=targs['pattern'], result = handler(
search_type=targs['search_type'], pattern=targs['pattern'],
path=targs.get('path', '.'), search_type=targs['search_type'],
) path=targs.get('path', '.'),
elif tname == 'git_operation': )
return handler(args=targs['args']) elif tname == 'git_operation':
else: result = handler(args=targs['args'])
return handler(**targs) else:
except Exception as e: result = handler(**targs)
return f'Error executing tool: {str(e)}' except Exception as e:
result = f'Error executing tool: {str(e)}'
if app:
from tui.agent import _add_msg
_add_msg(app, "tool", str(result), tool_call_id=tool_call["id"])
return result
def run_agent_loop(session, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, on_tool_calls=None): def run_agent_loop(session, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, on_tool_calls=None):
for step in range(max_iterations): for step in range(max_iterations):

View File

@ -2,7 +2,8 @@ import json
import threading import threading
from datetime import datetime from datetime import datetime
import config import config
from scripts import ntro from scripts import ntro, agent_loop
def _add_msg(app, role, content, **kwargs): def _add_msg(app, role, content, **kwargs):
msg = {"role": role, "content": content} msg = {"role": role, "content": content}
@ -150,7 +151,8 @@ def _agent_loop(app):
"arguments": targs, "arguments": targs,
})) }))
app.scroll = 999999 app.scroll = 999999
execute_tool(app, tc) agent_loop.execute_tool(tc, app.TOOL_HANDLERS, app=app)
# Log content AI setelah tools (jika ada) # Log content AI setelah tools (jika ada)
if response.content and response.content.strip(): if response.content and response.content.strip():
@ -171,25 +173,6 @@ def _agent_loop(app):
app.agent_done.set() app.agent_done.set()
def execute_tool(app, tool_call): app.agent_done.set()
tname = tool_call["function"]["name"] ntro.end(stamp)
targs = json.loads(tool_call["function"]["arguments"])
handler = app.TOOL_HANDLERS.get(tname)
if not handler:
result = f"Tool {tname} not found"
else:
try:
if tname == "search_code":
result = handler(
pattern=targs["pattern"],
search_type=targs["search_type"],
path=targs.get("path", "."),
)
elif tname == "git_operation":
result = handler(args=targs["args"])
else:
result = handler(**targs)
except Exception as e:
result = f"Error executing tool: {str(e)}"
_add_msg(app, "tool", str(result), tool_call_id=tool_call["id"])