69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
import json
|
|
from datetime import datetime
|
|
|
|
|
|
def _ts():
|
|
return datetime.now().strftime('%H:%M:%S')
|
|
|
|
|
|
def execute_tool(tool_call, TOOL_HANDLERS):
|
|
tname = tool_call['function']['name']
|
|
targs = json.loads(tool_call['function']['arguments'])
|
|
handler = TOOL_HANDLERS.get(tname)
|
|
if not handler:
|
|
return f'Tool {tname} not found'
|
|
try:
|
|
if tname == 'search_code':
|
|
return handler(
|
|
pattern=targs['pattern'],
|
|
search_type=targs['search_type'],
|
|
path=targs.get('path', '.'),
|
|
)
|
|
elif tname == 'git_operation':
|
|
return handler(args=targs['args'])
|
|
else:
|
|
return handler(**targs)
|
|
except Exception as e:
|
|
return f'Error executing tool: {str(e)}'
|
|
|
|
|
|
def run_agent_loop(session, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, on_tool_calls=None):
|
|
for step in range(max_iterations):
|
|
print(f'[{_ts()}] Step {step + 1} — calling LLM...')
|
|
response = llm_client.chat(session.messages, tools=TOOLS)
|
|
|
|
if response.tool_calls:
|
|
amsg = {
|
|
'role': 'assistant',
|
|
'content': response.content,
|
|
'tool_calls': response.tool_calls,
|
|
}
|
|
session.messages.append(amsg)
|
|
|
|
tnames = [tc['function']['name'] for tc in response.tool_calls]
|
|
print(f'[{_ts()}] Using tools: {", ".join(tnames)}')
|
|
|
|
if on_tool_calls:
|
|
on_tool_calls(tnames)
|
|
|
|
for tc in response.tool_calls:
|
|
result = execute_tool(tc, TOOL_HANDLERS)
|
|
session.messages.append({
|
|
'role': 'tool',
|
|
'tool_call_id': tc['id'],
|
|
'content': str(result),
|
|
})
|
|
else:
|
|
if response.content:
|
|
print(f'[{_ts()}] Response generated ({len(response.content)} chars)')
|
|
session.messages.append({'role': 'assistant', 'content': response.content})
|
|
return response.content
|
|
return None
|
|
|
|
print(f'[{_ts()}] Max iterations ({max_iterations}) reached')
|
|
session.messages.append({
|
|
'role': 'assistant',
|
|
'content': 'Max iterations reached without final answer.',
|
|
})
|
|
return None
|