1  from typing import Dict, Any, List
   2  from modules.llm.ollama_client import query_ollama
   3  from ..tools.tool_executor import execute_tool
   4  
   5  class Planner:
   6      def __init__(self):
   7          self.chat_history = []
   8          self.context = {}
   9  
  10      def receive_input(self, user_input: str) -> None:
  11          self._update_history(role='user', content=user_input)
  12  
  13      async def process(self) -> str:
  14          max_cycles = 5
  15          for _ in range(max_cycles):
  16              llm_response = await query_ollama(
  17                  prompt=self.chat_history,
  18                  context=self.context,
  19                  agent_type='planner'
  20              )
  21              
  22              if '[ToolCall:' in llm_response:
  23                  tool_result = await self._handle_tool_call(llm_response)
  24                  self._update_context(tool_result)
  25              else:
  26                  self._update_history(role='assistant', content=llm_response)
  27                  return llm_response
  28          return "Maximum processing cycles reached"
  29  
  30      async def _handle_tool_call(self, response: str) -> Dict[str, Any]:
  31          tool_call = response.split('[ToolCall:')[1].split(']')[0].strip()
  32          tool_name, args = self._parse_tool_call(tool_call)
  33          result = await execute_tool(tool_name, args)
  34          self._update_history(role='tool', content=str(result))
  35          return result
  36  
  37      def _parse_tool_call(self, tool_call: str) -> tuple:
  38          return tool_call.split('(', 1)[0], eval(tool_call.split('(', 1)[1].rstrip(')'))
  39  
  40      def _update_history(self, role: str, content: str) -> None:
  41          self.chat_history.append({'role': role, 'content': content})
  42  
  43      def _update_context(self, tool_data: Dict[str, Any]) -> None:
  44          self.context.update(tool_data)