107 lines
4.5 KiB
Python
107 lines
4.5 KiB
Python
import json
|
|
from datetime import datetime
|
|
import config
|
|
from lib.personality import PERSONALITY
|
|
from tools.vision import ImagePayload
|
|
from scripts import loggorilla
|
|
|
|
def _ts():
|
|
return datetime.now().strftime('%H:%M:%S')
|
|
|
|
def _cap_output(result):
|
|
# Jaring pengaman ukuran output tool. Hormati semantics config:
|
|
# AGENT_MAX_TOOL_OUTPUT <= 0 = unlimited (tidak dipotong).
|
|
limit = getattr(config, "AGENT_MAX_TOOL_OUTPUT", 0)
|
|
if not limit or limit <= 0:
|
|
return result
|
|
if isinstance(result, str) and len(result) > limit:
|
|
omitted = len(result) - limit
|
|
return result[:limit] + f"\n... [truncated: {omitted} chars omitted]"
|
|
return result
|
|
|
|
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 _cap_output(f'Tool {tname} not found')
|
|
try:
|
|
if tname == 'search_code':
|
|
return _cap_output(handler(
|
|
pattern=targs['pattern'],
|
|
search_type=targs['search_type'],
|
|
path=targs.get('path', '.'),
|
|
))
|
|
elif tname == 'git_operation':
|
|
return _cap_output(handler(args=targs['args']))
|
|
else:
|
|
return _cap_output(handler(**targs))
|
|
except Exception as e:
|
|
return _cap_output(f'Error executing tool: {str(e)}')
|
|
|
|
def run_agent_loop(session, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, on_tool_calls=None, tool_reminder=None):
|
|
session_ended = False
|
|
|
|
for step in range(max_iterations):
|
|
loggorilla.prcss("agent_loop", f"Step {step + 1} — calling LLM...")
|
|
|
|
disable_reasoning = PERSONALITY.disable_reasoning
|
|
|
|
response = llm_client.chat(session.messages, tools=TOOLS, disable_reasoning=disable_reasoning, tool_reminder=tool_reminder)
|
|
|
|
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]
|
|
loggorilla.fyinf("agent_loop", f"Using tools: {', '.join(tnames)}")
|
|
|
|
if on_tool_calls:
|
|
on_tool_calls(response.content or "")
|
|
|
|
for tc in response.tool_calls:
|
|
if tc['function']['name'] == 'end_session':
|
|
session_ended = True
|
|
result = execute_tool(tc, TOOL_HANDLERS)
|
|
if isinstance(result, ImagePayload):
|
|
session.messages.append({
|
|
'role': 'tool',
|
|
'tool_call_id': tc['id'],
|
|
'content': [
|
|
{"type": "text", "text": f"[Image loaded: {result.path}]"},
|
|
{"type": "image_url", "image_url": {"url": f"data:{result.mime};base64,{result.data}"}}
|
|
]
|
|
})
|
|
else:
|
|
session.messages.append({
|
|
'role': 'tool',
|
|
'tool_call_id': tc['id'],
|
|
'content': str(result),
|
|
})
|
|
else:
|
|
has_content = bool(response.content) or (isinstance(response.content, list) and len(response.content) > 0)
|
|
if has_content:
|
|
session.messages.append({'role': 'assistant', 'content': response.content})
|
|
if isinstance(response.content, list):
|
|
text_parts = [b.get('text', '') for b in response.content
|
|
if isinstance(b, dict) and b.get('type') == 'text']
|
|
text = '\n'.join(text_parts) if text_parts else "[Image generated]"
|
|
loggorilla.prcss("agent_loop", "Response generated (multimodal)")
|
|
return text, session_ended
|
|
loggorilla.prcss("agent_loop", f"Response generated ({len(response.content)} chars)")
|
|
return response.content, session_ended
|
|
# Respons kosong: tetap bawa flag session_ended — bisa jadi
|
|
# end_session sudah dipanggil di iterasi sebelumnya.
|
|
return None, session_ended
|
|
|
|
loggorilla.prcss("agent_loop", f"Max iterations ({max_iterations}) reached")
|
|
session.messages.append({
|
|
'role': 'assistant',
|
|
'content': 'Max iterations reached without final answer.',
|
|
})
|
|
return None, session_ended
|