Knowledge Graph Memory Server

Instructions

Implementation Reference

• optimized_memory_mcp_server/storage/sqlite/operations/entity_ops.py:110-138 (handler)

  The actual database operation implementation for deleting observations from an entity.

  async def delete_observations(
      self, 
      deletions: List[Dict[str, Any]], 
      batch_size: int = 1000
  ) -> None:
      """Remove specific observations from entities using batch processing."""
      async with self.pool.get_connection() as conn:
          async with self.pool.transaction(conn):
              for i in range(0, len(deletions), batch_size):
                  batch = deletions[i:i + batch_size]
                  
                  for deletion in batch:
                      entity_name = sanitize_input(deletion["entityName"])
                      to_delete = set(deletion["observations"])
  
                      cursor = await conn.execute(
                          "SELECT observations FROM entities WHERE name = ?",
                          (entity_name,)
                      )
                      result = await cursor.fetchone()
                      if result:
                          current_obs = result['observations'].split(',') if result['observations'] else []
                          updated_obs = [obs for obs in current_obs if obs not in to_delete]
                          
                          await conn.execute(
                              "UPDATE entities SET observations = ? WHERE name = ?",
                              (','.join(updated_obs), entity_name)
                          )