Method retrieve

runtime/node/agent/memory/file_memory.py:97–162

Retrieve relevant file chunks based on query. Args: agent_role: Agent role (not used in file memory) inputs: Query text top_k: Number of results to return similarity_threshold: Minimum similarity score Returns: Li

(
        self,
        agent_role: str,
        query: MemoryContentSnapshot,
        top_k: int,
        similarity_threshold: float,
    )

        logger.info(f"Index saved to {self.index_path} ({len(self.contents)} chunks)")

    def retrieve(
        self,
        agent_role: str,
        query: MemoryContentSnapshot,
        top_k: int,
        similarity_threshold: float,
    ) -> List[MemoryItem]:
        """
        Retrieve relevant file chunks based on query.

        Args:
            agent_role: Agent role (not used in file memory)
            inputs: Query text
            top_k: Number of results to return
            similarity_threshold: Minimum similarity score

        Returns:
            List of MemoryItem with file chunks
        """
        if self.count_memories() == 0:
            return []

        # Generate query embedding
        query_embedding = self.embedding.get_embedding(query.text)
        if isinstance(query_embedding, list):
            query_embedding = np.array(query_embedding, dtype=np.float32)
        query_embedding = query_embedding.reshape(1, -1)
        faiss.normalize_L2(query_embedding)

        expected_dim = query_embedding.shape[1]

        # Collect embeddings from memory items
        memory_embeddings = []
        valid_items = []
        for item in self.contents:
            if item.embedding is not None:
                if len(item.embedding) != expected_dim:
                    logger.warning(
                        "Skipping memory item %s: embedding dim %d != expected %d",
                        item.id, len(item.embedding), expected_dim,
                    )
                    continue
                memory_embeddings.append(item.embedding)
                valid_items.append(item)

        if not memory_embeddings:
            return []

        memory_embeddings = np.array(memory_embeddings, dtype=np.float32)

        # Build FAISS index and search
        index = faiss.IndexFlatIP(memory_embeddings.shape[1])
        index.add(memory_embeddings)

        similarities, indices = index.search(query_embedding, min(top_k, len(valid_items)))

        # Filter by threshold and return results
        results = []