hendrik/lib/agent_loop.py

import json
from datetime import datetime
from tools.vision import ImagePayload

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, tool_reminder=None):
    session_ended = False

    for step in range(max_iterations):
        print(f'[{_ts()}] Step {step + 1} — calling LLM...', flush=True)
        
        # Ambil konfigurasi disable_reasoning dari personality karakter
        # Default ke False jika tidak didefinisikan
        personality = getattr(session, 'personality', {})
        disable_reasoning = personality.get('disable_reasoning', False)
        
        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]
            print(f'[{_ts()}] Using tools: {", ".join(tnames)}', flush=True)

            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]"
                    print(f'[{_ts()}] Response generated (multimodal)', flush=True)
                    return text, session_ended
                print(f'[{_ts()}] Response generated ({len(response.content)} chars)', flush=True)
                return response.content, session_ended
            return None, False

    print(f'[{_ts()}] Max iterations ({max_iterations}) reached', flush=True)
    session.messages.append({
        'role': 'assistant',
        'content': 'Max iterations reached without final answer.',
    })
    return None, False