"""Agent loop terpusat — satu engine untuk semua interface (TUI, GUI, service). Dua boundary pemisah, supaya engine tidak pernah mengenal interface: SessionStore → ke mana pesan percakapan ditulis (list aktif + persist opsional) AgentObserver → siapa yang melihat prosesnya (log, streaming, timing) Engine sengaja tetap sync (semua handler tool saat ini sync; MCP masih jauh). """ import json import config from lib.personality import PERSONALITY from tools.vision import ImagePayload from scripts import loggorilla def _cap_output(result): 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)}') # --------------------------------------------------------------- contracts --- class SessionStore: """Tujuan tulis untuk setiap pesan percakapan selama satu loop berjalan. `messages` state aktif yang dikirim ke LLM (list dict). `on_append` dipanggil setelah tiap pesan masuk — untuk persistensi (TUI/GUI/API via NeoSessionManager). Boleh None. `keep_model_info` False (default): kunci `model_info` dibuang sebelum pesan masuk list, jadi payload ke provider identik dengan sebelum refactor. Interface yang menampilkan info model menyalakannya. """ def __init__(self, messages, on_append=None, keep_model_info=False): self.messages = messages self._on_append = on_append self.keep_model_info = keep_model_info def append(self, message): if not self.keep_model_info: message.pop('model_info', None) message.pop('usage', None) # metadata display, bukan payload LLM self.messages.append(message) if self._on_append: self._on_append(message) class AgentResult: # Hasil satu kali jalan loop. def __init__(self, content=None, session_ended=False, model_info=None, warning=None, is_multimodal=False, usage=None): self.content = content # str | None self.session_ended = session_ended # tool end_session pernah dipanggil self.model_info = model_info or {} # model yang benar-benar menjawab self.warning = warning # mis. fallback chain model self.is_multimodal = is_multimodal self.usage = usage # token turn ini: prompt/completion/total (dict | None) class AgentObserver: """Hook tampilan. Semua no-op — subclass cukup menimpa yang dibutuhkan. Engine tidak pernah mencetak atau menyentuh UI; seluruh efek visual hidup di sini, sehingga interface baru tinggal membuat subclass sendiri. """ # True bila observer membutuhkan delta streaming dari LLM client. wants_streaming = False def on_run_start(self): pass def on_run_end(self): pass def on_step_start(self, step): pass def on_response(self, model_info, warning, usage=None): """llm.chat() sudah kembali, sebelum engine memproses response. `usage` : token pemakaian satu permintaan (prompt/completion/total), None jika provider tidak mengirimnya.""" pass def on_step_end(self, step): """Satu iterasi selesai dan loop berlanjut ke iterasi berikutnya.""" pass def on_stream_chunk(self, chunk): pass def on_tool_calls(self, content, names): """Assistant menjadwalkan tool call; `content` narasi sebelumnya.""" pass def on_tool_call(self, name, arguments): pass def on_final(self, result): pass def on_max_iterations(self, max_iterations=None): pass class ConsoleObserver(AgentObserver): """Observer default proses headless (service bot): log gorilla ke stdout. `on_tool_calls` mengirim narasi assistant sebelum tool dijalankan — dipakai Telegram/XMPP untuk mengirim pesan interim ke chat. """ def __init__(self, on_tool_calls=None): self._on_tool_calls = on_tool_calls def on_step_start(self, step): loggorilla.prcss("agent_loop", f"Step {step + 1} — calling LLM...") def on_tool_calls(self, content, names): loggorilla.fyinf("agent_loop", f"Using tools: {', '.join(names)}") if self._on_tool_calls: self._on_tool_calls(content) def on_final(self, result): # Content None = respons final kosong (mis. sesudah end_session): # tidak ada yang dicetak, sama seperti sebelumnya. if result.content is None: return if result.is_multimodal: loggorilla.prcss("agent_loop", "Response generated (multimodal)") else: loggorilla.prcss("agent_loop", f"Response generated ({len(result.content)} chars)") def on_max_iterations(self, max_iterations=None): loggorilla.prcss("agent_loop", f"Max iterations ({max_iterations}) reached") # ------------------------------------------------------------------- loop --- def run_agent_loop(store, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, observer=None, tool_reminder=None, disable_reasoning=None): """Jalankan loop LLM <-> tool sampai ada jawaban final atau batas iterasi. `store` : SessionStore (bukan list mentah — biar persistensi eksplisit). `observer` : None → ConsoleObserver (headless, tanpa streaming). `disable_reasoning` None → ikut PERSONALITY (yaml karakter). Return AgentResult. Engine tidak mengklasifikasikan konten sebagai error; itu keputusan interface (lihat CursesObserver di interfaces/tui/agent.py). """ if observer is None: observer = ConsoleObserver() if disable_reasoning is None: disable_reasoning = PERSONALITY.disable_reasoning session_ended = False model_info = {} warning = None stream_cb = observer.on_stream_chunk if observer.wants_streaming else None # Akumulator token satu turn (semua langkah LLM dijumlahkan — tiap # panggilan mengirim ulang konteks sehingga dihitung ulang oleh provider). any_usage = False turn_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} def _accumulate(u): nonlocal any_usage if not isinstance(u, dict): return any_usage = True for k in ("prompt_tokens", "completion_tokens", "total_tokens"): v = u.get(k) if isinstance(v, (int, float)): turn_usage[k] += int(v) # Ukuran context window saat itu = prompt tokens panggilan terakhir. pt = u.get("prompt_tokens") if isinstance(pt, (int, float)): turn_usage["context_tokens"] = int(pt) observer.on_run_start() for step in range(max_iterations): observer.on_step_start(step) response = llm_client.chat( store.messages, tools=TOOLS, on_stream_chunk=stream_cb, disable_reasoning=disable_reasoning, tool_reminder=tool_reminder, ) # Model final diambil dari response (bukan state llm saat ini) supaya # auto-switch / rotasi key / fallback tercatat dengan benar. model_info = getattr(response, "model_info", None) or {} warning = getattr(response, "warning", None) usage = getattr(response, "usage", None) _accumulate(usage) observer.on_response(model_info, warning, usage) if response.tool_calls: amsg = { 'role': 'assistant', 'content': response.content, 'tool_calls': response.tool_calls, } if model_info: amsg['model_info'] = model_info if isinstance(usage, dict) and usage: # Usage per langkah (bukan akumulasi turn) — dipakai interface # untuk menampilkan token di blok tool call yang dihasilkannya. # Hanya metadata display: dibuang sebelum jadi payload LLM. amsg['usage'] = dict(usage) store.append(amsg) observer.on_tool_calls(response.content or "", [tc['function']['name'] for tc in response.tool_calls]) for tc in response.tool_calls: observer.on_tool_call(tc['function']['name'], tc['function']['arguments']) if tc['function']['name'] == 'end_session': session_ended = True result = execute_tool(tc, TOOL_HANDLERS) if isinstance(result, ImagePayload): store.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: store.append({ 'role': 'tool', 'tool_call_id': tc['id'], 'content': str(result), }) observer.on_step_end(step) continue has_content = bool(response.content) or (isinstance(response.content, list) and len(response.content) > 0) if has_content: is_multimodal = isinstance(response.content, list) msg = {'role': 'assistant', 'content': response.content} if model_info: msg['model_info'] = model_info if any_usage: # Total token turn ini — ikut dipersist supaya tampil saat # session dibuka kembali. msg['usage'] = dict(turn_usage) store.append(msg) if is_multimodal: 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]" else: text = response.content final = AgentResult( content=text, session_ended=session_ended, model_info=model_info, warning=warning, is_multimodal=is_multimodal, usage=dict(turn_usage) if any_usage else None, ) observer.on_final(final) observer.on_run_end() return final # Respons kosong: tetap bawa flag session_ended — bisa jadi # end_session sudah dipanggil di iterasi sebelumnya. empty = AgentResult(content=None, session_ended=session_ended, model_info=model_info, warning=warning) observer.on_final(empty) observer.on_run_end() return empty store.append({ 'role': 'assistant', 'content': 'Max iterations reached without final answer.', }) observer.on_max_iterations(max_iterations) observer.on_run_end() return AgentResult(content=None, session_ended=session_ended, model_info=model_info, warning=warning)