extensions/python/reasoning_stream/_10_log_from_stream.py

from helpers import persist_chat, tokens
from helpers.extension import Extension
from agent import LoopData
import asyncio
from helpers.log import LogItem
from helpers import log
import math
from extensions.python.before_main_llm_call._10_log_for_stream import build_heading, build_default_heading

class LogFromStream(Extension):

    async def execute(self, loop_data: LoopData = LoopData(), text: str = "", **kwargs):
        if not self.agent:
            return

        # thought length indicator
        length = f"({len(text)})" if text else ""
        pipes = "|" * math.ceil(math.sqrt(len(text))/2)
        heading = build_heading(self.agent, f"Reasoning... {pipes}")
        step = f"Reasoning... {length}"

        # create log message and store it in loop data temporary params
        if "log_item_generating" not in loop_data.params_temporary:
            loop_data.params_temporary["log_item_generating"] = (
                self.agent.context.log.log(
                    type="agent",
                    heading=heading,
                    step=step
                )
            )

        # update log message
        log_item = loop_data.params_temporary["log_item_generating"]
        log_item.update(heading=heading, reasoning=text, step=step)