# Create a temporary directory
        self.temp_dir = tempfile.mkdtemp()
    def _reset(self) -> None:
        for k, v in self.metrics.items():
            self.metrics[k] = 0
[docs]    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> None:
        """Run when LLM starts."""
        self.metrics["step"] += 1
        self.metrics["llm_starts"] += 1
        self.metrics["starts"] += 1
        llm_starts = self.metrics["llm_starts"]
        resp: Dict[str, Any] = {}
        resp.update({"action": "on_llm_start"})
        resp.update(flatten_dict(serialized))
        resp.update(self.metrics)
        for idx, prompt in enumerate(prompts):
            prompt_resp = deepcopy(resp)
            prompt_resp["prompt"] = prompt
            self.jsonf(
                prompt_resp,
                self.temp_dir,
                f"llm_start_{llm_starts}_prompt_{idx}",
            )
[docs]    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        """Run when LLM generates a new token."""
        self.metrics["step"] += 1
        self.metrics["llm_streams"] += 1
        llm_streams = self.metrics["llm_streams"]
        resp: Dict[str, Any] = {}
        resp.update({"action": "on_llm_new_token", "token": token})
        resp.update(self.metrics)
        self.jsonf(resp, self.temp_dir, f"llm_new_tokens_{llm_streams}")