class HybridRetriever:
"""
Retrieve relevant memory through semantic, symbolic, topological paths.
Combines results using Reciprocal Rank Fusion.
"""
def __init__(self, memory_optimizer):
self.memory = memory_optimizer.memory
self.topic_graph = memory_optimizer.topic_graph
self.model = memory_optimizer.model
def macro_semantic_navigation(self, query, k=5):
"""
Pathway 1: Navigate through Topics for high-level localization.
Reduces search space before detailed retrieval.
"""
topic_scores = {
topic: self.model.similarity(query, ' '.join(notes))
for topic, notes in self.topic_graph.items()
}
best_topic = max(topic_scores, key=topic_scores.get)
return self.topic_graph[best_topic][:k]
def micro_symbolic_anchoring(self, query, k=5):
"""
Pathway 2: Extract entities from query and match in keywords.
Provides precise entity-based retrieval.
"""
entities = self.model.extract_entities(query)
relevant_notes = []
for note in self.memory['notes']:
note_entities = self.model.extract_entities(note)
overlap = len(entities & note_entities)
if overlap > 0:
relevant_notes.append((note, overlap))
relevant_notes.sort(key=lambda x: x[1], reverse=True)
return [note for note, _ in relevant_notes[:k]]
def topological_expansion(self, seed_notes, k=5):
"""
Pathway 3: Expand from seed notes through topic associations.
Finds logically related evidence.
"""
expanded = set(seed_notes)
for note in seed_notes:
for topic, notes in self.topic_graph.items():
if note in notes:
expanded.update(notes)
return list(expanded)[:k]
def retrieve_with_fusion(self, query, task_context, k=5):
"""
Combine all three pathways using Reciprocal Rank Fusion.
"""
semantic_results = self.macro_semantic_navigation(query, k)
symbolic_results = self.micro_symbolic_anchoring(query, k)
topological_results = self.topological_expansion(
symbolic_results, k)
rrf_scores = {}
for i, note in enumerate(semantic_results):
rrf_scores[note] = rrf_scores.get(note, 0) + 1/(i+1)
for i, note in enumerate(symbolic_results):
rrf_scores[note] = rrf_scores.get(note, 0) + 1/(i+1)
for i, note in enumerate(topological_results):
rrf_scores[note] = rrf_scores.get(note, 0) + 1/(i+1)
sorted_notes = sorted(
rrf_scores.items(),
key=lambda x: x[1],
reverse=True
)
return [note for note, _ in sorted_notes[:k]]
def iterative_refinement(self, query, task_context, max_rounds=3):
"""
For complex queries, progressively expand evidence.
Stop when sufficient evidence gathered.
"""
evidence = self.retrieve_with_fusion(query, task_context, k=5)
for round_num in range(max_rounds):
sufficiency = self.model.evaluate_evidence_sufficiency(
query, evidence, task_context)
if sufficiency > 0.8:
break
evidence.extend(
self.topological_expansion(evidence, k=3)
)
return evidence
Information bottleneck memory optimization enables efficient long-term memory for agents by compressing redundancy while preserving task-relevant information. The hybrid retrieval system leverages semantic, symbolic, and topological pathways to find evidence across scales of abstraction.
