Method query

methods/mem0_agent.py:342–406

Query the agent.

(
        self,
        question: str,
        system_message: Optional[str] = None,
        **kwargs
    )

        return memories

    def query(
        self,
        question: str,
        system_message: Optional[str] = None,
        **kwargs
    ) -> AgentResponse:
        """Query the agent."""
        # Retrieve relevant memories
        retrieved_memories = self._retrieve(question)

        # Bound question tokens first
        full_question = f"{question}\n\nCurrent Time: {time.strftime('%Y-%m-%d %H:%M:%S')}"
        full_question = self._truncate_to_tokens(full_question, self.max_question_tokens)

        base_system = system_message or ""
        reserved_tokens = self.max_tokens + 400
        available_tokens = max(self.max_context_tokens - reserved_tokens, 0)
        question_tokens = self._llm_client.count_tokens(full_question)
        base_system_tokens = self._llm_client.count_tokens(base_system) if base_system else 0
        memory_budget = max(available_tokens - question_tokens - base_system_tokens, 0)

        # Build memory string
        if retrieved_memories:
            memory_lines: List[str] = []
            used_tokens = 0
            for entry in retrieved_memories:
                line = f"- {entry['memory']}"
                line_tokens = self._llm_client.count_tokens(line)
                if used_tokens + line_tokens <= memory_budget:
                    memory_lines.append(line)
                    used_tokens += line_tokens
                else:
                    break
            memories_str = "\n".join(memory_lines)
            # Build system message with retrieved content
            memory_prompt = f"You are a helpful AI. Answer the question based on the following memories:\n{memories_str}\n"
            if system_message:
                full_system = f"{system_message}\n\n{memory_prompt}"
            else:
                full_system = memory_prompt
        else:
            full_system = system_message

        messages = format_messages(full_question, full_system)

        # Call LLM
        response = self._llm_client.chat(messages)

        # Build retrieved_memories format
        formatted_memories = [
            {"memory": m["memory"], "type": "mem0_retrieval", "score": m.get("score", 0)}
            for m in retrieved_memories
        ]

        return AgentResponse(
            output=response.content,
            query_time=0.0,
            retrieved_count=len(retrieved_memories),