Class GenerateAnswer

AskAgent/python/methods/generate_answer.py:31–236

class GenerateAnswer(NLWebHandler):

    GATHER_ITEMS_THRESHOLD = 55

    RANKING_PROMPT_NAME = "RankingPromptForGenerate"
    SYNTHESIZE_PROMPT_NAME = "SynthesizePromptForGenerate"
    DESCRIPTION_PROMPT_NAME = "DescriptionPromptForGenerate"

    def __init__(self, query_params, handler):
        super().__init__(query_params, handler)
        self.items = []
        self._results_lock = asyncio.Lock()  # Add lock for thread-safe operations
        logger.info(f"GenerateAnswer initialized with query_params: {query_params}")
        log(f"GenerateAnswer query_params: {query_params}")

    async def runQuery(self):
        try:
            logger.info(f"Starting query execution for conversation_id: {self.conversation_id}")
            await self.prepare()
            if (self.query_done):
                logger.info("Query done prematurely")
                return self.return_value
            await self.get_ranked_answers()
            self.return_value["conversation_id"] = self.conversation_id
            logger.info(f"Query execution completed for conversation_id: {self.conversation_id}")
            return self.return_value
        except Exception as e:
            logger.exception(f"Error in runQuery: {e}")
            traceback.print_exc()
            raise

    async def prepare(self):
        # runs the tasks that need to be done before retrieval, ranking, etc.
        logger.info("Starting preparation phase")
        tasks = []

        # Adding all necessary preparation tasks
        tasks.append(asyncio.create_task(analyze_query.DetectItemType(self).do()))
        tasks.append(asyncio.create_task(self.decontextualizeQuery().do()))
        tasks.append(asyncio.create_task(relevance_detection.RelevanceDetection(self).do()))
        tasks.append(asyncio.create_task(memory.Memory(self).do()))
        tasks.append(asyncio.create_task(required_info.RequiredInfo(self).do()))

        try:
            logger.debug(f"Running {len(tasks)} preparation tasks concurrently")
            await asyncio.gather(*tasks, return_exceptions=True)
        except Exception as e:
            logger.exception(f"Error during preparation tasks: {e}")
        finally:
            self.pre_checks_done_event.set()  # Signal completion regardless of errors
            self.state.set_pre_checks_done()

        logger.info("Preparation phase completed")

    async def rankItem(self, url, json_str, name, site):
        if not self.connection_alive_event.is_set():
            logger.warning("Connection lost, skipping item ranking")
            return