Method query

methods/letta_agent.py:560–636

Query using Letta's official agent API. Sends the question to the Letta agent which will use its internal archival_memory_search tool to retrieve relevant memories and respond. Question is truncated if exceeding max_question_tokens.

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

        )

    def query(
        self,
        question: str,
        system_message: Optional[str] = None,
        **kwargs,
    ) -> AgentResponse:
        """Query using Letta's official agent API.

        Sends the question to the Letta agent which will use its internal
        archival_memory_search tool to retrieve relevant memories and respond.

        Question is truncated if exceeding max_question_tokens.
        """
        context_id = self._get_context_id()
        agent_id = self._ensure_agent(context_id)

        # Truncate question if needed
        bounded_question = self._truncate_to_tokens(question, self.max_question_tokens)

        # Build query message
        if system_message:
            query_message = f"{system_message}\n\nQuestion: {bounded_question}"
        else:
            query_message = bounded_question

        start_time = time.time()
        try:
            # Switch to restricted context_window for query phase (token budget truncation)
            # Only update once per agent (build phase is always completed before query phase)
            if self.max_context_tokens and self.max_context_tokens < self.context_window:
                if not getattr(self, '_query_context_applied', {}).get(agent_id):
                    query_llm_config = self._build_llm_config().model_copy(
                        update={"context_window": self.max_context_tokens}
                    )
                    self._client.update_agent(agent_id, llm_config=query_llm_config)
                    if not hasattr(self, '_query_context_applied'):
                        self._query_context_applied = {}
                    self._query_context_applied[agent_id] = True

            response = self._client.user_message(agent_id=agent_id, message=query_message)
            query_time = time.time() - start_time
        except Exception as e:
            logger.error(f"Letta user_message failed: {e}")
            return AgentResponse(
                output=f"[Error: {e}]",
                query_time=0.0,
                retrieved_count=0,
                extra={
                    "method": "letta",
                    "agent_id": agent_id,
                    "error": str(e),
                },
            )

        parsed = self._parse_letta_response(response)
        self._record_usage_to_tracker(parsed, "query", query_time)

        # Extract retrieved memories from search function calls