什么是智能体记忆？使用 LangGraph 和 Redis 的示例

from redisvl.query import FilterQuery

def consolidate_memories(user_id: str, batch_size: int = 10):
    """
    Periodically merge similar long-term memories for a user.
    """
    logger.info(f"Starting memory consolidation for user {user_id}")
    
    # For each memory type, consolidate separately

    for memory_type in MemoryType:
        all_memories = []

        # Get all memories of this type for the user
        of_type_for_user = (Tag("user_id") == user_id) & (
            Tag("memory_type") == memory_type
        )
        filter_query = FilterQuery(filter_expression=of_type_for_user)
        
        for batch in long_term_memory_index.paginate(filter_query, page_size=batch_size):
            all_memories.extend(batch)
            
        all_memories = long_term_memory_index.query(filter_query)
        if not all_memories:
            continue

        # Group similar memories
        processed_ids = set()
        for memory in all_memories:
            if memory["id"] in processed_ids:
                continue

            memory_embedding = memory["embedding"]
            vector_query = VectorRangeQuery(
                vector=memory_embedding,
                num_results=10,
                vector_field_name="embedding",
                filter_expression=of_type_for_user
                & (Tag("memory_id") != memory["memory_id"]),
                distance_threshold=0.1,
                return_fields=[
                    "content",
                    "metadata",
                ],
            )
            similar_memories = long_term_memory_index.query(vector_query)

            # If we found similar memories, consolidate them
            if similar_memories:
                combined_content = memory["content"]
                combined_metadata = memory["metadata"]

                if combined_metadata:
                    try:
                        combined_metadata = json.loads(combined_metadata)
                    except Exception as e:
                        logger.error(f"Error parsing metadata: {e}")
                        combined_metadata = {}

                for similar in similar_memories:
                    # Merge the content of similar memories
                    combined_content += f" {similar['content']}"

                    if similar["metadata"]:
                        try:
                            similar_metadata = json.loads(similar["metadata"])
                        except Exception as e:
                            logger.error(f"Error parsing metadata: {e}")
                        similar_metadata = {}

                    combined_metadata = {**combined_metadata, **similar_metadata}

                # Create a consolidated memory
                new_metadata = {
                    "consolidated": True,
                    "source_count": len(similar_memories) + 1,
                    **combined_metadata,
                }
                consolidated_memory = {
                    "content": summarize_memories(combined_content, memory_type),
                    "memory_type": memory_type.value,
                    "metadata": json.dumps(new_metadata),
                    "user_id": user_id,
                }

                # Delete the old memories
                delete_memory(memory["id"])
                for similar in similar_memories:
                    delete_memory(similar["id"])

                # Store the new consolidated memory
                store_memory(
                    content=consolidated_memory["content"],
                    memory_type=memory_type,
                    user_id=user_id,
                    metadata=consolidated_memory["metadata"],
                )

                logger.info(
                    f"Consolidated {len(similar_memories) + 1} memories into one"
                )

def delete_memory(memory_id: str):
    """Delete a memory from Redis"""
    try:
        result = long_term_memory_index.drop_keys([memory_id])
    except Exception as e:
        logger.error(f"Deleting memory {memory_id} failed: {e}")
    if result == 0:
        logger.debug(f"Deleting memory {memory_id} failed: memory not found")
    else:
        logger.info(f"Deleted memory {memory_id}")

def summarize_memories(combined_content: str, memory_type: MemoryType) -> str:
    """Use the LLM to create a concise summary of similar memories"""
    try:
        system_prompt = f"""
        You are a memory consolidation assistant. Your task is to create a single, 
        concise memory from these similar memory fragments. The new memory should
        be a {memory_type.value} memory.
        
        Combine the information without repetition while preserving all important details.
        """

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(
                content=f"Consolidate these similar memories into one:\n\n{combined_content}"
            ),
        ]

        response = summarizer.invoke(messages)
        return str(response.content)
    except Exception as e:
        logger.error(f"Error summarizing memories: {e}")
        # Fall back to just using the combined content
        return combined_content


def memory_consolidation_worker(user_id: str):
    """
    Worker that periodically consolidates memories for the active user.

    NOTE: In production, this would probably use a background task framework, such
          as rq or Celery, and run on a schedule.
    """
    while True:
        try:
            consolidate_memories(user_id)
            # Run every 10 minutes
            time.sleep(10 * 60)
        except Exception as e:
            logger.exception(f"Error in memory consolidation worker: {e}")
            # If there's an error, wait an hour and try again
            time.sleep(60 * 60)