src/memory.py

• MemoryVectorDB · class · L11-L363 — class MemoryVectorDB
• init · method · L12-L35 — def init(self, persistent_dir: str, collection_name: str, embedding_model: EmbeddingModel, tokenizer: Optional[Callable[[str], List[str]]] = word_tokenize, stemmer: Optional[Callable[[str], List[str]]] = SnowballStemmer("english"), verbose: bool = True)
• load_doc_embeddings · method · L41-L54 — def load_doc_embeddings(self, npz_path: str) -> None
• add_memory · method · L56-L80 — def add_memory(self, db_id: str, embedding_text: str, document_text: str, add_meta_datas: Dict = None) -> None
• add_batch_memories · method · L82-L109 — def add_batch_memories(self, db_ids: List[str], embedding_texts: List[str], document_texts: List[str], add_meta_datas: List[Dict] = None) -> None
• query_similar · method · L111-L144 — def query_similar(self, query_text: str, n_results: int = 3, filter_metadata: Dict = None) -> List[Tuple[Dict, str, float, Optional[np.ndarray]]]
• batch_query_similar · method · L146-L189 — def batch_query_similar( self, query_texts: List[str], n_results: int = 3, filter_metadata: Dict = None, ) -> List[List[Tuple[Dict, str, float, Optional[np.ndarray]]]]
• batch_get_random · method · L191-L241 — def batch_get_random( self, n_queries: int, n_results: int = 3, filter_metadata: Dict = None, distance: float = 1.0, seed: Optional[int] = None, ) -> List[List[Tuple[Dict, str, float, Optional[np.ndarray]]]]
• get_by_ids · method · L243-L253 — def get_by_ids(self, select_ids: List[str]) -> List[Tuple[Dict, str]]
• update_memory · method · L255-L281 — def update_memory(self, db_id: str, embedding_text: str, document_text: str, add_meta_datas: Dict = None) -> None
• delete_memory · method · L283-L289 — def delete_memory(self, db_id: str) -> None
• get_memories_count · method · L291-L292 — def get_memories_count(self) -> int
• get_all_ids · method · L294-L297 — def get_all_ids(self, filter_metadata=None)
• _make_resp_id · method · L304-L305 — def _make_resp_id(query_id, round_idx: int, agent_id: str) -> str
• add_batch_responses · method · L307-L317 — def add_batch_responses(self, db_ids: List[str], texts: List[str], meta_datas: List[Dict]) -> None
• get_response_embeddings · method · L319-L339 — def get_response_embeddings(self, query_id, round_idx: int, agent_ids: List[str]) -> Dict[str, Optional[np.ndarray]]
• batch_get_response_embeddings · method · L341-L363 — def batch_get_response_embeddings( self, keys: List[Tuple], # list of (query_id, round_idx, agent_id_str) ) -> Dict[Tuple, Optional[np.ndarray]]