refactor(memory): handle varied memory search and retrieval formats
Some checks failed
Build and Push Docker / build-and-push (push) Failing after 17m18s
Some checks failed
Build and Push Docker / build-and-push (push) Failing after 17m18s
This commit is contained in:
@ -38,10 +38,19 @@ In your response, consider the memories above to provide a personalized answer."
|
||||
)
|
||||
# Debug: Print the actual format of results
|
||||
print(f"[DEBUG] Search results type: {type(results)}")
|
||||
if results:
|
||||
print(f"[DEBUG] First result type: {type(results[0])}")
|
||||
print(f"[DEBUG] First result: {results[0]}")
|
||||
return results
|
||||
if isinstance(results, dict):
|
||||
# Handle dictionary response format (based on official docs)
|
||||
memories = results.get("memories", [])
|
||||
if memories:
|
||||
print(f"[DEBUG] First memory type: {type(memories[0])}")
|
||||
print(f"[DEBUG] First memory: {memories[0]}")
|
||||
return memories
|
||||
elif isinstance(results, list):
|
||||
# Handle list response format
|
||||
return results
|
||||
else:
|
||||
print(f"[WARNING] Unexpected search results format: {type(results)}")
|
||||
return []
|
||||
except Exception as e:
|
||||
print(f"[ERROR] Failed to search memories: {e}")
|
||||
return []
|
||||
@ -90,8 +99,11 @@ In your response, consider the memories above to provide a personalized answer."
|
||||
# Step 1: Search for relevant memories
|
||||
memories = self.search_memories(user_input, user_id)
|
||||
|
||||
# Step 2: Format memories for the prompt
|
||||
formatted_memories = self.format_memories_for_prompt(memories)
|
||||
# Step 2: Format memories for the prompt (or use empty if no memories)
|
||||
if memories:
|
||||
formatted_memories = self.format_memories_for_prompt(memories)
|
||||
else:
|
||||
formatted_memories = "No previous memories about this user."
|
||||
|
||||
# Step 3: Create the enhanced prompt
|
||||
enhanced_prompt = self.memory_template.format(
|
||||
@ -139,16 +151,25 @@ In your response, consider the memories above to provide a personalized answer."
|
||||
"user_id": user_id
|
||||
}
|
||||
|
||||
def get_all_memories(self, user_id: str) -> List[Dict[str, Any]]:
|
||||
def get_all_memories(self, user_id: str) -> List[Any]:
|
||||
"""Get all memories for a user."""
|
||||
try:
|
||||
memories = self.memory.get_all(user_id=user_id)
|
||||
# Debug: Print the actual format of memories
|
||||
print(f"[DEBUG] All memories type: {type(memories)}")
|
||||
if memories:
|
||||
print(f"[DEBUG] First memory type: {type(memories[0])}")
|
||||
print(f"[DEBUG] First memory: {memories[0]}")
|
||||
return memories
|
||||
if isinstance(memories, dict):
|
||||
# Handle dictionary response format
|
||||
memories_list = memories.get("memories", [])
|
||||
if memories_list:
|
||||
print(f"[DEBUG] First memory type: {type(memories_list[0])}")
|
||||
print(f"[DEBUG] First memory: {memories_list[0]}")
|
||||
return memories_list
|
||||
elif isinstance(memories, list):
|
||||
# Handle list response format
|
||||
return memories
|
||||
else:
|
||||
print(f"[WARNING] Unexpected memories format: {type(memories)}")
|
||||
return []
|
||||
except Exception as e:
|
||||
print(f"[ERROR] Failed to get all memories: {e}")
|
||||
return []
|
||||
|
Reference in New Issue
Block a user