Method execute

nextpy/ai/engine/_program.py:599–645

Execute the current program. Note that as execution progresses the program will be incrementally converted from a template into a completed string (with variables stored). At each point in this process the current template remains valid.

(self, memory=None, from_agent=False)

        self._displayed = True

    async def execute(self, memory=None, from_agent=False):
        """Execute the current program.

        Note that as execution progresses the program will be incrementally converted
        from a template into a completed string (with variables stored). At each point
        in this process the current template remains valid.
        """

        log.debug(
            f"Executing program (self.async_mode={self.async_mode}, self.silent={self.silent}, self._displaying_html={self._displaying_html})"
        )

        # if we are already displaying html, we need to yield to the event loop so the jupyter comm can initialize
        if self._displaying_html:
            await asyncio.sleep(0)

        # run the program and capture the output
        try:
            if self.llm is None:
                await self._executor.run(None)
            else:
                with self.llm.session(asynchronous=True) as llm_session:
                    await self._executor.run(llm_session)
            self._text = self._variables["@raw_prefix"]

        # if the execution failed, capture the exception so it can be re-raised
        # in the main coroutine
        except Exception as exception:
            self._exception = exception

        finally:
            # delete the executor and so mark the program as not executing
            self._executor = None

            # update the display with the final output
            self.update_display(last=True)
            await self.update_display.done()

            # Handle memory updates here in case async_mode=True
            if not from_agent and memory is not None:
                all_text = extract_text(self.text)
                for text_block in all_text:
                    for value in text_block:
                        memory.add_memory(prompt=value, llm_response=text_block[value])

            # fire an event noting that execution is complete (this will release any await calls waiting on the program)
            self._execute_complete.set()

    def __getitem__(self, key):
        return self._variables[key]